Outbox

Publishes events to other systems, such as Kafka, NATS or SQS, as integration events.

const TodoOutbox = Outbox.make({
  name: "todo-outbox",
  aggregates: [Todo],
  messages: {
    TodoCreated: ({ title }) => ({ type: "todo.created.v1", payload: { title } })
  },
  publish: (messages) => producer.send(messages)
})

const Publishing = Outbox.layer([TodoOutbox])

An outbox is a view model: it reads the store from its checkpoint, so it publishes what was stored while the process was down, and nothing is lost between storing an event and publishing it. Delivery is at least once. After a crash, events published since the last checkpoint are published again, so consumers deduplicate on Message.id, which stays the same.

Messages are published in batches of up to batchSize. A batch goes out when it is full or linger after its first message, and the checkpoint never moves past a message that wasn’t published. A failed batch is retried, in order, until it succeeds.

Classes

OutboxError

Defined in: src/Outbox.ts:83

Publishing failed, which the outbox retries.

Extends

• object & YieldableError<this>

Constructors

Constructor

new OutboxError(…args): OutboxError

Defined in: node_modules/.pnpm/effect@4.0.6/node_modules/effect/dist/Schema.d.ts:10979

Parameters

args

…[props: object, options?: MakeOptions]

Returns

OutboxError

Inherited from

Schema.TaggedError<OutboxError>()("OutboxError", { outbox: Schema.String, cause: Schema.Defect() }).constructor

Properties

cause

cause: unknown

Defined in: src/Outbox.ts:85

Inherited from

Schema.TaggedError<OutboxError>()("OutboxError", { outbox: Schema.String, cause: Schema.Defect() }).cause

outbox

readonly outbox: string = Schema.String

Defined in: src/Outbox.ts:84

Inherited from

Schema.TaggedError<OutboxError>()("OutboxError", { outbox: Schema.String, cause: Schema.Defect() }).outbox

---

Outboxes

Defined in: src/Outbox.ts:142

Extends

• Shape<"effect-cqrs/Outboxes", { status: Effect<readonly Status[]>; published(position?): Effect<void, EventStoreError | OutboxError>; }, this>

Constructors

Constructor

new Outboxes(_): Outboxes

Defined in: node_modules/.pnpm/effect@4.0.6/node_modules/effect/dist/Context.d.ts:121

Parameters

_

never

Returns

Outboxes

Inherited from

Context.Service<Outboxes, { /** * Waits until every event up to `position`, by default the latest, is * published. Mostly for tests. */ published(position?: number): Effect.Effect<void, EventStoreError | OutboxError> /** Where each outbox is, and the failure it is retrying, see `ViewModel.Status`. */ readonly status: Effect.Effect<ReadonlyArray<ViewModel.Status>> }>()(“effect-cqrs/Outboxes”).constructor

Interfaces

Mapped

Defined in: src/Outbox.ts:57

What a mapping makes of an event. undefined publishes nothing for it.

Properties

payload

readonly payload: unknown

Defined in: src/Outbox.ts:59

type

readonly type: string

Defined in: src/Outbox.ts:58

---

Message

Defined in: src/Outbox.ts:40

An integration event, what a publisher sends.

Properties

aggregateId

readonly aggregateId: string

Defined in: src/Outbox.ts:46

aggregateName

readonly aggregateName: string

Defined in: src/Outbox.ts:45

causationId

readonly causationId: string | null

Defined in: src/Outbox.ts:53

correlationId

readonly correlationId: string | null

Defined in: src/Outbox.ts:52

id

readonly id: string

Defined in: src/Outbox.ts:42

The id of the event it comes from, the same however often it is published: todo/a@1.

payload

readonly payload: unknown

Defined in: src/Outbox.ts:44

position

readonly position: number

Defined in: src/Outbox.ts:50

The position of the event in the store.

timestamp

readonly timestamp: Utc

Defined in: src/Outbox.ts:51

type

readonly type: string

Defined in: src/Outbox.ts:43

version

readonly version: number

Defined in: src/Outbox.ts:48

The version of the event in its stream.

---

Outbox

Defined in: src/Outbox.ts:69

Type Parameters

Name

Name extends string

E

E

R

R

Properties

aggregates

readonly aggregates: readonly Any[]

Defined in: src/Outbox.ts:71

batchSize

readonly batchSize: number

Defined in: src/Outbox.ts:74

linger

readonly linger: Duration

Defined in: src/Outbox.ts:75

mappings

readonly mappings: Record<string, (event, meta) => Mapped | undefined>

Defined in: src/Outbox.ts:72

name

readonly name: Name

Defined in: src/Outbox.ts:70

publish

readonly publish: (messages) => Effect<void, E, R>

Defined in: src/Outbox.ts:73

Parameters

messages

readonly Message[]

Returns

Effect<void, E, R>

Type Aliases

Any

Any = Outbox<string, any, any>

Defined in: src/Outbox.ts:78

---

Context

Context<O> = O extends Outbox<any, any, infer R> ? R : never

Defined in: src/Outbox.ts:80

Type Parameters

O

O extends Any

---

Mappings

Mappings<Event> = { readonly [Tag in Event[“_tag”]]?: (event: Extract<Event, { _tag: Tag }>, meta: EventMeta) => Mapped | undefined }

Defined in: src/Outbox.ts:62

Type Parameters

Event

Event extends Tagged

Functions

layer()

layer<Os>(outboxes): Layer<Outboxes, EventStoreError, EventStore | Checkpoints | Context<Os[number]>>

Defined in: src/Outbox.ts:233

Runs the outboxes in the background for as long as the layer lives. Each one keeps a checkpoint under its name in Checkpoints, which should be durable: with an in-memory checkpoint an outbox publishes everything again on restart.

Type Parameters

Os

Os extends readonly Any[]

Parameters

outboxes

Os

Returns

Layer<Outboxes, EventStoreError, EventStore | Checkpoints | Context<Os[number]>>

---

make()

make<Name, Aggregates, E, R>(options): Outbox<Name, E, R>

Defined in: src/Outbox.ts:99

Defines an outbox. messages maps events, by tag, to integration events, and only those are published. Without it every event of the aggregates is published, typed <aggregate>.<tag>, with its JSON form as payload.

Mapping is where domain events become a contract with other teams, and where personal data is left out: events reach it decrypted, see PersonalData.

Type Parameters

Name

Name extends string

Aggregates

Aggregates extends readonly Any[]

E

E = never

R

R = never

Parameters

options

aggregates

Aggregates

batchSize?

number

The most messages in one call to publish. Defaults to 100.

linger?

Input

How long a batch waits for more messages before it is published. Defaults to 10 milliseconds.

messages?

Mappings<Event<Aggregates[number]>>

name

Name

publish

(messages) => Effect<void, E, R>

Returns

Outbox<Name, E, R>