An order, from its payment to its shipment or its cancellation. It shows operations with idempotency keys: a charge and a refund that the payment provider performs once however often the runner tries them, the warehouse told under a key of its own, and an email sent at most once. A cancellation that comes after the payment refunds it, and one that comes before the shipment wins over it. A query finds the orders nobody paid, by an index of the projection. Two topics of the outbox tell a billing service and a dashboard what changed, and a report counts the orders by status from the log. A runner of Effect Cluster runs the same program, unchanged.
order.ts: the program.report.ts: a store that a consumer of the topic status fills, page by page.scenarios.test.ts: the order's scenarios, rehearsed.order.test.ts: the order's properties, checked for generated states.runtime.test.ts: the orders on the runtime, with their query, their outbox and the report.cluster.test.ts: the orders on a runner of Effect Cluster.The guide's page on the outbox is written on this example.
/**
* An order, from its creation to its shipment or its cancellation. The customer pays before a deadline, then
* the warehouse is told and ships. A cancellation after the payment refunds it. The warehouse reports a shipment
* before it hands the parcel over, and ships only if the order accepts the report: a cancellation that comes
* first wins, and nothing needs to stop the warehouse.
*/
import * as Context from 'effect/Context'
import * as DateTime from 'effect/DateTime'
import * as Effect from 'effect/Effect'
import * as Match from 'effect/Match'
import * as Schema from 'effect/Schema'
import {
defect,
type IdempotencyKey,
noWork,
type Origin,
operation,
program,
reject,
type Rejected,
timer,
} from '@fgaudo/effect-tea-persistent/program'
const Order = Schema.Struct({
customer: Schema.String,
totalInCents: Schema.Int,
})
type Order = typeof Order.Type
const ChargeId = Schema.String
const State = Schema.Union([
Schema.TaggedStruct('AwaitingPayment', { order: Order, deadline: Schema.DateTimeUtcFromMillis }),
// Only the charge's completion leaves it: a cancellation meanwhile is rejected.
Schema.TaggedStruct('Charging', { order: Order, deadline: Schema.DateTimeUtcFromMillis, payment: Schema.String }),
Schema.TaggedStruct('Paid', {
order: Order,
charge: ChargeId,
// A cancellation waits for `Told`: until then the warehouse is being told, an operation only its completion
// ends. Once told, the warehouse ships only if the order accepts its report, so a refund after `Told` never
// leaves a parcel on its way.
warehouse: Schema.Literals(['Telling', 'Told']),
}),
Schema.TaggedStruct('Refunding', { order: Order, charge: ChargeId }),
Schema.TaggedStruct('Shipped', {
order: Order,
charge: ChargeId,
tracking: Schema.String,
// An email is not idempotent: after a crash while sending it, the customer may or may not have it.
email: Schema.Literals(['Sending', 'Sent', 'Unsure']),
}),
Schema.TaggedStruct('Cancelled', {
order: Order,
reason: Schema.Literals(['ByCustomer', 'Unpaid', 'Refunded']),
}),
])
export type State = typeof State.Type
const Flags = Schema.Struct({ order: Order })
type Flags = typeof Flags.Type
const Inbound = Schema.Union([
Schema.TaggedStruct('PayRequested', { payment: Schema.String }),
Schema.TaggedStruct('CancelRequested', {}),
Schema.TaggedStruct('ShipmentReported', { tracking: Schema.String }),
])
type Inbound = typeof Inbound.Type
const Internal = Schema.Union([
Schema.TaggedStruct('ChargeSucceeded', { charge: ChargeId }),
Schema.TaggedStruct('ChargeDeclined', {}),
Schema.TaggedStruct('WarehouseTold', {}),
Schema.TaggedStruct('RefundCompleted', {}),
Schema.TaggedStruct('PaymentDeadlinePassed', {}),
Schema.TaggedStruct('ShipmentEmailed', {}),
Schema.TaggedStruct('ShipmentEmailUnsure', {}),
])
type Internal = typeof Internal.Type
// Who may send a message: it arrives from outside, and waits in the mailbox with it, so it has a schema.
const Sender = Schema.Union([
Schema.TaggedStruct('Customer', { id: Schema.String }),
Schema.TaggedStruct('Warehouse', {}),
])
export type Sender = typeof Sender.Type
// What callers read: it leaves the process, so it has a schema too.
const Projection = Schema.Struct({
status: Schema.Literals(['AwaitingPayment', 'Charging', 'Paid', 'Refunding', 'Shipped', 'Cancelled']),
// By when the customer has to pay, while there is still something to pay: it is what a query reads to find
// the orders nobody paid, so what it filters by is what the customer is told, in milliseconds, since the
// projection is JSON (decisions 0032, 0139).
deadline: Schema.NullOr(Schema.Number),
// What the order costs, which a sum over the orders still awaiting their payment adds up (decision 0148).
totalInCents: Schema.Int,
canPay: Schema.Boolean,
canCancel: Schema.Boolean,
// What billing acts on: the charge the order took, and whether it was given back. The projection is the one
// contract of the order, so what a consumer of the log needs is in it too (decision 0138).
charge: Schema.NullOr(ChargeId),
refunded: Schema.Boolean,
})
export type Projection = typeof Projection.Type
export class Payments extends Context.Service<Payments, {
readonly charge: (
request: Readonly<{ payment: string; totalInCents: number; idempotencyKey: IdempotencyKey }>,
) => Effect.Effect<string, 'Declined'>
readonly refund: (request: Readonly<{ charge: string; idempotencyKey: IdempotencyKey }>) => Effect.Effect<void>
}>()('Payments') {}
export class Warehouse extends Context.Service<Warehouse, {
readonly prepare: (request: Readonly<{ order: Order; idempotencyKey: IdempotencyKey }>) => Effect.Effect<void>
}>()('Warehouse') {}
/**
* The application's mailer, which finds a customer's address in its own table: the order knows its customer by id
* alone, and holds no address.
*/
export class Mailer extends Context.Service<Mailer, {
readonly send: (email: Readonly<{ customer: string; subject: string }>) => Effect.Effect<void>
}>()('Mailer') {}
const paymentWindow = { minutes: 30 }
const isCustomer = (from: Origin<Sender>, order: Order) =>
from._tag === 'Caller' && from.sender._tag === 'Customer' && from.sender.id === order.customer
const isWarehouse = (from: Origin<Sender>) => from._tag === 'Caller' && from.sender._tag === 'Warehouse'
// The work an order's states hold, as data: a test sees it without running it, and the runtime stores it.
const Operation = Schema.TaggedUnion({
Charge: { payment: Schema.String, totalInCents: Schema.Int },
TellWarehouse: { order: Order },
EmailShipment: { customer: Schema.String, tracking: Schema.String },
Refund: { charge: ChargeId },
})
type Operation = typeof Operation.Type
const charge = Effect.fn('Charge')(
function*({ payment, totalInCents }: Extract<Operation, { _tag: 'Charge' }>, idempotencyKey: IdempotencyKey) {
const payments = yield* Payments
return yield* payments.charge({ payment, totalInCents, idempotencyKey })
},
Effect.match({
onSuccess: (charge): Internal => ({ _tag: 'ChargeSucceeded', charge }),
onFailure: (): Internal => ({ _tag: 'ChargeDeclined' }),
}),
)
const tellWarehouse = Effect.fn('TellWarehouse')(
function*({ order }: Extract<Operation, { _tag: 'TellWarehouse' }>, idempotencyKey: IdempotencyKey) {
const warehouse = yield* Warehouse
yield* warehouse.prepare({ order, idempotencyKey })
return { _tag: 'WarehouseTold' } satisfies Internal
},
)
const emailShipment = Effect.fn('EmailShipment')(
function*({ customer, tracking }: Extract<Operation, { _tag: 'EmailShipment' }>) {
const mailer = yield* Mailer
yield* mailer.send({ customer, subject: `Your order has shipped: ${tracking}` })
return { _tag: 'ShipmentEmailed' } satisfies Internal
},
)
const refund = Effect.fn('Refund')(
function*({ charge }: Extract<Operation, { _tag: 'Refund' }>, idempotencyKey: IdempotencyKey) {
const payments = yield* Payments
yield* payments.refund({ charge, idempotencyKey })
return { _tag: 'RefundCompleted' } satisfies Internal
},
)
export const order = program({
state: State,
flags: Flags,
inbound: Inbound,
internal: Internal,
operation: Operation,
sender: Sender,
projection: Projection,
})({
name: 'Order',
keys: ['payment-deadline', 'charge', 'tell-warehouse', 'email-shipment', 'refund'],
init: ({ order }, at) => ({
_tag: 'AwaitingPayment',
order,
deadline: DateTime.add(at, paymentWindow),
}),
// Inbound messages may arrive in any state, and are rejected where they do not apply. An internal message
// arrives only while its key is in `work`, so elsewhere it is a defect.
update: (message, { from }) => (state) =>
Match.value(message).pipe(
Match.withReturnType<State | Rejected>(),
Match.tagsExhaustive({
PayRequested: ({ payment }) =>
state._tag === 'AwaitingPayment' && isCustomer(from, state.order)
? { _tag: 'Charging', order: state.order, deadline: state.deadline, payment }
: reject,
CancelRequested: () =>
!isCustomer(from, state.order)
? reject
: state._tag === 'AwaitingPayment'
? { _tag: 'Cancelled', order: state.order, reason: 'ByCustomer' }
: state._tag === 'Paid' && state.warehouse === 'Told'
? { _tag: 'Refunding', order: state.order, charge: state.charge }
: reject,
ShipmentReported: ({ tracking }) =>
state._tag === 'Paid' && state.warehouse === 'Told' && isWarehouse(from)
? { _tag: 'Shipped', order: state.order, charge: state.charge, tracking, email: 'Sending' }
: reject,
ChargeSucceeded: ({ charge }) =>
state._tag === 'Charging'
? { _tag: 'Paid', order: state.order, charge, warehouse: 'Telling' }
: defect('a charge completed while nothing was charging'),
// The deadline may have passed meanwhile: its timer comes back, and fires at once.
ChargeDeclined: () =>
state._tag === 'Charging'
? { _tag: 'AwaitingPayment', order: state.order, deadline: state.deadline }
: defect('a charge was declined while nothing was charging'),
WarehouseTold: () =>
state._tag === 'Paid' && state.warehouse === 'Telling'
? { ...state, warehouse: 'Told' }
: defect('the warehouse was told while nobody was telling it'),
RefundCompleted: () =>
state._tag === 'Refunding'
? { _tag: 'Cancelled', order: state.order, reason: 'Refunded' }
: defect('a refund completed while nothing was refunding'),
PaymentDeadlinePassed: () =>
state._tag === 'AwaitingPayment'
? { _tag: 'Cancelled', order: state.order, reason: 'Unpaid' }
: defect('the payment deadline passed outside its wait'),
ShipmentEmailed: () =>
state._tag === 'Shipped' && state.email === 'Sending'
? { ...state, email: 'Sent' }
: defect('a shipment email was sent while nothing was sending it'),
ShipmentEmailUnsure: () =>
state._tag === 'Shipped' && state.email === 'Sending'
? { ...state, email: 'Unsure' }
: defect('a shipment email was unsure while nothing was sending it'),
}),
),
work: (state) =>
Match.valueTags(state, {
AwaitingPayment: ({ deadline }) => ({
'payment-deadline': timer(deadline, { _tag: 'PaymentDeadlinePassed' }),
}),
Charging: ({ order, payment }) => ({
charge: Operation.cases.Charge.make({ payment, totalInCents: order.totalInCents }),
}),
Paid: ({ order, warehouse }) =>
warehouse === 'Telling' ? { 'tell-warehouse': Operation.cases.TellWarehouse.make({ order }) } : noWork,
Refunding: ({ charge }) => ({ refund: Operation.cases.Refund.make({ charge }) }),
Shipped: ({ order, tracking, email }) =>
email === 'Sending'
? { 'email-shipment': Operation.cases.EmailShipment.make({ customer: order.customer, tracking }) }
: noWork,
Cancelled: () => noWork,
}),
operations: {
// A payment provider caps the length of an idempotency key, and an order's id may be long: the provider is
// given the key's digest.
Charge: operation.digest(charge),
TellWarehouse: operation(tellWarehouse),
// Sent at most once: a second email is worse than none, and the state says when it is not known.
EmailShipment: operation.atMostOnce(emailShipment, (): Internal => ({ _tag: 'ShipmentEmailUnsure' })),
Refund: operation.digest(refund),
},
project: (state) => ({
status: state._tag,
deadline: state._tag === 'AwaitingPayment' || state._tag === 'Charging'
? DateTime.toEpochMillis(state.deadline)
: null,
totalInCents: state.order.totalInCents,
canPay: state._tag === 'AwaitingPayment',
canCancel: state._tag === 'AwaitingPayment' || (state._tag === 'Paid' && state.warehouse === 'Told'),
charge: 'charge' in state ? state.charge : null,
refunded: state._tag === 'Cancelled' && state.reason === 'Refunded',
}),
// The topics of the log, each selecting of the projection what a consumer acts on: a row is written in a topic
// only when that changed, and gives it (decisions 0138, 0140). A dashboard reads the status. Billing reads the
// charge and the refund, and is not told when the warehouse is.
topics: {
status: ({ status }) => status,
billing: ({ charge, refunded }) => ({ charge, refunded }),
},
// "The orders nobody paid by now": the deadline of an order still awaiting its payment, as the projection
// holds it, and nothing for an order that is being charged, paid or over (decisions 0032, 0139). "What they
// owe": the total of such an order, which a sum adds up and leaves out where it is null (decision 0148). "How
// many orders are in a status": its status, which a count takes one status at a time; how many are in each is
// a report, built from the log in `report.ts` (decision 0149).
indexes: {
unpaidUntil: ({ deadline, status }) => (status === 'AwaitingPayment' ? deadline : null),
owedInCents: ({ status, totalInCents }) => (status === 'AwaitingPayment' ? totalInCents : null),
status: ({ status }) => status,
},
})
/**
* The first report built from the log: how many orders are in each status, which a dashboard shows at every view.
* The read model answers it too, with `countBy` (decision 0150). A report answers it from a store made for the
* question, which a consumer of the topic `status` fills, page by page (decision 0149): a store of its own is what
* a question the read model does not answer needs. The store is a plain table of the test's database, beside the
* library's.
*/
import * as Effect from 'effect/Effect'
import * as Schema from 'effect/Schema'
import { SqlClient } from 'effect/sql'
import type { Published, TypesOf } from '@fgaudo/effect-tea-persistent'
import type { order } from './order.ts'
type Order = TypesOf<typeof order>
const Counts = Schema.Array(Schema.Struct({ status: Schema.String, count: Schema.Number }))
export const statusReport = Effect.gen(function*() {
const sql = yield* SqlClient.SqlClient
// The store holds the last status of each order, and not a count per status: a row of the topic gives the
// status the step left and not the one before, so a count cannot be moved from one status to the other on a row
// alone. The count per status is a grouping over the store.
yield* sql`CREATE TABLE order_status (id TEXT PRIMARY KEY, status TEXT NOT NULL)`
return {
/**
* Writes a page of the topic `status` to the store, as the handler of `consume`. A reader of `consume` writes it
* and acks it in one transaction of the database the runtime was launched with (decisions 0150, 0184): the store
* and the cursors commit together, and a page read again after a crash was never written.
*/
record: (page: ReadonlyArray<Published<Order, 'status'>>) =>
Effect.forEach(
page,
({ id, value }) =>
sql`
INSERT INTO order_status (id, status) VALUES (${id}, ${value})
ON CONFLICT (id) DO UPDATE SET status = excluded.status
`,
{ discard: true },
).pipe(Effect.orDie),
/** How many orders are in each status, by status, with no entry for a status no order is in. */
counts: sql`SELECT status, count(*) AS count FROM order_status GROUP BY status`.pipe(
Effect.flatMap(Schema.decodeUnknownEffect(Counts)),
Effect.map((rows) => Object.fromEntries(rows.map(({ status, count }) => [status, count]))),
Effect.orDie,
),
}
})
/**
* The orders on a runner of Effect Cluster: the launcher's side of the runtime on Cluster. The runner names the
* programs it runs where its `Sharding` is built, by `sharding`, which records their versions before Cluster
* registers it (decisions 0159, 0174), and its runtime is the `launch` of `index.cluster.ts`. Every request goes
* through Cluster's client of the program's door (decision 0165). On Cluster the runtime runs on SQLite alone
* (decision 0084).
*/
import { SqliteClient } from '@effect/sql-sqlite-bun'
import { expect, test } from 'bun:test'
import {
RunnerAddress,
RunnerHealth,
Runners,
ShardingConfig,
SqlMessageStorage,
SqlRunnerStorage,
} from 'effect/cluster'
import { Digest } from 'effect/crypto'
import * as Effect from 'effect/Effect'
import * as Layer from 'effect/Layer'
import * as Option from 'effect/Option'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { launch, sharding } from '@fgaudo/effect-tea-persistent/cluster'
import { Mailer, order, Payments, Warehouse } from './order.ts'
const ada = { _tag: 'Customer', id: 'ada' } as const
/** How many charges the world was asked for in the test. */
let charges = 0
/** The world the orders act on, which counts its charges and records nothing else. */
const world = Layer.mergeAll(
Layer.succeed(Payments, Payments.of({
charge: () =>
Effect.sync(() => {
charges++
return 'ch_1'
}),
refund: () => Effect.void,
})),
Layer.succeed(Warehouse, Warehouse.of({ prepare: () => Effect.void })),
Layer.succeed(Mailer, Mailer.of({ send: () => Effect.void })),
)
/**
* A runner of the orders over the SQLite file `filename`. It is the layer `SingleRunner.layer` of Cluster builds,
* with `sharding` in place of `Sharding.layer`: `SingleRunner.layer` builds `Sharding` itself, so the versions would
* be recorded after Cluster registered the runner. A runner of its own process sends to the others with
* `Runners.layerRpc`, and a launcher that pings its runners or asks Kubernetes provides its own `RunnerHealth`.
*/
const runner = (filename: string) =>
sharding([order]).pipe(
Layer.provideMerge(Runners.layerNoop),
Layer.provideMerge(SqlMessageStorage.layer),
Layer.provide([Layer.orDie(SqlRunnerStorage.layer), RunnerHealth.layerNoop]),
Layer.provideMerge(ShardingConfig.layer({ runnerAddress: Option.some(RunnerAddress.make('localhost', 34431)) })),
Layer.provideMerge(SqliteClient.layer({ filename })),
Layer.provide(Digest.layerWebCrypto),
)
test('a runner of Cluster creates, pays and reads the orders, each request through the door of the orders', async () => {
const directory = mkdtempSync(join(tmpdir(), 'effect-tea-persistent-example-'))
try {
await Effect.runPromise(
Effect.scoped(Effect.gen(function*() {
const runtime = yield* launch()
const orders = yield* Effect.provide(runtime.deploy(order), world)
for (const id of ['o-1', 'o-2']) {
expect(
yield* orders.create(id, {
flags: { order: { customer: 'ada', totalInCents: 1_000 } },
requestId: `c-${id}`,
sender: ada,
}),
).toEqual({ _tag: 'Created' })
}
const pay = { _tag: 'PayRequested', payment: 'card-1' } as const
expect(yield* orders.send('o-2', pay, { requestId: 'p-o-2', sender: ada, keep: 'forever' })).toEqual({ _tag: 'Accepted' })
// A retry under the same request id is answered from what Cluster stored, and charges nothing again.
expect(yield* orders.send('o-2', pay, { requestId: 'p-o-2', sender: ada, keep: 'forever' })).toEqual({ _tag: 'Accepted' })
yield* runtime.settled
expect(yield* orders.read('o-1')).toMatchObject({ status: 'AwaitingPayment' })
expect(yield* orders.read('o-2')).toMatchObject({ status: 'Paid', charge: 'ch_1' })
expect(charges).toBe(1)
expect(yield* orders.count({ where: [{ index: 'unpaidUntil', is: '<>', value: null }] })).toBe(1)
})).pipe(Effect.provide(Layer.orDie(runner(join(directory, 'orders.db'))))),
)
} finally {
rmSync(directory, { recursive: true, force: true })
}
}, 30_000)
/**
* The order's properties, for generated states.
*/
import { describe, expect, test } from 'bun:test'
import * as DateTime from 'effect/DateTime'
import fc from 'fast-check'
import { projectionViolations } from '@fgaudo/effect-tea-persistent/testing'
import { order, type Sender, type State } from './order.ts'
const at = DateTime.makeUnsafe('2026-10-30T10:00:00Z')
const time = fc.date({ min: new Date('2026-01-01'), max: new Date('2027-01-01'), noInvalidDate: true })
.map(DateTime.fromDateUnsafe)
const anOrder = fc.record({ customer: fc.constantFrom('ada', 'bob'), totalInCents: fc.nat() })
// Every case of the order's states, with any data.
const state: fc.Arbitrary<State> = fc.oneof(
fc.record({ _tag: fc.constant('AwaitingPayment'), order: anOrder, deadline: time }),
fc.record({ _tag: fc.constant('Charging'), order: anOrder, deadline: time, payment: fc.string() }),
fc.record({
_tag: fc.constant('Paid'),
order: anOrder,
charge: fc.string(),
warehouse: fc.constantFrom('Telling', 'Told'),
}),
fc.record({ _tag: fc.constant('Refunding'), order: anOrder, charge: fc.string() }),
fc.record({
_tag: fc.constant('Shipped'),
order: anOrder,
charge: fc.string(),
tracking: fc.string(),
email: fc.constantFrom('Sending', 'Sent', 'Unsure'),
}),
fc.record({
_tag: fc.constant('Cancelled'),
order: anOrder,
reason: fc.constantFrom('ByCustomer', 'Unpaid', 'Refunded'),
}),
)
describe('order', () => {
// The projection speaks to the customer: the warehouse's reports are not among its actions.
test('the projection allows what update accepts, all of it', () => {
const check = projectionViolations(order, {
PayRequested: (projection) => projection.canPay,
CancelRequested: (projection) => projection.canCancel,
})
fc.assert(fc.property(state, (state) => {
const customer: Sender = { _tag: 'Customer', id: state.order.customer }
expect(check(state, [
{ message: { _tag: 'PayRequested', payment: 'card-1' }, sender: customer },
{ message: { _tag: 'CancelRequested' }, sender: customer },
], at)).toEqual([])
}))
})
})
/**
* The orders on the runtime, and the question the stored states could not answer: which of them nobody paid.
* The read model is the projection, written in each step's transaction, and `unpaidUntil` is the index a query
* reads it by (decision 0032).
*/
import { SqliteClient } from '@effect/sql-sqlite-bun'
import { expect, test } from 'bun:test'
import * as DateTime from 'effect/DateTime'
import * as Deferred from 'effect/Deferred'
import * as Effect from 'effect/Effect'
import * as Exit from 'effect/Exit'
import * as Fiber from 'effect/Fiber'
import * as Layer from 'effect/Layer'
import { SqlClient } from 'effect/sql'
import { launch } from '@fgaudo/effect-tea-persistent'
import { Mailer, order, Payments, Warehouse } from './order.ts'
import { statusReport } from './report.ts'
const ada = { _tag: 'Customer', id: 'ada' } as const
/** A limit no reading of this test reaches: a reading says how many rows it takes (decision 0141). */
const all = 100
/**
* Who reads the orders' log: a billing service and a dashboard, each a consumer of its topic. They are the
* deployment's, declared where the orders are deployed, so a name mistyped does not compile (decision 0152).
*/
const consumers = { consumers: { billing: ['invoices'], status: ['dashboard'] } } as const
/** The world the orders act on, which answers and records nothing: this test only reads states. */
const world = Layer.mergeAll(
Layer.succeed(Payments, Payments.of({ charge: () => Effect.succeed('ch_1'), refund: () => Effect.void })),
Layer.succeed(Warehouse, Warehouse.of({ prepare: () => Effect.void })),
Layer.succeed(Mailer, Mailer.of({ send: () => Effect.void })),
)
const database = <A, E>(test: Effect.Effect<A, E, SqlClient.SqlClient>) =>
Effect.runPromise(
Effect.scoped(test).pipe(
Effect.provide(SqliteClient.layer({ filename: ':memory:' })),
Effect.orDie,
),
)
test('a query finds the orders nobody paid, and leaves out the one that was', () =>
database(Effect.scoped(Effect.gen(function*() {
const runtime = yield* launch()
const orders = yield* Effect.provide(runtime.deploy(order), world)
for (const id of ['o-1', 'o-2', 'o-3']) {
yield* orders.create(id, {
flags: { order: { customer: 'ada', totalInCents: 1_000 } },
requestId: `c-${id}`,
sender: ada,
})
}
yield* orders.send('o-2', { _tag: 'PayRequested', payment: 'card-1' }, { requestId: 'p-2', sender: ada, keep: 'forever' })
yield* runtime.settled
expect(yield* orders.read('o-2')).toMatchObject({ status: 'Paid', deadline: null })
// Every order is created with half an hour to pay: the ones still awaiting it are the ones to chase.
const now = DateTime.toEpochMillis(yield* DateTime.now)
const unpaid = yield* orders.query({
limit: all,
where: [{ index: 'unpaidUntil', is: '<=', value: now + 3_600_000 }],
order: { index: 'unpaidUntil' },
})
expect(unpaid.map(({ id }) => id)).toEqual(['o-1', 'o-3'])
expect(unpaid[0]?.projection).toMatchObject({ status: 'AwaitingPayment', canPay: true })
// A page reads on from the last row of the page before, by the index it orders by and then by the id.
const firstPage = yield* orders.query({ limit: 1, order: { index: 'unpaidUntil' } })
const secondPage = yield* orders.query({ limit: 2, order: { index: 'unpaidUntil' }, after: firstPage.at(-1) })
expect([...firstPage, ...secondPage].map(({ id }) => id)).toEqual(['o-2', 'o-1', 'o-3'])
// Nobody is late yet, and the order that was paid holds no deadline to be found by.
expect(yield* orders.query({ limit: all, where: [{ index: 'unpaidUntil', is: '<=', value: now }] })).toEqual([])
expect((yield* orders.query({ limit: all, where: [{ index: 'unpaidUntil', is: '=', value: null }] })).map(({ id }) => id))
.toEqual(['o-2'])
expect((yield* orders.query({ limit: all, where: [{ index: 'unpaidUntil', is: '<>', value: null }] })).map(({ id }) => id))
.toEqual(['o-1', 'o-3'])
// An order against no value finds nothing, as in SQL.
expect(yield* orders.query({ limit: all, where: [{ index: 'unpaidUntil', is: '<', value: null }] })).toEqual([])
// A number over the orders is one statement, which reads no projection: how many await their payment, and
// what they owe, the paid order left out by its null (decision 0148).
expect(yield* orders.count({ where: [{ index: 'unpaidUntil', is: '<>', value: null }] })).toBe(2)
expect(yield* orders.count({})).toBe(3)
expect(yield* orders.sum({ index: 'owedInCents' })).toBe(2_000)
}))))
test('what an order published is what another system learns of it, each consumer in its topic', () =>
database(Effect.scoped(Effect.gen(function*() {
const runtime = yield* launch()
const orders = yield* Effect.provide(runtime.deploy(order, consumers), world)
yield* orders.create('o-1', {
flags: { order: { customer: 'ada', totalInCents: 1_000 } },
requestId: 'c-o-1',
sender: ada,
})
yield* orders.send('o-1', { _tag: 'PayRequested', payment: 'card-1' }, { requestId: 'p-1', sender: ada, keep: 'forever' })
yield* runtime.settled
// The steps were the creation, the request to pay, the charge coming back, and the warehouse told. A topic has a
// row for each step that changed what it selects, holding the message the step handled and the value selected
// (decisions 0038, 0138, 0140): the status changed at the first three, the charge at the creation and at the
// charge alone, and the warehouse told changed only `canCancel`, which no topic selects.
const statuses = yield* orders.published('status', [], all)
expect(statuses.map(({ id, kind, message, value }) => [id, kind, message?._tag ?? null, value]))
.toEqual([
['o-1', 'Creation', null, 'AwaitingPayment'],
['o-1', 'Inbound', 'PayRequested', 'Charging'],
['o-1', 'Internal', 'ChargeSucceeded', 'Paid'],
])
const billing = yield* orders.published('billing', [], all)
expect(billing.map(({ kind, message, value }) => [kind, message?._tag ?? null, value.charge])).toEqual([
['Creation', null, null],
['Internal', 'ChargeSucceeded', 'ch_1'],
])
// The rows of one step share its `seq`, so a topic's `seq`s have gaps where a step published only others.
expect(billing.map(({ seq }) => seq)).toEqual([0, 2].map((at) => statuses[at]?.seq ?? 0))
// A consumer is a name within its topic, and its cursors are stored: it reads from where it acked, and a prune
// drops, in each topic, what every consumer of the topic has read (decisions 0039, 0138, 0152). The rows of one order
// are in its partition, in the order of their `seq`, and an ack is given the rows acted on (decision 0137).
expect(yield* orders.fetch('billing', 'invoices', all)).toEqual(billing)
expect(yield* orders.fetch('status', 'dashboard', all)).toEqual(statuses)
yield* orders.ack('status', 'dashboard', statuses.slice(0, 2))
expect(yield* orders.fetch('status', 'dashboard', all)).toEqual(statuses.slice(2))
expect(yield* orders.prune).toBe(2)
expect(yield* orders.published('status', [], all)).toEqual(statuses.slice(2))
expect(yield* orders.published('billing', [], all)).toEqual(billing)
// What each consumer still has to act on, as an operator sees it.
const partition = statuses[0]?.partition ?? 0
const stored = yield* orders.consumers
expect(stored.map(({ since: _, ...consumer }) => consumer)).toEqual([
{ topic: 'billing', consumer: 'invoices', declared: true, behind: 2, cursors: [{ partition, seq: 0 }] },
{
topic: 'status',
consumer: 'dashboard',
declared: true,
behind: 1,
cursors: [{ partition, seq: statuses[1]?.seq ?? 0 }],
},
])
// A consumer's next row is the oldest it has not read: the creation for `invoices`, the charge for `dashboard`,
// each at the time of its step on the runtime's clock (decision 0224).
const [invoices = Number.NaN, dashboard = Number.NaN] = stored.map(({ since }) =>
since === null ? Number.NaN : DateTime.toEpochMillis(since)
)
expect(invoices).toBeLessThanOrEqual(dashboard)
expect(dashboard).toBeLessThanOrEqual(DateTime.toEpochMillis(yield* DateTime.now))
// A consumer that has read everything waits for the next row of its topic instead of polling: the creation of
// another order ends its wait, and a wait with no row would give none after ten seconds (decision 0142).
yield* orders.ack('status', 'dashboard', statuses)
const next = yield* Effect.forkChild(orders.fetch('status', 'dashboard', all, { wait: { millis: 10_000 } }))
yield* orders.create('o-2', { flags: { order: { customer: 'ada', totalInCents: 500 } }, requestId: 'c-o-2', sender: ada })
expect((yield* Fiber.join(next)).map(({ id, value }) => [id, value])).toEqual([['o-2', 'AwaitingPayment']])
}))))
test('a report built from the log counts the orders in each status, as the read model counts them', () =>
database(Effect.scoped(Effect.gen(function*() {
const runtime = yield* launch()
const orders = yield* Effect.provide(runtime.deploy(order, consumers), world)
const report = yield* statusReport
for (const id of ['o-1', 'o-2', 'o-3', 'o-4']) {
yield* orders.create(id, {
flags: { order: { customer: 'ada', totalInCents: 1_000 } },
requestId: `c-${id}`,
sender: ada,
})
}
for (const id of ['o-2', 'o-3']) {
yield* orders.send(id, { _tag: 'PayRequested', payment: 'card-1' }, { requestId: `p-${id}`, sender: ada, keep: 'forever' })
}
yield* orders.send('o-4', { _tag: 'CancelRequested' }, { requestId: 'x-o-4', sender: ada, keep: 'forever' })
yield* runtime.settled
// The consumer: `consume` writes each page of its topic to the store and acks it in one transaction, until a
// page is empty. A consumer that serves a dashboard passes `wait: true` instead, and stops with the runtime
// (decision 0184). The first reader crashes after its writes, as a crash before the commit leaves it: the store
// and the cursors stay as they were, so the next reader reads the page again (decision 0150).
const first = yield* orders.fetch('status', 'dashboard', 3)
const crashed = (page: Parameters<typeof report.record>[0]) => Effect.andThen(report.record(page), Effect.fail('crash'))
expect(yield* Effect.exit(orders.consume('status', 'dashboard', crashed, { limit: 3 }))).toEqual(Exit.fail('crash'))
expect(yield* report.counts).toEqual({})
expect(yield* orders.fetch('status', 'dashboard', 3)).toEqual(first)
yield* orders.consume('status', 'dashboard', report.record, { limit: 3 })
expect(yield* orders.fetch('status', 'dashboard', 3)).toEqual([])
// The store answers the question in one statement, and so does the read model, on one snapshot
// (decisions 0149, 0150).
const counts = yield* report.counts
expect(counts).toEqual({ AwaitingPayment: 1, Paid: 2, Cancelled: 1 })
expect(yield* orders.countBy({ index: 'status', limit: all })).toEqual([
{ value: 'AwaitingPayment', count: 1 },
{ value: 'Cancelled', count: 1 },
{ value: 'Paid', count: 2 },
])
// A limit that is no non-negative integer is the caller's defect, as a query's is (decision 0141).
expect(Exit.isFailure(yield* Effect.exit(orders.countBy({ index: 'status', limit: -1 })))).toBe(true)
}))))
// A consumer read by several readers: each names the partitions it reads, and the cursors stay the consumer's. A
// reader that waits runs as long as the runtime, and records each row once its step commits (decisions 0142, 0184).
test('readers that divide the partitions read each row once, and a reader that waits reads the next', () =>
database(Effect.scoped(Effect.gen(function*() {
const runtime = yield* launch()
const orders = yield* Effect.provide(runtime.deploy(order, consumers), world)
const report = yield* statusReport
for (const id of ['o-1', 'o-2', 'o-3', 'o-4']) {
yield* orders.create(id, {
flags: { order: { customer: 'ada', totalInCents: 1_000 } },
requestId: `c-${id}`,
sender: ada,
})
}
const n = 2
for (let i = 0; i < n; i++) {
const mine = Array.from({ length: orders.partitions }, (_, at) => at + 1).filter((p) => p % n === i)
yield* orders.consume('status', 'dashboard', report.record, { limit: 3, partitions: mine })
}
expect(yield* report.counts).toEqual({ AwaitingPayment: 4 })
expect(yield* orders.fetch('status', 'dashboard', 3)).toEqual([])
// The handler signals inside the reader's transaction. The client of SQLite runs one statement or transaction
// at a time, so the counts read after the signal wait for the commit.
const recorded = yield* Deferred.make<void>()
const record = (page: Parameters<typeof report.record>[0]) =>
Effect.andThen(report.record(page), Deferred.succeed(recorded, undefined))
yield* Effect.forkChild(orders.consume('status', 'dashboard', record, { limit: 3, wait: true }))
yield* orders.send('o-4', { _tag: 'CancelRequested' }, { requestId: 'x-o-4', sender: ada, keep: 'forever' })
yield* Deferred.await(recorded)
expect(yield* report.counts).toEqual({ AwaitingPayment: 3, Cancelled: 1 })
}))))
/**
* The order's scenarios, rehearsed.
*/
import { describe, expect, test } from 'bun:test'
import * as DateTime from 'effect/DateTime'
import * as Effect from 'effect/Effect'
import { accepted, rehearse, rejected } from '@fgaudo/effect-tea-persistent/testing'
import { order, type Sender } from './order.ts'
const placedAt = DateTime.makeUnsafe('2026-10-30T10:00:00Z')
const start = { flags: { order: { customer: 'ada', totalInCents: 4_200 } }, at: placedAt }
const ada: Sender = { _tag: 'Customer', id: 'ada' }
const warehouse: Sender = { _tag: 'Warehouse' }
describe('order', () => {
test('a paid order is shipped', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-1' }, { sender: ada })
yield* rehearsal.answer('charge', { _tag: 'ChargeSucceeded', charge: 'ch-1' })
yield* rehearsal.answer('tell-warehouse', { _tag: 'WarehouseTold' })
expect(yield* rehearsal.send({ _tag: 'ShipmentReported', tracking: 'tr-1' }, { sender: warehouse })).toEqual(accepted)
yield* rehearsal.answer('email-shipment', { _tag: 'ShipmentEmailed' })
expect(yield* rehearsal.state).toMatchObject({ _tag: 'Shipped', email: 'Sent' })
}),
)))
test('a charge declined after the deadline cancels the order at once', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
const deadline = yield* rehearsal.pending('payment-deadline')
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-1' }, { sender: ada })
// The timer left the state with the charge: its firing is dropped.
expect(yield* rehearsal.fire(deadline)).toBe('Dropped')
yield* rehearsal.answer('charge', { _tag: 'ChargeDeclined' })
expect(yield* rehearsal.fire('payment-deadline')).toBe('Delivered')
expect(yield* rehearsal.state).toMatchObject({ _tag: 'Cancelled', reason: 'Unpaid' })
}),
)))
test('the late answer of a first charge does not complete the second', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-1' }, { sender: ada })
const first = yield* rehearsal.pending('charge')
yield* rehearsal.answer(first, { _tag: 'ChargeDeclined' })
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-2' }, { sender: ada })
const second = yield* rehearsal.pending('charge')
expect(second.idempotencyKey).not.toBe(first.idempotencyKey)
expect(yield* rehearsal.answer(first, { _tag: 'ChargeSucceeded', charge: 'ch-1' })).toBe('Dropped')
expect((yield* rehearsal.state)._tag).toBe('Charging')
}),
)))
test('a charge under way when the process dies runs again with the same idempotency key', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-1' }, { sender: ada })
const before = yield* rehearsal.pending('charge')
yield* rehearsal.restart
const after = yield* rehearsal.pending('charge')
expect(after.idempotencyKey).toBe(before.idempotencyKey)
// The payment provider is given the key's digest, which fits the length it caps a key at.
expect(after.idempotencyKey).toMatch(/^[A-Za-z0-9_-]{43}$/)
}),
)))
test('a shipment email under way when the process dies is not sent again', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-1' }, { sender: ada })
yield* rehearsal.answer('charge', { _tag: 'ChargeSucceeded', charge: 'ch-1' })
yield* rehearsal.answer('tell-warehouse', { _tag: 'WarehouseTold' })
yield* rehearsal.send({ _tag: 'ShipmentReported', tracking: 'tr-1' }, { sender: warehouse })
yield* rehearsal.restart
expect(yield* rehearsal.state).toMatchObject({ _tag: 'Shipped', email: 'Unsure' })
}),
)))
test('a cancellation before the shipment refunds, and the warehouse does not ship', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-1' }, { sender: ada })
yield* rehearsal.answer('charge', { _tag: 'ChargeSucceeded', charge: 'ch-1' })
yield* rehearsal.answer('tell-warehouse', { _tag: 'WarehouseTold' })
expect(yield* rehearsal.send({ _tag: 'CancelRequested' }, { sender: ada })).toEqual(accepted)
// The warehouse hands the parcel over only on an accepted report.
expect(yield* rehearsal.send({ _tag: 'ShipmentReported', tracking: 'tr-1' }, { sender: warehouse }))
.toMatchObject(rejected)
yield* rehearsal.answer('refund', { _tag: 'RefundCompleted' })
expect(yield* rehearsal.state).toMatchObject({ _tag: 'Cancelled', reason: 'Refunded' })
}),
)))
test('a cancellation is rejected while a charge is under way', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
yield* rehearsal.send({ _tag: 'PayRequested', payment: 'card-1' }, { sender: ada })
expect((yield* rehearsal.projection).canCancel).toBe(false)
expect(yield* rehearsal.send({ _tag: 'CancelRequested' }, { sender: ada })).toMatchObject(rejected)
}),
)))
test('only the customer cancels', () =>
Effect.runPromise(rehearse(
order,
start,
Effect.fnUntraced(function*(rehearsal) {
expect(yield* rehearsal.send({ _tag: 'CancelRequested' }, { sender: { _tag: 'Customer', id: 'eve' } }))
.toMatchObject(rejected)
expect(yield* rehearsal.send({ _tag: 'CancelRequested' }, { sender: warehouse })).toMatchObject(rejected)
}),
)))
})