factos/factos_pog

PostgreSQL backend for Factos using the pog package.

This backend stores accepted facts in an append-only factos_events table. The event history is the source of truth; projections and stream-shaped reads are derived views over that history.

The context dispatch flow follows the Command Context Consistency idea from “Simply Event Sourcing”: a command selects the facts required for its decision, folds them into temporary state, decides new facts, and appends those facts only when no relevant facts appeared after the observed context position.

The query contract is intentionally tag-based. PostgreSQL stores opaque event bytes, an event type, and tags. This keeps domain serialization outside the backend, but it means any payload value needed for a selective consistency query must be written as a tag.

Types

Append

</>

pub type Append {
  Append(
    current_revision: Int,
    position: factos.SequencePosition,
  )
}

Constructors

```
Append(current_revision: Int, position: factos.SequencePosition)
```
Result of a successful append.

current_revision is the latest revision of the target stream after the append. position is the global position of the last inserted event, or NoPosition when no events were produced.

DeadLetter

</>

A dead-lettered durable effect available for operational inspection.

pub type DeadLetter {
  DeadLetter(
    id: Int,
    subscription: String,
    consumer: String,
    key: String,
    target: String,
    type_: String,
    attempt: Int,
    reason: String,
    created_at: String,
    failed_at: String,
    replay_count: Int,
    last_replayed_at: option.Option(String),
  )
}

Constructors

DeadLetter(
  id: Int,
  subscription: String,
  consumer: String,
  key: String,
  target: String,
  type_: String,
  attempt: Int,
  reason: String,
  created_at: String,
  failed_at: String,
  replay_count: Int,
  last_replayed_at: option.Option(String),
)

DecodeError

</>

pub type DecodeError {
  UnknownEvent
  InvalidData
}

Constructors

```
UnknownEvent
```
```
InvalidData
```

DeliveryConfigurationError

</>

Invalid durable-delivery configuration.

pub type DeliveryConfigurationError {
  InvalidMaximumAttempts
  InvalidInitialDelay
  InvalidMaximumDelay
  InvalidMaximumAge
  InvalidJitterPercent
  InvalidConsumer
  InvalidTarget
}

Constructors

```
InvalidMaximumAttempts
```
```
InvalidInitialDelay
```
```
InvalidMaximumDelay
```
```
InvalidMaximumAge
```
```
InvalidJitterPercent
```
```
InvalidConsumer
```
```
InvalidTarget
```

DeliveryFinalization

</>

Result of atomically applying a configured delivery policy.

pub type DeliveryFinalization {
  DeliveryAcknowledged
  RetryScheduled(attempt: Int, delay_milliseconds: Int)
  DeadLettered(attempt: Int, reason: String)
  StaleLease
}

Constructors

```
DeliveryAcknowledged
```

RetryScheduled(attempt: Int, delay_milliseconds: Int)

DeadLettered(attempt: Int, reason: String)

```
StaleLease
```
The row is no longer pending under the supplied current lease token.

DeliveryOutcome

opaque </>

How an application worker classified one delivery attempt.

A delivery outcome classifies one already-executed leased message. It does not control whether the worker process continues.

pub opaque type DeliveryOutcome

Dispatch

</>

pub type Dispatch(event) {
  Dispatch(append: Append, events: List(factos.Recorded(event)))
}

Constructors

```
Dispatch(append: Append, events: List(factos.Recorded(event)))
```
Result of a successful dispatch.

append has the stream revision and final global position. events are the committed events recorded by this dispatch, suitable for pure Factos reactors or backend-specific durable effect adapters.

DispatchBuilder

opaque </>

pub opaque type DispatchBuilder(command, state, event, domain_error)

Dispatcher

opaque </>

The immutable event codec and delivery topology for one bounded context.

Command handlers must receive a configured dispatcher. They cannot add or remove subscriptions for an individual command.

pub opaque type Dispatcher(event)

DispatcherConfigurationError

</>

Invalid configured subscription topology.

