Factos PostgreSQL

factos_pog stores Factos events in PostgreSQL through pog. Dispatches run in serializable transactions and retry serialization or deadlock conflicts.

Setup

Apply priv/migrations.sql, or copy the SQL into your migration history. The schema contains:

• factos.factos_events, the globally ordered event log;
• factos.factos_event_tags, the selective context index.

Runtime code should not execute migrations.

Configure and dispatch

let model = user.model()

let configuration = factos_pog.configure(model, connection: connection)

configuration
|> factos_pog.dispatch(
  command,
  decision_context: user.decision_context(command),
  event_id: new_event_id,
)

A successful dispatch returns append-ordered factos.Recorded(event) values and the final global position.

Transactional subscriptions

Subscriptions run after the event insert and before commit. They receive the same Pog transaction connection, in subscription order and event order.

let projection =
  factos.subscription(fn(connection, recorded) {
    use _ <- result.try(user_projection.apply(connection, recorded))
    Ok(connection)
  })

let configuration =
  factos_pog.Configuration(
    ..factos_pog.configure(model, connection: connection),
    subscriptions: [projection],
  )

A subscription error rolls back its writes, earlier subscription writes, and the event append.

Simulate projections

Use factos/factos_pog/simulate to run the same subscriptions from an in-memory domain scenario against a real PostgreSQL projection database:

import factos/factos_pog/simulate as simulate_pog

simulate_pog.new(
  model,
  connection:,
  subscriptions: [projection],
)
|> simulate.given(existing_events)
|> simulate.dispatch(decision_context:, command:)
|> simulate.assert_errors([])

assert Ok(rows) = user_projection.load(connection)

Each accepted event batch runs in its own serializable transaction. Subscription or transaction failure rolls back projection writes and leaves the batch out of simulated history. The simulator does not write the PostgreSQL event log; event-store atomicity and concurrency remain dispatch integration concerns.

Durable work with M25

m25 is a natural companion when an accepted event must create durable background work. Enqueue the M25 job from a Factos subscription using the supplied transaction connection:

let durable_job =
  factos.subscription(fn(connection, recorded) {
    use _ <- result.try(
      recorded
      |> job_for_event
      |> m25.new_job
      |> m25.enqueue(connection, queue, _),
    )
    Ok(connection)
  })

The event and job commit together. M25 then provides supervised, at-least-once execution and retries after commit. External IO does not run inside the Factos transaction.

Recovery reads

read_after returns a bounded, globally ordered page after a sequence position. Use it for projection rebuilds and application-owned recovery workers.

Development

From the repository root:

docker compose up --wait -d
trellis run test factos_pog

See How factos_pog works for transaction and schema details. The stress benchmark lives in benchmark/.