@callback activate_projection(non_neg_integer()) :: :ok | {:error, term()}

Makes generation the one load_event_total/3 reads.

@callback assign_legacy_plan_versions(pos_integer()) ::
  {:ok, %{assigned: non_neg_integer(), orphans: non_neg_integer()}}
  | {:error, term()}
  | {:error, {:unsupported, capability()}}

Names the contract of at most limit subscriptions that have no plan_version, and reports what it did.

One bounded transaction. Rows are taken with FOR UPDATE SKIP LOCKED, so two nodes running AuroraMeter.Plans.register!/0 at the same time take disjoint batches rather than queueing. assigned is how many rows this batch named and orphans how many of those carry a plan_id with no registered version, which get a version and a NULL fingerprint: retiring a plan id from code is ordinary and refusing the upgrade for every customer who still holds one would make the upgrade unrunnable for exactly the installs with the longest histories.

assigned: 0 means there is nothing left to do and the caller stops.

Not one of the two callbacks api-change-map.md 1.2 lists. It is here rather than as raw SQL inside AuroraMeter.Plans so that the whole of this feature sits behind one capability, and an adapter that cannot store snapshots is not asked to run a statement against a table it does not have (finding X286).

@callback begin_projection_generation() ::
  {:ok, %{optional(atom()) => term()}} | {:error, term()}

Announces a new building generation and returns the state it announced.

One short transaction that takes the projection row FOR UPDATE, so it is granted only once every record transaction holding FOR SHARE on that row has committed or rolled back. In that transaction it reads the watermark max(seq), copies the active generation's rows into both the new building generation and a frozen seed generation, and publishes the building generation so every later record writes both.

Idempotent: when a building generation already exists it is returned with resumed: true and nothing is copied.

@callback capabilities() :: [capability()]

The durable operations this adapter supports.

An adapter that returns [] still defines every callback below; the dispatchers refuse the call on its behalf.

@callback drain_projection_seed(integer(), pos_integer()) ::
  {:ok, non_neg_integer()} | {:error, term()}

Subtracts up to limit seed rows from the generation they seeded and deletes them, in one transaction, returning how many were consumed.

The seed row's own existence is the cursor: draining is finished when none is left, so an interrupted drain needs no checkpoint and can never subtract a row twice.

@callback list_plan_versions(String.t() | :all) ::
  [AuroraMeter.Schema.PlanVersion.t()] | {:error, {:unsupported, capability()}}

Every stored snapshot for plan_id, or for every plan when given :all.

Ordered by (plan_id, version) so two nodes comparing fingerprints compare them in the same order and report the same list.

@callback list_subscriptions(cursor :: String.t() | nil, opts :: keyword()) ::
  {[AuroraMeter.Schema.Subscription.t()], next_cursor :: String.t() | nil}

One bounded page of subscriptions, in keyset order.

cursor is the opaque value the previous page ended on, or nil for the first page; the page returned is strictly after it. The second element of the return is the cursor to pass next, or nil when that page was the last one.

Options:

• :limit (default 100);
• :status_in: a list of status strings; omitted means every status;
• :order: :tenant_key (default) or :scheduled_effective_at, which walks pending plan transitions in the order they come due;
• :transition_state, :transition_confirm: exact matches on the transition mirror columns, which is how Aurora Meter Pro finds the provider-confirmed work it owns without querying a core table directly;
• :scheduled_before: a DateTime, inclusive.

Keyset, not offset, because the caller is a worker that walks every subscription while other processes insert and delete them. An offset page silently skips a row when an earlier one is removed under it, and a skipped subscription is usage nobody bills.

@callback load_event(String.t(), String.t()) ::
  {:ok, AuroraMeter.Event.t()}
  | {:error, :not_found | {:unsupported, capability()}}

Reads one recorded event by its caller identity.

