better-effect-mq-postgres

better-effect-mq-postgres is an extension of better-effect-mq. The core package owns queues, Jobs, handlers, retries, and worker supervision; this adapter supplies a durable PostgreSQL JobStore through a Layer. Your application code keeps the same queue descriptor and Worker APIs used with the in-memory store.

Install and migrate

bun add better-effect-mq-postgres better-effect-mq better-effect better-result better-effect-schema zod pg

Run migrations as an explicit deployment step. Store acquisition validates the schema but does not silently change it.

import { Pool } from 'pg'
import { PostgresMigrator } from 'better-effect-mq-postgres'

const pool = new Pool({ connectionString: process.env.DATABASE_URL })
await PostgresMigrator.run(pool, { schema: 'public' })
await PostgresMigrator.validate(pool, { schema: 'public' })

Quick start: the canonical MQ shape

The descriptor and handler are storage-neutral. Only PostgresJobStore.layer changes when you replace another adapter.

import * as z from 'zod'
import { Effect, Layer, Runtime } from 'better-effect'
import { ClockLive } from 'better-effect/standard-services'
import { Codec, JobContext, JobEncodeFailure, Queue, Worker } from 'better-effect-mq'
import { PostgresJobStore } from 'better-effect-mq-postgres'
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() })
})

const EmailWorker = Worker.service('@app/EmailWorker')
const emailHandler = Worker.handle(SendEmail, (payload) =>
  Effect.fn(async function* () {
    const context = yield* JobContext
    // Use the message ID as the downstream provider's idempotency key.
    return Result.ok(`sent:${payload.recipient}:${context.attempt}`)
  })
)

const StorageLive = PostgresJobStore.layer({
  pool,
  namespace: 'emails'
})
const WorkerLive = EmailWorker.layer(() => ({
  handlers: [emailHandler] as const,
  concurrency: 4,
  pollIntervalMs: 500
}))
const AppLive = Layer.complete(Layer.merge(StorageLive, Layer.merge(ClockLive, WorkerLive)))

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
  console.log(`enqueued ${result.value}`)
} finally {
  await runtime.dispose()
  await pool.end()
}

The pool is borrowed by PostgresJobStore.layer; the application closes it after the Runtime. Use PostgresJobStore.layerFromConfig when the adapter should create and release the pool itself. The payload codec uses a Zod 4 schema-backed class from better-effect-schema through Codec.standardSchema. Its explicit CoreSchema.encode callback maps the class value to JSON for persistence.

Optional capabilities

Add event, schedule, flow, or outbox Layers only when the application needs them. Use the same PostgreSQL pool, schema, and namespace for providers that share a queue. For a transactional handoff from a domain write to a Job, prepare the record before entering the transaction and let PostgresOutbox.transaction own the client and lifecycle:

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

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

The adapter appends the prepared outbox record after the callback succeeds, commits only after that append succeeds, rolls back on thrown, rejected, or nominal Result.err failures, and always releases the client. Neither helper turns a remote publish into a distributed transaction. After commit, compose the Runtime-owned publisher beside the existing WorkerLive and the record/SendConfirmation values prepared in the outbox transaction guide:

import { JobStore } from 'better-effect-mq'
import { OutboxPublisher, OutboxRoutes } from 'better-effect-mq-outbox'
import { PostgresOutbox } from 'better-effect-mq-postgres'

const Routes = OutboxRoutes.make({ orders: JobStore })
const Publisher = OutboxPublisher.service('@app/OrderOutboxPublisher')
const PostgresOutboxLive = PostgresOutbox.layer({ pool, namespace: 'orders' })
const PublisherLive = Publisher.layer(() => ({
  outboxes: [PostgresOutbox] as const,
  routes: Routes,
  concurrency: 2,
  pollIntervalMs: 500
}))
// Continue with the `runtime` and `WorkerLive` from the quick start. Add
// PublisherLive and PostgresOutboxLive to
// the same Runtime as the existing JobStore and Worker layers.
const delivered = await runtime.run(() =>
  SendConfirmation.awaitResult('order-confirmation:order-123')
)

OutboxRoutes stores the JobStore Service token, never a store instance. A crash after enqueue and before settlement can replay the request; use the deterministic Job ID as the downstream idempotency key. See the outbox transaction guide for the complete order-to-confirmation workflow, including request preparation and record creation.

Advanced: caller-owned transactions

PostgresOutbox.appendIn remains available when an application already owns a compatible transaction; it never owns that client's lifecycle. The normal path is PostgresOutbox.transaction.

PostgreSQL delivery is at least once. Make handlers and external side effects idempotent, monitor leases and retry age, and keep polling enabled even when database notifications provide a wake hint.

better-effect-mq-postgres

Install and migrate

Quick start: the canonical MQ shape

Optional capabilities

Advanced: caller-owned transactions

On this page