pub type DispatcherConfigurationError {
  InvalidSubscriptionName(name: String)
  DuplicateSubscriptionName(name: String)
}

Constructors

```
InvalidSubscriptionName(name: String)
```
```
DuplicateSubscriptionName(name: String)
```

EffectCodec

opaque </>

pub opaque type EffectCodec(effect)

Error

</>

pub type Error(domain_error) {
  DomainError(domain_error)
  StoreError(pog.QueryError)
  AppendConditionFailed(factos.AppendCondition)
  DecodeError(DecodeError)
  SubscriptionNotFound(name: String)
  SourceEventNotFound(event_id: String)
  InvalidDeliveryReason
}

Constructors

```
DomainError(domain_error)
```
The decider rejected the command with a domain error.
```
StoreError(pog.QueryError)
```
PostgreSQL or pog returned an error while running a query.
```
AppendConditionFailed(factos.AppendCondition)
```
A stream revision or context append condition failed.
```
DecodeError(DecodeError)
```
```
SubscriptionNotFound(name: String)
```
The requested configured subscription does not exist.
```
SourceEventNotFound(event_id: String)
```
An outbox message no longer identifies a stored source event.
```
InvalidDeliveryReason
```
A retry or permanent failure must have a non-empty machine-readable reason.

EventCodec

</>

pub type EventCodec(event) {
  EventCodec(
    encode: fn(event) -> Proposed(event),
    decode: fn(StoredEvent) -> Result(
      factos.Decoded(event),
      DecodeError,
    ),
  )
}

Constructors

```
EventCodec(
  encode: fn(event) -> Proposed(event),
  decode: fn(StoredEvent) -> Result(
    factos.Decoded(event),
    DecodeError,
  ),
)
```
Application-owned PostgreSQL event codec.

encode converts a domain event into bytes and metadata. decode converts a stored row back into a factos.Decoded domain event. Decode failures are kept in the application’s own error type and wrapped as DecodeError.

WARNING: codecs used by dispatch must be pure.

Serializable dispatch can retry the same command after a transaction conflict. That can call encode and decode more than once for the same logical operation, so codec functions must not perform IO, mutate external state, allocate ids from an external system, publish messages, or otherwise create host-system side effects.

Return deterministic stored bytes and metadata from input values only. Put external effects in durable outbox records and execute them after commit.

OutboxConsumer

opaque </>

A durable consumer identity and its single retry policy.

pub opaque type OutboxConsumer

OutboxMessage

</>

pub type OutboxMessage {
  OutboxMessage(
    id: Int,
    source_position: factos.SequencePosition,
    source_event_id: String,
    source_context: String,
    subscription: String,
    consumer: String,
    key: String,
    target: String,
    type_: String,
    metadata: factos.Metadata,
    payload: BitArray,
    attempt: Int,
    lock_token: String,
  )
}

Constructors

OutboxMessage(
  id: Int,
  source_position: factos.SequencePosition,
  source_event_id: String,
  source_context: String,
  subscription: String,
  consumer: String,
  key: String,
  target: String,
  type_: String,
  metadata: factos.Metadata,
  payload: BitArray,
  attempt: Int,
  lock_token: String,
)

A leased outbox message ready for delivery by an application worker.

Arguments

subscription: The configured subscription that produced this durable effect.
attempt: One-based delivery attempt number for this lease.
lock_token: Unguessable identity of this specific lease acquisition.

Proposed

opaque </>

A domain event prepared for PostgreSQL persistence.

Start with new_proposed, then optionally replace its empty tags or metadata through the builder functions.

pub opaque type Proposed(event)

ProposedEffect

opaque </>

pub opaque type ProposedEffect

ReplayAttempts

</>

Whether an administrative replay retains the previous delivery count.

pub type ReplayAttempts {
  PreserveAttempts
  ResetAttempts
}

Constructors

```
PreserveAttempts
```
```
ResetAttempts
```

ReplayResult

</>

