persistence
API
Generated with
persistence/com.worxbend.persistence/com.worxbend.persistence.repository

com.worxbend.persistence.repository

package com.worxbend.persistence.repository

Members list

Type members

Classlikes

final case class Checkpoint(groupId: String, topic: String, partition: Int, nextOffset: Long, records: Long, owner: Option[String], updatedAt: OffsetDateTime)

One partition's durable position.

One partition's durable position.

Value parameters

nextOffset

the offset of the next record to read, which is Kafka's own commit convention (last processed + 1). Storing "last processed" instead reads identically and is off by one at every seek, in the direction that reprocesses a record — safe here, because the insert is idempotent, and still wrong.

records

how many records this checkpoint accounts for, cumulative. The cheapest way to tell an idle partition from a stuck one: both have a static nextOffset, only one has a static count. Accounted for, not written: the writer counts every record whose offset this position covers, dead letters included, because that is the set nextOffset is derived from and the two must describe the same batch. It is also only a diagnostic — a batch that has to be bisected writes each half separately and the halves overlap, so treat the count as a floor on activity rather than as an exact tally.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class CheckpointCommit(groupId: String, owner: Option[String], positions: Vector[CheckpointWrite])

What to checkpoint alongside a batch of events.

What to checkpoint alongside a batch of events.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
trait CheckpointStore

Externalised consumer offsets.

Externalised consumer offsets.

This does not replace Kafka's __consumer_offsets, which stays authoritative for fetching. It answers the two questions Kafka's own storage cannot:

  1. Retention. offsets.retention.minutes defaults to seven days. A group that stops committing for longer has its offsets deleted, and the next start resolves auto.offset.reset=earliest and replays the whole retained log. Survivable, because the insert deduplicates — and completely silent, which is the problem.
  2. Transactionality. record is called inside the same transaction as the batch insert, so the offset and the rows it accounts for commit or roll back together. That is the only arrangement in which "the last durably stored offset" has an exact answer; Kafka's commit is a separate round trip afterwards, and the window between them is precisely the replay window.

Why Postgres and not Redis or Cassandra. The store has to be external to the broker, and the honest answer for this system is the database that is already a hard dependency of the write path. Redis is a second datastore to run, back up and monitor, and — not being transactional with respect to the insert — gives up property 2, which is the only reason worth doing this at all. Cassandra adds an operational surface larger than the rest of the stack for a table holding one row per partition. If Postgres is down, cobalt is not consuming; no new failure mode.

Attributes

Companion
object
Supertypes
class Object
trait Matchable
class Any
Known subtypes
object CheckpointStore

Attributes

Companion
trait
Supertypes
class Object
trait Matchable
class Any
Self type
final case class CheckpointWrite(topic: String, partition: Int, nextOffset: Long, records: Long)

A position to write, without the bookkeeping the store fills in.

A position to write, without the bookkeeping the store fills in.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
trait CheckpointingWriter

Insert-and-checkpoint, in one transaction.

Insert-and-checkpoint, in one transaction.

A separate interface from EventRepository, deliberately. Only the consumer has offsets; ferrite is a reader and its repository stub should not have to implement — or be able to forget — a method it has no use for. Adding a defaulted method to EventRepository instead would put a no-op checkpoint on every implementation, which is the failure this arrangement makes unrepresentable: a writer that silently discards the offset it was handed.

Attributes

Supertypes
class Object
trait Matchable
class Any
Known subtypes
final case class EventDetail(summary: EventSummary, raw: Json)

A single event with its payload.

A single event with its payload.

Composed of a EventSummary plus raw rather than being a flat 14-field record, so the list projection and the detail projection cannot drift: the detail query is literally the list query plus one column.

Attributes

Companion
object
Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
object EventDetail

Attributes

Companion
class
Supertypes
trait Product
trait Mirror
class Object
trait Matchable
class Any
Self type
final case class EventRef(occurredAt: OffsetDateTime, eventUid: UUID)

The primary key of an event.

The primary key of an event.

occurred_at is part of the identity, not a convenience: the table is partitioned on it, so a lookup by event_uid alone would have to scan every partition. Handing callers a two-field reference makes that structural instead of something a repository has to remember.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
trait EventRepository

