better-effect-mq-outbox

Publish prepared queue requests with explicit routes and at-least-once recovery.

better-effect-mq-outbox is the storage-neutral outbox package for better-effect-mq. It defines immutable OutboxRecord values, an OutboxStore contract, explicit target routes and a Layer-owned publisher. Database adapters remain separate packages and own their transaction boundaries. In the concrete order-to-confirmation workflow below, the order row and the confirmation request become visible together, then a Runtime-owned publisher and Worker deliver the confirmation after commit.

Without an outbox, an application can commit an order and crash before enqueueing its confirmation, or enqueue first and then roll back the order. Those two writes have no shared boundary. The outbox stores the prepared request beside the order so either both are durable or neither is; it does not put an email provider or other remote side effect inside the database transaction. Use a provider-backed Zod 4 schema through Codec.standardSchema for the Job payload; the better-effect-schema guide covers schema-backed classes and normalized failures. The trusted JSON escape hatch is for trusted, already-JSON-safe values and is not the default boundary example.

Quick start: the canonical MQ shape

The outbox is an extension around the same queue and worker model. Keep the Queue and Job descriptor, Worker.handle, Worker.service, Layer composition, and Runtime boundary identical to the MQ quick start. Only the transactional append and publisher are new:

import * as z from 'zod'
import { Effect, Layer, Runtime } from 'better-effect'
import { ClockLive } from 'better-effect/standard-services'
import {
  Codec,
  JobContext,
  JobEncodeFailure,
  JobStore,
  MemoryJobStore,
  Queue,
  Worker
} from 'better-effect-mq'
import {
  MemoryOutboxStore,
  OutboxPublisher,
  OutboxRoutes,
  OutboxStore
} from 'better-effect-mq-outbox'
import { Schema as CoreSchema } from 'better-effect-schema'
import { Schema } from 'better-effect-schema/zod'
import { Result } from 'better-result'

const Emails = Queue.define('emails')
const SendEmailPayloadSchema = z.object({
  messageId: z.string().min(1),
  recipient: z.email()
})
class SendEmailPayload extends Schema.Class<SendEmailPayload>('app/SendEmailPayload')(
  SendEmailPayloadSchema
) {}
const SendEmailPayloadCodec = Codec.standardSchema({
  schema: SendEmailPayload,
  encode: (value) =>
    CoreSchema.encode(SendEmailPayload, value).mapError(
      (error) => new JobEncodeFailure({ message: error.message, code: 'schema-encode' })
    )
})
const SendEmail = Emails.job('send-email', {
  version: 1,
  payload: SendEmailPayloadCodec,
  result: Codec.standardSchema({ schema: z.string() }),
  idempotencyKey: ({ messageId }) => `confirmation:${messageId}`
})

const EmailWorker = Worker.service('@app/EmailWorker')
const emailHandler = Worker.handle(SendEmail, (payload) =>
  Effect.fn(async function* () {
    const context = yield* JobContext
    return Result.ok(`sent:${payload.recipient}:${context.attempt}`)
  })
)

const BillingOutbox = OutboxStore
const Routes = OutboxRoutes.make({ jobs: JobStore })
const Publisher = OutboxPublisher.service('@app/OutboxPublisher')
const StorageLive = Layer.merge(
  Layer.succeed(JobStore, JobStore.of(MemoryJobStore.make())),
  Layer.succeed(BillingOutbox, OutboxStore.of(MemoryOutboxStore.make()))
)
const WorkerLive = EmailWorker.layer(() => ({
  handlers: [emailHandler] as const,
  concurrency: 1,
  pollIntervalMs: 500
}))
const PublisherLive = Publisher.layer(() => ({
  outboxes: [BillingOutbox] as const,
  routes: Routes,
  concurrency: 1,
  leaseDurationMs: 30_000,
  heartbeatIntervalMs: 10_000,
  pollIntervalMs: 1_000
}))
const AppLive = Layer.complete(
  Layer.merge(StorageLive, Layer.merge(ClockLive, Layer.merge(WorkerLive, PublisherLive)))
)

const runtime = await Runtime.make(AppLive)
try {
  const result = await runtime.run(() =>
    Effect.gen(async function* () {
      const jobId = yield* SendEmail.enqueue({
        messageId: 'message-123',
        recipient: 'ada@example.test'
      })
      return Result.ok(jobId)
    })
  )
  if (Result.isError(result)) throw result.error
} finally {
  await runtime.dispose()
}