Result of attempting to replay a dead letter.

pub type ReplayResult {
  Replayed
  NotDeadLettered
}

Constructors

```
Replayed
```
```
NotDeadLettered
```

RetryPolicy

opaque </>

A validated retry and dead-letter policy.

pub opaque type RetryPolicy

RetryPolicyBuilder

opaque </>

Configurable values used to build a validated retry policy.

pub opaque type RetryPolicyBuilder

StoredEvent

</>

pub type StoredEvent {
  StoredEvent(
    position: Int,
    id: String,
    stream: String,
    revision: Int,
    type_: factos.EventType,
    version: Int,
    tags: List(factos.Tag),
    metadata: factos.Metadata,
    data: BitArray,
  )
}

Constructors

StoredEvent(
  position: Int,
  id: String,
  stream: String,
  revision: Int,
  type_: factos.EventType,
  version: Int,
  tags: List(factos.Tag),
  metadata: factos.Metadata,
  data: BitArray,
)

A raw event row read from PostgreSQL before domain decoding.

Decoders receive this value so they can inspect stored metadata and bytes. position is the global append order. revision is the per-stream revision.

Subscription

opaque </>

A named durable reaction configured once for a bounded context.

The effect type is erased when the subscription is constructed, allowing one dispatcher to contain heterogeneous Ledger, provider, email, and projection subscriptions. The reactor and codec remain pure and may run again after a serializable transaction conflict.

pub opaque type Subscription(event)

Values

backfill_subscription

</>

pub fn backfill_subscription(
  dispatcher configured_dispatcher: Dispatcher(event),
  connection connection: pog.Connection,
  name subscription_name: String,
  query query: factos.Query,
) -> Result(Nil, Error(Nil))

Recreate durable effects for one explicitly named subscription.

Backfill is idempotent because normal outbox uniqueness applies. It is intentionally scoped to one subscription so callers cannot accidentally replay every external integration effect in a bounded context.

build

</>

pub fn build(
  builder: RetryPolicyBuilder,
) -> Result(RetryPolicy, DeliveryConfigurationError)

Validate and build one retry and dead-letter policy.

codec

</>

pub fn codec(
  encode encode: fn(event) -> Proposed(event),
  decode decode: fn(StoredEvent) -> Result(
    factos.Decoded(event),
    DecodeError,
  ),
) -> EventCodec(event)

Create a new codec.

encode turns a domain event into a Proposed event ready for persistence. decode turns a stored row back into a domain event. Decode failures are returned as DecodeError and stop load/read flows rather than panicking.

WARNING: codecs used by dispatch must be pure. Serializable dispatch may retry and call encode or decode more than once for the same logical operation. Codec functions must be deterministic conversions only; external side effects belong in durable outbox records after commit.

decoded_event

</>

pub fn decoded_event(
  stored: StoredEvent,
  event event: event,
) -> Result(factos.Decoded(event), DecodeError)

Return one successfully decoded domain event with its stored envelope.

The decoder remains responsible for validating the event type and payload. This helper preserves the stored type, version, tags, and metadata.

delivered

</>

pub fn delivered() -> DeliveryOutcome

Classify one leased message as successfully delivered.

dispatch

</>

pub fn dispatch(
  builder: DispatchBuilder(command, state, event, domain_error),
  command: command,
  event_id event_id: fn() -> String,
) -> Result(Dispatch(event), Error(domain_error))

Dispatch a command through the bounded context’s configured dispatcher.

Every configured subscription observes every newly appended event.

dispatcher

</>

pub fn dispatcher(
  codec codec: EventCodec(event),
  subscriptions subscriptions: List(Subscription(event)),
) -> Result(Dispatcher(event), DispatcherConfigurationError)

Configure and validate the complete subscription topology for a bounded context.

Subscription names must be non-blank and unique. Valid names are preserved exactly as supplied.

effect_codec

</>

pub fn effect_codec(
  encode encode: fn(effect) -> ProposedEffect,
) -> EffectCodec(effect)