The database surface the two consuming services need, and nothing else.

The database surface the two consuming services need, and nothing else.

Split by direction on purpose: ferrite calls only the read methods, cobalt only insertAll. Keeping them in one trait rather than two is a bet that the split is a fact about callers and not about the schema — both halves talk to the same table and any divergence between how a row is written and how it is read is a bug this interface should make obvious rather than hide behind two files.

No Magnum type appears in any signature (ADR §5). Callers see domain values and Future, so the ADR's budgeted Magnum 2.0 migration rewrites implementations and touches nothing above them.

Every method returns Future because the implementation is blocking JDBC on a bounded dispatcher (ADR §0, decision 8): the pool, not the caller, is where load is shed.

Attributes

Supertypes
class Object
trait Matchable
class Any
Known subtypes
final case class EventSummary(occurredAt: OffsetDateTime, eventUid: UUID, ingestedAt: OffsetDateTime, ceId: String, ceSource: String, ceType: String, ceSubject: Option[String], deviceId: Option[String], roomId: Option[String], personId: Option[String], severity: Option[String], severityRank: Option[Short], metricValue: Option[Double])

A row of the result list.

A row of the result list.

raw and data are deliberately absent. ADR §6.3: including them would make the planner de-TOAST every payload on the page, including the rows a user scrolls past without opening. The detail view fetches the payload by key, which is one extra round trip for the one event that was actually asked for.

Field order is the column order of the SELECT in PostgresEventRepository. Magnum's derived codec reads positionally, so the two are a matched pair — reordering either alone compiles and then silently returns a device id in the room field.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
object FacetDimension

Attributes

Companion
enum
Supertypes
trait Sum
trait Mirror
class Object
trait Matchable
class Any
Self type
enum FacetDimension(val groupingMask: Int, val label: String)

The dimensions the facet panel offers.

The dimensions the facet panel offers.

groupingMask is the value PostgreSQL's GROUPING() returns for the row belonging to this dimension: a bit is set for every argument not in that grouping set. Encoding it here rather than emitting one query per dimension is the whole point of ADR §6.3's single-pass facet query — six queries over the same candidate set is six scans.

The masks assume the argument order (ce_type, ce_source, device_id, room_id, person_id, severity) used in PostgresEventRepository. Changing that order without changing these numbers silently relabels every facet.

Attributes

Companion
object
Supertypes
trait Enum
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class FacetRequest

A facet query, validated.

A facet query, validated.

Attributes

Companion
object
Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
object FacetRequest

Attributes

Companion
class
Supertypes
trait Product
trait Mirror
class Object
trait Matchable
class Any
Self type
final case class FacetValue(value: String, count: Long)

One facet entry: a dimension value and how many candidate rows carry it.

One facet entry: a dimension value and how many candidate rows carry it.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class Facets(dimensions: Map[FacetDimension, Vector[FacetValue]], tags: Vector[FacetValue], candidates: Long, capped: Boolean)

Facet counts, plus the honesty flag.

Facet counts, plus the honesty flag.

Value parameters

candidates

how many rows the capped candidate set actually held.

capped

true when the cap was reached, in which case every count is a lower bound and the UI must render "50,000+" rather than a number. ADR §6.3 makes this a signed-off product decision, not an implementation detail — a count that is silently approximate is a count that will be quoted in a meeting.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class HistogramBucket(bucket: OffsetDateTime, count: Long)

One bucket of the time histogram. count is bigint, so Long: a month of a busy smart home overflows Int sooner than anyone expects, and the failure is a negative bar on a chart.

One bucket of the time histogram. count is bigint, so Long: a month of a busy smart home overflows Int sooner than anyone expects, and the failure is a negative bar on a chart.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class HistogramRequest

A histogram query over a half-open window, with a server-chosen bucket width.

A histogram query over a half-open window, with a server-chosen bucket width.

No instance holds more buckets than HistogramRequest.MaxBuckets — (until - from) / width is checked, not hoped for. That is an invariant of this type and not a consequence of whatever bounds the filter grammar happens to impose: every bucket becomes a generate_series row, a presenter Bar carrying a re-rendered query string and an `

  • ` in the response, so the cap is the only thing standing between one GET and a page measured in hundreds of megabytes.

