Durable Subscriptions

Durable subscriptions define which post-commit effects a bounded context produces from newly appended events. Read How factos_pog works first for the transaction and storage overview.

Configure the topology once

Four public types define the production topology:

• A pure factos.Reactor(event, effect) derives zero or more typed effects from each recorded event.
• EffectCodec(effect) deterministically turns each typed effect into a non-generic opaque ProposedEffect durable envelope.
• Subscription(event) binds a stable name to one reactor and its effect codec.
• Dispatcher(event) binds the event codec to the bounded context’s complete, immutable subscription list.

Construct and validate the dispatcher once in the composition root, then inject it wherever commands are dispatched. Construction rejects blank subscription names and the first exact duplicate before database IO. Individual command paths cannot add or remove subscriptions. Every configured subscription therefore observes every new event; a reactor that does not apply to an event returns [].

Define an effect codec

Keep the application-side effect typed. Its codec prepares the stable durable representation without executing external work:

type Effect {
  SendWelcome(username: String)
}

fn encode(effect: Effect) -> factos_pog.ProposedEffect {
  case effect {
    SendWelcome(username:) -> {
      let metadata = factos.metadata([
        #(factos.correlation_id, "corr-" <> username),
      ])
      let payload = bit_array.from_string(username)

      factos_pog.proposed_effect(
        consumer: "wallets",
        key: "welcome:" <> username,
        target: "ledgers",
        type_: "SendWelcome",
        metadata:,
        payload:,
      )
    }
  }
}

fn welcome_effect_codec() -> factos_pog.EffectCodec(Effect) {
  factos_pog.effect_codec(encode:)
}

The application-owned typed effect exists only as the codec input. proposed_effect intentionally crosses the representation boundary with the consumer, key, target, type, metadata, and opaque payload that Factos Pog prepares for persistence.

Effect codecs must be deterministic. Do not publish messages, call services, mutate state, or allocate externally visible identifiers in an encoder.

Define a named subscription

The reactor remains a pure event-to-effects function. Bind it to its codec under a stable, versioned name:

fn welcome_reactor() -> factos.Reactor(Event, Effect) {
  factos.reactor(react: fn(recorded) {
    case recorded.event {
      UserRegistered(username:) -> [SendWelcome(username:)]
    }
  })
}

fn welcome_subscription() -> factos_pog.Subscription(Event) {
  factos_pog.subscription(
    name: "welcome.v1",
    reactor: welcome_reactor(),
    codec: welcome_effect_codec(),
  )
}

The configured name is durable identity. Dispatcher construction requires every name to be non-blank and exactly unique, while preserving valid names without normalization. Factos Pog persists it in OutboxMessage.subscription and prefixes the application key, producing welcome.v1:welcome:<username>. Outbox uniqueness remains (consumer, effect_key), so this namespace prevents different subscriptions from colliding. Use stable, versioned names such as welcome.v1; do not rename a deployed subscription casually.

Build and inject the dispatcher

Pair the already-defined domain event codec with the complete subscription list, then pass that dispatcher to every command dispatch:

let assert Ok(dispatcher) =
  factos_pog.dispatcher(
    codec: event_codec(),
    subscriptions: [welcome_subscription()],
  )

let assert Ok(dispatch) =
  factos_pog.new_dispatch(
    dispatcher:,
    connection:,
    stream: "user-renata",
    decider: decider(),
  )
  |> factos_pog.dispatch(
    RegisterUser(username: "renata"),
    event_id: uuid.v4_string,
  )

event_codec() and decider() remain ordinary domain functions. Command builders choose command consistency and dispatch inputs, not which reactions are active.

Dispatch atomically

For every event appended by dispatch, Factos Pog runs every configured subscription inside the same PostgreSQL SERIALIZABLE transaction. Irrelevant reactors return []; relevant reactors encode their effects, and those rows are inserted atomically with their source events. External IO starts only after the transaction commits and a worker leases the durable effect.

A serialization or deadlock conflict can rerun the transaction. Deciders, reactors, event codecs, effect codecs, and event-id generators must therefore be pure or idempotent.

Multiple effect types

A dispatcher can hold subscriptions with different effect types and codecs while all of them observe the same event type. For example, registration can also produce a separate audit effect:

type AuditEffect {
  RecordRegistration(username: String)
}

fn audit_subscription() -> factos_pog.Subscription(Event) {
  factos_pog.subscription(
    name: "registration-audit.v1",
    reactor: factos.reactor(react: fn(recorded) {
      case recorded.event {
        UserRegistered(username:) -> [RecordRegistration(username:)]
      }
    }),
    codec: factos_pog.effect_codec(encode: fn(effect) {
      let RecordRegistration(username:) = effect
      factos_pog.proposed_effect(
        consumer: "audit",
        key: "registration:" <> username,
        target: "warehouse",
        type_: "RecordRegistration",
        metadata: factos.empty_metadata(),
        payload: bit_array.from_string(username),
      )
    }),
  )
}

let assert Ok(dispatcher) =
  factos_pog.dispatcher(
    codec: event_codec(),
    subscriptions: [welcome_subscription(), audit_subscription()],
  )

The opaque Subscription(Event) erases each subscription’s effect type only after pairing its typed reactor and codec. The dispatcher can consequently store both Effect and AuditEffect subscriptions without weakening either codec.

Backfill one subscription

Use the explicitly named backfill API to create effects for matching historical events:

backfill_subscription(
  dispatcher:,
  connection:,
  name:,
  query:,
) -> Result(Nil, factos_pog.Error(Nil))

For example, backfill only welcome effects for one user:

let query =
  factos.query([
    factos.query_item(
      types: [factos.event_type("UserRegistered")],
      tags: [factos.tag("username:" <> username)],
    ),
  ])

let assert Ok(Nil) =
  factos_pog.backfill_subscription(
    dispatcher: welcome_dispatcher(),
    connection:,
    name: "welcome.v1",
    query:,
  )

// Normal outbox uniqueness makes the same backfill idempotent.
let assert Ok(Nil) =
  factos_pog.backfill_subscription(
    dispatcher: welcome_dispatcher(),
    connection:,
    name: "welcome.v1",
    query:,
  )

backfill_subscription selects only the named subscription, loads matching historical events, decodes them through the dispatcher’s event codec, and reruns only that reactor. Normal outbox uniqueness makes repeated calls idempotent.

An unknown name returns SubscriptionNotFound(name:). A matching event that cannot be loaded, decoded, or stored returns the corresponding normal error. Never emulate an all-subscription backfill: doing so could repeat Ledger, provider, email, or other external effects that were not intended for replay.

Rules to preserve

• Define the complete subscription list once per bounded context.
• Give each subscription a stable, versioned, non-empty name.
• Let irrelevant reactors return []; do not attach reactors conditionally per command.
• Keep deciders, reactors, codecs, and event-id generators pure or idempotent.
• Perform external IO only from workers after commit.
• Backfill exactly one named subscription with an explicit query.

Continue with Durable Effects to lease, execute, settle, inspect, and replay the outbox rows produced here.