Create an effect codec.

WARNING: effect codecs used by dispatch must be pure. Serializable dispatch may retry and call encode more than once for the same effect value. This function must only build a deterministic durable outbox envelope; it must not execute the effect or perform any other host-system side effect.

error_to_string

</>

pub fn error_to_string(
  error: Error(domain_error),
  domain_error_to_string: fn(domain_error) -> String,
) -> String

lease_outbox

</>

pub fn lease_outbox(
  connection: pog.Connection,
  consumer consumer: OutboxConsumer,
  limit limit: Int,
  lease_for_milliseconds lease_for_milliseconds: Int,
) -> Result(List(OutboxMessage), Error(domain_error))

Lease pending outbox messages for a configured consumer and target.

Leased messages stay in pending status but are hidden from competing workers until locked_until expires. attempt is incremented atomically and starts at one.

list_dead_letters

</>

pub fn list_dead_letters(
  connection: pog.Connection,
  consumer consumer: option.Option(String),
  target target: option.Option(String),
  limit limit: Int,
) -> Result(List(DeadLetter), Error(domain_error))

List dead letters, optionally filtered by consumer and target.

load_stream

</>

pub fn load_stream(
  connection: pog.Connection,
  stream stream_name: String,
  decider decider: factos.Decider(
    command,
    state,
    event,
    domain_error,
  ),
  codec codec: EventCodec(event),
) -> Result(
  factos.LoadedStream(event, state),
  Error(domain_error),
)

Load and fold one stream.

This supports classic stream-revision consistency. The returned factos.LoadedStream contains the folded state, decoded recorded events, and current stream revision.

new_dispatch

</>

pub fn new_dispatch(
  dispatcher dispatcher: Dispatcher(event),
  connection connection: pog.Connection,
  stream stream_name: String,
  decider decider: factos.Decider(
    command,
    state,
    event,
    domain_error,
  ),
) -> DispatchBuilder(command, state, event, domain_error)

Start building an event dispatch from a required configured dispatcher.

By default the builder uses one-stream consistency and 5 attempts for retryable serializable transaction conflicts.

WARNING: every function configured on the dispatcher or passed into dispatch must be pure. External work belongs in the durable outbox after commit.

new_proposed

</>

pub fn new_proposed(
  event event: event,
  type_ type_: factos.EventType,
  version version: Int,
  data data: BitArray,
) -> Proposed(event)

Prepare a domain event with empty tags and metadata for persistence.

new_retry_policy

</>

pub fn new_retry_policy() -> RetryPolicyBuilder

Start a retry policy builder with production-oriented defaults.

The defaults are 12 maximum attempts, a 30-second initial delay, a 15-minute maximum delay, a 24-hour maximum delivery age, and 20 percent jitter.

outbox_consumer

</>

pub fn outbox_consumer(
  name name: String,
  target target: String,
  policy policy: RetryPolicy,
) -> Result(OutboxConsumer, DeliveryConfigurationError)

Bind one consumer and target to a validated retry policy.

permanent_failure

</>

pub fn permanent_failure(
  reason reason: String,
) -> DeliveryOutcome

Classify one leased message as a failure that must not be retried.

proposed_effect

</>

pub fn proposed_effect(
  consumer consumer: String,
  key key: String,
  target target: String,
  type_ type_: String,
  metadata metadata: factos.Metadata,
  payload payload: BitArray,
) -> ProposedEffect

Prepare a durable integration effect for outbox persistence.

read_context

</>

pub fn read_context(
  connection: pog.Connection,
  query query: factos.Query,
  decider decider: factos.Decider(
    command,
    state,
    event,
    domain_error,
  ),
  codec codec: EventCodec(event),
) -> Result(factos.Context(event, state), Error(domain_error))

Read and fold the facts selected by a command-context query.

The backend reads stored rows, decodes them with the supplied codec, filters them with factos.matches_query, folds matching events with the decider’s evolve function, and returns a factos.Context with a FailIfEventsMatch(query, after) append condition.

