app.storage.postgres.events

Decision-event and intake-run queries of the PostgreSQL store.

Decision uniqueness per (identity, digest) is enforced by uq_decision_event_identity_digest; inserts use ON CONFLICT DO NOTHING so a conflict (including a concurrent one) is reported as a domain error without aborting the surrounding transaction.

Public API: append_decision(), list_decisions(), list_decisions_for(), start_intake_run(), finish_intake_run(), find_intake_run().

app.storage.postgres.events.append_decision(session, decision)

Append a decision event.

created_at is the server’s clock_timestamp() so events keep their real order even within one transaction.

Parameters:

  • session (Session) – Unit-of-work session.

  • decision (NewDecision) – The decision.

Return type:

DecisionEvent

Returns:

The stored event.

Raises:

DecisionExistsError – If (identity, digest) is decided.

app.storage.postgres.events.list_decisions(session, identity)

List all decision events of an identity.

Parameters:

  • session (Session) – Unit-of-work session.

  • identity (Identity) – Logical identity.

Return type:

list[DecisionEvent]

Returns:

Events, oldest first (ties by id).

app.storage.postgres.events.list_decisions_for(session, identities)

List the decision events of several identities in one query.

Parameters:

  • session (Session) – Unit-of-work session.

  • identities (Collection[Identity]) – Logical identities; empty means no query.

Return type:

list[DecisionEvent]

Returns:

Events of all of them, oldest first (ties by id).

app.storage.postgres.events.start_intake_run(session, source_id, kind, identity, delivery_id)

Record the start of an intake run.

started_at uses clock_timestamp() so runs inside a long unit of work get real times.

Parameters:

  • session (Session) – Unit-of-work session.

  • source_id (str) – Registration id as requested.

  • kind (IntakeKind) – Poll or push.

  • identity (JsonValue) – Who triggered the run.

  • delivery_id (str | None) – Push delivery id, if any.

Return type:

IntakeRun

Returns:

The running run.

Raises:

DeliveryExistsError – If the delivery id was used before.

app.storage.postgres.events.finish_intake_run(session, run_id, status, counts, errors)

Record the end of an intake run.

Parameters:

  • session (Session) – Unit-of-work session.

  • run_id (UUID) – Run id.

  • status (IntakeStatus) – Final status.

  • counts (JsonValue) – Outcome counters.

  • errors (JsonValue) – Error details.

Return type:

IntakeRun

Returns:

The finished run.

Raises:

LookupError – If the run does not exist.

app.storage.postgres.events.find_intake_run(session, source_id, delivery_id)

Find the run of an earlier push delivery.

Parameters:

  • session (Session) – Unit-of-work session.

  • source_id (str) – Registration id.

  • delivery_id (str) – Push delivery id.

Return type:

IntakeRun | None

Returns:

The run, or None.