@callback load_event_total(String.t(), atom() | String.t(), DateTime.t()) ::
  {:ok, %{quantity: non_neg_integer(), events: non_neg_integer()}}
  | {:error, {:unsupported, capability()}}

Reads the projected total for one feature and period, in the active generation.

@callback projection_state() :: {:ok, projection_state()} | {:error, term()}

Reads the projection state without locking anything.

@callback put_plan_version(map()) ::
  {:ok, AuroraMeter.Schema.PlanVersion.t()}
  | {:error, term()}
  | {:error, {:unsupported, capability()}}

Inserts one plan version snapshot, or does nothing if (plan_id, version) is already stored.

Never an update. A snapshot is the commercial content a customer was sold and it is immutable; a differing compiled definition is AuroraMeter.PlanVersionConflictError, not a write.

@callback record_correction(
  correction_entry(),
  keyword()
) ::
  {:ok, [record_outcome()]}
  | {:error, {:conflict, non_neg_integer(), AuroraMeter.Event.t()}}
  | {:error, {:invalid, [{atom(), atom()}]}}
  | {:error, {:not_found, :original}}
  | {:error, term()}

Records one correction, and optionally its replacement, in one transaction.

A correction is a new immutable row reducing the effective quantity of an existing event in the same tenant; no historical row is ever updated. The cumulative magnitude of the corrections of one original may never exceed that original's quantity (I09), which an adapter enforces by serialising every corrector of that original against one another. The Ecto adapter does it with SELECT ... FOR UPDATE on the original row.

The order of the two checks is load bearing and an adapter must keep it: the duplicate check comes before the bound check. A retry of a correction has its own committed row inside the cumulative sum, so evaluating the bound first makes every retry of a correct correction fail with exceeds_original, which is the difference between an idempotent financial operation and one that cannot be retried at all.

Results are [{correction, outcome}], or [{correction, outcome}, {replacement, outcome}] when opts[:replacement] carries a usage entry to insert in the same transaction (AuroraMeter.replace/4).

Refusals return {:error, {:invalid, _}} or {:error, {:not_found, :original}} without rolling back, so a host transaction that wrapped the call keeps its own writes. Conflicts roll back, because by then a write has been attempted and the caller's intent is unsatisfiable.

Options: :timeout, :outbox and :replacement.

@callback record_events(
  [event_entry()],
  keyword()
) ::
  {:ok, [record_outcome()]}
  | {:error, {:conflict, non_neg_integer(), AuroraMeter.Event.t()}}
  | {:error, term()}

Records a batch of events in one transaction, with their projection deltas and the configured outbox's intent.

The three are one commit or none of them (L-03b-1). Results are returned in the caller's input order. An entry whose (tenant_key, event_id) already exists with an equal payload_hash is :duplicate and contributes no totals delta and no outbox item; one with a different hash rolls the whole batch back with {:error, {:conflict, index, existing}}.

Options: :timeout (milliseconds for the transaction and every statement in it) and :outbox (the AuroraMeter.Events.Outbox module to invoke).

@callback stream_events(
  non_neg_integer(),
  keyword()
) :: {:ok, [AuroraMeter.Event.t()]} | {:error, {:unsupported, capability()}}

Reads one bounded page of events after cursor, ordered by seq.

seq and not id: event ids are random v4 UUIDs, so a keyset scan ordered by id can silently miss a row committed by a transaction that started earlier (open-findings.md L20).

Options: :limit, :tenant, :feature, :from and :to (on occurred_at, half-open).

@callback write_projection_totals(non_neg_integer(), [projection_total()]) ::
  :ok | {:error, term()}

Adds projection deltas to generation, creating rows that are absent.

Adds rather than sets, because a record that commits while a generation is being built writes its own delta there too, and a replay's batch must not overwrite it. Entries are applied in one statement, sorted by {tenant_key, feature, period_start} so two writers cannot deadlock.

Call it inside the caller's transaction: a replay commits the totals and its checkpoint together, which is the whole of its resumability.