read_events_after

</>

pub fn read_events_after(
  connection: pog.Connection,
  query query: factos.Query,
  after after: factos.SequencePosition,
  limit limit: Int,
  codec codec: EventCodec(event),
) -> Result(List(factos.Recorded(event)), Error(domain_error))

Read committed events after a global sequence position.

This is the bounded polling primitive used by durable subscriptions and process managers. Events are returned in global append order.

read_outbox_source_event

</>

pub fn read_outbox_source_event(
  dispatcher configured_dispatcher: Dispatcher(event),
  connection connection: pog.Connection,
  message message: OutboxMessage,
) -> Result(factos.Recorded(event), Error(Nil))

Load and decode the authoritative event referenced by an outbox message.

replay_dead_letter

</>

pub fn replay_dead_letter(
  connection: pog.Connection,
  id id: Int,
  attempts attempts: ReplayAttempts,
) -> Result(ReplayResult, Error(domain_error))

Replay a dead letter without replacing its original payload.

retryable_failure

</>

pub fn retryable_failure(
  reason reason: String,
) -> DeliveryOutcome

Classify one leased message as a failure that may be retried.

settle_outbox

</>

pub fn settle_outbox(
  connection: pog.Connection,
  consumer consumer: OutboxConsumer,
  message message: OutboxMessage,
  outcome outcome: DeliveryOutcome,
) -> Result(DeliveryFinalization, Error(domain_error))

Atomically settle one leased delivery using its consumer’s configured policy.

Applications classify failures but cannot choose retry delays or bypass the configured attempt and age budgets.

subscription

</>

pub fn subscription(
  name name: String,
  reactor reactor: factos.Reactor(event, effect),
  codec effect_codec: EffectCodec(effect),
) -> Subscription(event)

Define a named durable subscription.

Subscription names namespace effect keys, so two subscriptions can react to the same event without colliding in the outbox.

with_initial_delay

</>

pub fn with_initial_delay(
  builder: RetryPolicyBuilder,
  milliseconds milliseconds: Int,
) -> RetryPolicyBuilder

Override the delay before the first retry.

with_jitter

</>

pub fn with_jitter(
  builder: RetryPolicyBuilder,
  percent percent: Int,
) -> RetryPolicyBuilder

Override the retry-delay jitter percentage.

with_maximum_age

</>

pub fn with_maximum_age(
  builder: RetryPolicyBuilder,
  milliseconds milliseconds: Int,
) -> RetryPolicyBuilder

Override the maximum age of an effect eligible for retry.

with_maximum_attempts

</>

pub fn with_maximum_attempts(
  builder: RetryPolicyBuilder,
  attempts attempts: Int,
) -> RetryPolicyBuilder

Override the maximum number of delivery attempts.

with_maximum_delay

</>

pub fn with_maximum_delay(
  builder: RetryPolicyBuilder,
  milliseconds milliseconds: Int,
) -> RetryPolicyBuilder

Override the maximum retry delay.

with_metadata

</>

pub fn with_metadata(
  proposed: Proposed(event),
  metadata metadata: factos.Metadata,
) -> Proposed(event)

Replace the metadata on a proposed event.

with_query

</>

pub fn with_query(
  builder: DispatchBuilder(command, state, event, domain_error),
  query query: factos.Query,
) -> DispatchBuilder(command, state, event, domain_error)

Use a context query instead of one-stream consistency.

with_retry_attempts

</>

pub fn with_retry_attempts(
  builder: DispatchBuilder(command, state, event, domain_error),
  attempts attempts: Int,
) -> DispatchBuilder(command, state, event, domain_error)

Override retry attempts for PostgreSQL serializable/deadlock conflicts.

with_tags

</>

pub fn with_tags(
  proposed: Proposed(event),
  tags tags: List(factos.Tag),
) -> Proposed(event)

Replace the tags on a proposed event.