better-effect-mq-mongodb

better-effect-mq-mongodb is the MongoDB extension for better-effect-mq. It supplies the persistent JobStore and optional extension stores; Queue.define, Job descriptors, Worker.handle, Worker.service, and Runtime composition remain storage-neutral.

Install and prepare MongoDB

bun add better-effect-mq-mongodb better-effect-mq better-effect better-result better-effect-schema zod mongodb

Use a replica set or a compatible mongos deployment because queue state uses transactions. Apply the layout explicitly before acquiring the application Layer.

import { MongoClient } from 'mongodb'
import { MongoJobStore } from 'better-effect-mq-mongodb'

const client = new MongoClient(
  process.env.MONGODB_URI ?? 'mongodb://localhost:27017/?replicaSet=rs0'
)
await client.connect()
const db = client.db('application')
await MongoJobStore.migrate({ db })

Quick start: the canonical MQ shape

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 { MongoJobStore } from 'better-effect-mq-mongodb'
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
    return Result.ok(`sent:${payload.recipient}:${context.attempt}`)
  })
)

const StorageLive = MongoJobStore.layer({ db, 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 client.close()
}

The supplied Db and its MongoClient are caller-owned. Use MongoJobStore.layerFromConfig when the adapter should create and release the client with the Runtime. Add the schedule, event, flow, or outbox Layer only when needed, keeping the same database and namespace.

MongoDB change streams can reduce wake latency but are not the source of truth. Keep polling, backups, transaction limits, and document size limits in your operational plan. Delivery remains at least once, so external effects must tolerate retries. The quick-start payload 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.

Transactional outbox

Use MongoOutbox.transaction when a MongoDB domain write and a prepared Job request must commit together. Prepare the request and OutboxRecord before entering the helper; the adapter owns the session, transaction, automatic append, and cleanup:

import { Effect } from 'better-effect'
import { OutboxId, makeOutboxRecord } from 'better-effect-mq-outbox'
import { MongoOutbox } from 'better-effect-mq-mongodb'
import { Result } from 'better-result'

const prepared = await runtime.run(() =>
  Effect.gen(async function* () {
    return Result.ok(
      yield* SendEmail.prepare(
        { messageId: 'message-123', recipient: 'ada@example.test' },
        { jobId: 'email:message-123' }
      )
    )
  })
)
if (Result.isError(prepared)) throw prepared.error

const record = makeOutboxRecord({
  id: OutboxId.make('outbox:message-123').unwrap(),
  target: 'emails',
  request: prepared.value,
  attemptsMax: 5
})
if (Result.isError(record)) throw record.error

const committed = await MongoOutbox.transaction(
  db,
  record.value,
  async (session) => {
    await db
      .collection('emails')
      .insertOne({ _id: 'message-123', recipient: 'ada@example.test' }, { session })
    return Result.ok(undefined)
  },
  { namespace: 'emails' }
)
if (Result.isError(committed)) throw committed.error

The adapter appends the supplied record after the callback succeeds and commits both writes in the same MongoDB transaction. A thrown, rejected, or nominal Result.err callback aborts the work; MongoDB may retry a transaction callback, so domain writes must be safe to retry. After commit, continue with the SendEmail, runtime, and WorkerLive values from the quick start and compose the Runtime-owned publisher beside that Worker. It routes to the JobStore Service token and the Worker delivers the Job:

import { JobStore } from 'better-effect-mq'
import { OutboxPublisher, OutboxRoutes, OutboxStore } from 'better-effect-mq-outbox'
import { MongoOutboxStore } from 'better-effect-mq-mongodb'

const Routes = OutboxRoutes.make({ emails: JobStore })
const Publisher = OutboxPublisher.service('@app/EmailOutboxPublisher')
const MongoOutboxLive = MongoOutboxStore.layer({ db, namespace: 'emails' })
const PublisherLive = Publisher.layer(() => ({
  outboxes: [OutboxStore] as const,
  routes: Routes,
  concurrency: 2,
  pollIntervalMs: 500
}))
// Continue with `WorkerLive` and `runtime` from the quick start. Add this
// PublisherLive and MongoOutboxLive to
// the same Runtime as the existing JobStore and Worker layers.
const delivered = await runtime.run(() => SendEmail.awaitResult('email:message-123'))

OutboxRoutes stores the JobStore token, never a store instance. Delivery is post-commit and at least once: a crash after enqueue and before settlement can replay the same request, so keep the deterministic Job ID as the downstream idempotency key. If the first argument is a MongoClient instead of db, pass the database as options.db. See the outbox transaction guide for the complete order-to-confirmation workflow.

better-effect-mq-mongodb

Install and prepare MongoDB

Quick start: the canonical MQ shape

Transactional outbox

On this page