Attributes

Companion
object
Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
object HistogramRequest

Attributes

Companion
class
Supertypes
trait Product
trait Mirror
class Object
trait Matchable
class Any
Self type
object NewEvent

Attributes

Companion
class
Supertypes
trait Product
trait Mirror
class Object
trait Matchable
class Any
Self type
NewEvent.type
final case class NewEvent

A row to be written.

A row to be written.

occurredAt is passed separately from raw even though it is in raw, because it is the partition key and ADR §5 forbids generating a partition key column (text::timestamptz is only STABLE). The ingestion layer is what validated and clamped it; the database takes it on trust and files the row accordingly.

Value parameters

canonical

raw.noSpaces, rendered once. It is a field and not a method because it is needed twice per record on the consume path — to compute payloadSha256 and to bind the jsonb parameter — and rendering a CloudEvent is the most expensive pure operation in that path. It also makes the two uses provably the same bytes: the digest is over the string that is sent, not over a second rendering that merely ought to be equal to it. The constructor is private so those three fields cannot be assembled inconsistently. There is one way to build a NewEvent and it derives all three from the document.

Attributes

Companion
object
Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
trait OverviewRepository

The reader events.event_rollup_hourly never had.

The reader events.event_rollup_hourly never had.

Why this is a separate interface from EventRepository. They answer different questions from different relations with different cost models. Everything here reads the materialized view — one row per (hour, type, source, severity), at most a few thousand rows for a 30-day window — so a query that would scan millions of fact rows becomes an index range scan over a small aggregate. Everything in EventRepository reads the fact table. Keeping them apart makes "this page must not touch the fact table" a property of a type rather than a comment on a method.