The example uses in-memory stores only to show the publisher and Worker composition. In production, replace the two storage providers with a database adapter and use that adapter's transaction helper as shown below. OutboxRoutes.make({ jobs: JobStore }) maps the persisted target to the JobStore Service token; it never captures a store instance.

Prepare before the transaction

An outbox record stores a validated PreparedEnqueue, not a codec, callback, Service, Runtime, connection or transaction object. The application prepares the request before entering its domain transaction; the concrete adapter owns the transaction boundary and appends the record after the domain callback succeeds.

import { Effect, Runtime } from 'better-effect'
import { JobStore } from 'better-effect-mq'
import { OutboxId, OutboxRoutes, OutboxStore, makeOutboxRecord } from 'better-effect-mq-outbox'
import { Result } from 'better-result'

const Routes = OutboxRoutes.make({ jobs: JobStore })

const preparedResult = await runtime.run(() =>
  Effect.gen(async function* () {
    const prepared = yield* SendEmail.prepare(
      {
        messageId: 'order-123',
        recipient: 'ada@example.test'
      },
      { jobId: 'order-confirmation:order-123' }
    )
    return Result.ok(prepared)
  })
)

if (Result.isError(preparedResult)) throw preparedResult.error

const record = makeOutboxRecord({
  id: OutboxId.make('order-confirmation:order-123').unwrap(),
  target: 'jobs',
  request: preparedResult.value
})

if (Result.isError(record)) throw record.error

The record and the domain write commit or roll back together when the selected database adapter appends them through its real transaction. After commit, the Runtime-owned publisher resolves the route to the JobStore token, enqueues the prepared request, and the Runtime-owned Worker claims and handles the Job.

Transaction to publisher

When PostgreSQL is the source of truth, let the adapter own the concrete transaction boundary. The callback performs the domain write; the adapter appends the prepared record after the callback succeeds, commits both writes, and always cleans up the client:

import type { Pool } from 'pg'
import { PostgresOutbox } from 'better-effect-mq-postgres'
import { Result } from 'better-result'

declare const pool: Pool

const persisted = await PostgresOutbox.transaction(
  pool,
  record.value,
  async (transaction) => {
    await transaction.query('INSERT INTO orders (id, email, status) VALUES ($1, $2, $3)', [
      'order-123',
      'ada@example.test',
      'created'
    ])
    return Result.ok(undefined)
  },
  { namespace: 'orders' }
)
if (Result.isError(persisted)) throw persisted.error

After the adapter commits, PublisherLive claims the durable record, resolves the jobs route, and enqueues the prepared SendEmail request in the core JobStore. The same Runtime-owned Worker then claims that Job and runs the Worker.handle callback; the outbox record is marked published only after the enqueue succeeds.

Without the outbox, committing the order and enqueueing the email are two unrelated writes: a crash between them can leave a saved order with no email, or enqueue an email for an order whose transaction later rolled back. The record makes the handoff durable and replayable; it does not make a remote email provider part of the database transaction, so the handler still needs an idempotency key.

Layer-owned publisher

Publishers participate in the same Runtime root as Workers and application resources:

import { Layer } from 'better-effect'
import { OutboxPublisher } from 'better-effect-mq-outbox'

const Publisher = OutboxPublisher.service('@app/OutboxPublisher')
const PublisherLive = Publisher.layer(() => ({
  outboxes: [BillingOutbox] as const,
  routes: Routes,
  concurrency: 4,
  leaseDurationMs: 30_000,
  heartbeatIntervalMs: 10_000,
  pollIntervalMs: 1_000
}))

const ApplicationLive = Layer.complete(
  Layer.merge(StorageLive, Layer.merge(ClockLive, Layer.merge(WorkerLive, PublisherLive)))
)

The publisher stops claiming during Runtime quiesce. Work already admitted can finish its heartbeat, enqueue and settlement before the publisher is released. There is no detached loop and no second Runtime.

Delivery and adapters

Outbox delivery is at-least-once. A crash after the publisher enqueues the Job but before it settles the outbox record can redeliver the same record. The deterministic Job ID/idempotency key makes enqueue replay converge, while the confirmation handler must also use that key with the email provider or its own sent-message table. The pipeline does not promise exactly-once execution.

PostgreSQL, Redis/Valkey, MongoDB, SQLite, and MySQL provide concrete storage adapters. Each guide documents the adapter's migrations, transaction boundary and client ownership. The better-effect-mq-outbox/testing contract suite is runner-neutral and does not need a database or Runtime.