What the caller is buying, and what it costs. The view is refreshed on a schedule (cobalt's RollupRefresh, every few minutes), so every number from here is stale by up to one refresh interval and the current hour is always partial. That is the trade: a 90-day chart that costs nothing, in exchange for numbers that are not to-the-second. A UI built on this must say so — see freshness, which exists so the page can state the staleness rather than leave an operator to infer it from a count that disagrees with search.

Attributes

Supertypes
class Object
trait Matchable
class Any
Known subtypes
object OverviewRequest

Attributes

Companion
class
Supertypes
trait Product
trait Mirror
class Object
trait Matchable
class Any
Self type
final case class OverviewRequest

An overview query, validated.

An overview query, validated.

The constructor is private so from is always aligned to step. An unaligned origin puts the generate_series skeleton and date_bin's output on different boundaries, the LEFT JOIN then matches nothing, and the chart renders as a row of zeroes over a database full of events — a failure that looks like "no data" rather than like a bug.

Value parameters

topN

entries per breakdown. Bounded because each breakdown is ORDER BY count DESC LIMIT ? over the whole window and an unbounded list is both a slow sort and a page nobody reads the end of.

Attributes

Companion
object
Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final class PostgresCheckpointStore(read: Transactor, write: Transactor)(using x$3: ExecutionContext) extends CheckpointStore

The Magnum implementation.

The Magnum implementation.

Two transactors, as everywhere else in this module: reads go to the read pool, writes to the write pool. record takes neither, because it deliberately borrows the caller's transaction.

Attributes

Supertypes
class Object
trait Matchable
class Any
final class PostgresEventRepository(read: Transactor, write: Transactor)(using ec: ExecutionContext) extends EventRepository, CheckpointingWriter

The PostgreSQL implementation.

The PostgreSQL implementation.

Two transactors, not one: the read side is bound to the search pool (read-only, statement_timeout = 2s) and the write side to the ingest pool, so ADR §5's isolation between a runaway search and durable persistence is expressed in the constructor rather than trusted to convention.

The ExecutionContext is expected to be the bounded CustomExecutionContext sized to the pool (ADR §0, decision 8). Handing this class the global pool would work, and would relocate the queue from somewhere with a timeout to somewhere without one.

Every statement is built by a pure function in the companion, not inline here. That is what makes the SQL testable without a database — a missing FROM clause or a parameter bound out of order is a defect a unit test can find, and it is exactly the kind of defect that otherwise waits for an integration run on a Docker host.

Attributes

Companion
object
Supertypes
class Object
trait Matchable
class Any
object PostgresEventRepository

The statement builders.

The statement builders.

Deliberately pure, package-visible and separate from the class that runs them. A Frag is a value: it can be asserted on for placeholder count, parameter order and the presence of a FROM clause without a connection, which is the only kind of test that runs on the fast CI tier.

Attributes

Companion
class
Supertypes
class Object
trait Matchable
class Any
Self type
object PostgresOverviewRepository

Attributes

Companion
class
Supertypes
class Object
trait Matchable
class Any
Self type
final class PostgresOverviewRepository(read: Transactor)(using ec: ExecutionContext) extends OverviewRepository

The PostgreSQL implementation, bound to the read pool.

The PostgreSQL implementation, bound to the read pool.

One transactor, not two: nothing here writes, and there is no insert on the interface to make the question arise.

Every statement is built by a pure function in the companion, for the same reason as PostgresEventRepository: a Frag is a value, so parameter order and placeholder count are assertable without a database, and only the behaviour of the view needs the slow tier.

Attributes

Companion
object
Supertypes
class Object
trait Matchable
class Any
enum RollupDimension(val label: String, val filterKey: String)

The dimensions events.event_rollup_hourly can be sliced by.

The dimensions events.event_rollup_hourly can be sliced by.

Three, because three is what the view groups on. device_id is not a column of the rollup and cannot be added to this enum without changing the view: a device leaderboard is a fact-table aggregate, and ADR §0 decision 7 put the rollup here precisely so a dashboard never does that.

filterKey is the permalink parameter a web tier must use to turn a slice into a search. It lives here rather than in the web tier so that a dimension added to the view arrives with the link it implies, instead of the two being matched up by hand at the call site.

Attributes

Supertypes
trait Enum
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class RollupSlice(value: String, events: Long, errors: Long)

One row of a rollup breakdown: a dimension value with its event and error counts over the window.

One row of a rollup breakdown: a dimension value with its event and error counts over the window.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
enum RollupStep(val width: FiniteDuration)

The bucket width of the volume series.

The bucket width of the volume series.

A closed set of three, not an arbitrary interval. The rollup's own grain is one hour, so an hour is the finest anything can be, and every wider width has to be a whole multiple of it or a bucket would straddle two rollup rows and be double- or half-counted at its edges.

Attributes

Supertypes
trait Enum
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class RollupTotals(events: Long, errors: Long)

The window's totals, from the same relation as everything else on the page.

The window's totals, from the same relation as everything else on the page.

Deliberately does not carry a device count. The rollup's device_count is count(DISTINCT device_id) within each (hour, type, source, severity) group, so summing it counts a device once per group it appears in — a number that is always too large and looks plausible. Answering "how many devices are active" honestly means a count(DISTINCT device_id) over the fact table, which is the scan this view exists to avoid.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class SearchPage(rows: Vector[EventSummary], nextCursor: Option[String])

One page of search results.

One page of search results.

nextCursor is present only when the page was full. A page shorter than the limit is the end of the result set, and offering a cursor for it would produce an empty page and a "load more" control that does nothing.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
final case class SearchRequest

A search, validated.

A search, validated.

The constructor is private so that a limit can never arrive unbounded: this query is served from a pool of eight connections behind a two-second statement timeout, and one request asking for a million rows is a way to spend the whole pool's time budget and then serialise the result into the web tier's heap.

Attributes

Companion
object
Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
object SearchRequest

Attributes

Companion
class
Supertypes
trait Product
trait Mirror
class Object
trait Matchable
class Any
Self type
final case class VolumePoint(bucket: OffsetDateTime, events: Long, errors: Long)

One point of the overview's volume series.

One point of the overview's volume series.

errors is count(*) FILTER (WHERE severity_rank >= 50) as the rollup already computed it, not a second query: the two numbers have to describe the same rows or the "3 % of traffic is errors" reading on the page is a comparison between two different populations.

Attributes

Supertypes
trait Serializable
trait Product
trait Equals
class Object
trait Matchable
class Any
Show all
Generated with