better-effect-mq-sqlite

better-effect-mq-sqlite is the embedded storage extension for better-effect-mq. It is a good fit for one application or machine, local workers, command-line tools, and durable test fixtures. It keeps the same Queue, Job, Worker, Layer, and Runtime APIs as every other adapter.

Install and migrate

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

Bun applications can use the host binding without another SQLite driver:

import { openSqlite } from 'better-effect-mq-sqlite/bun'
import { SqliteMigrator } from 'better-effect-mq-sqlite'

const database = openSqlite('./data/jobs.sqlite')
SqliteMigrator.migrate({ database })

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 { SqliteJobStore } from 'better-effect-mq-sqlite'
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 = SqliteJobStore.layer({ database, namespace: 'emails' })
const WorkerLive = EmailWorker.layer(() => ({
  handlers: [emailHandler] as const,
  concurrency: 1,
  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()
  database.close()
}

The database is caller-owned in this form. Use layerFromFile when the adapter should open and close the file with the Runtime. SQLite serializes writes, so use it for local or single-node workloads; choose a distributed adapter when several hosts must share one queue. Schedules, events, flows, and outbox providers are optional Layers over the same database. 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.

Transactional outbox

The SQLite outbox adapter owns the serialized transaction lifecycle for a domain write and its outbox record. Prepare the request and record first, then let SqliteOutboxTransactions.transaction append the record after the callback succeeds:

import * as z from 'zod'
import { Database } from 'bun:sqlite'
import { Effect, Layer, Runtime } from 'better-effect'
import { ClockLive } from 'better-effect/standard-services'
import { Codec, JobEncodeFailure, JobStore, Queue, Worker } from 'better-effect-mq'
import {
  OutboxId,
  OutboxPublisher,
  OutboxRoutes,
  OutboxStore,
  makeOutboxRecord
} from 'better-effect-mq-outbox'
import {
  SqliteJobStore,
  SqliteMigrator,
  SqliteOutboxStore,
  SqliteOutboxTransactions
} from 'better-effect-mq-sqlite'
import { Schema as CoreSchema } from 'better-effect-schema'
import { Schema } from 'better-effect-schema/zod'
import { Result } from 'better-result'

const Orders = Queue.define('orders')
const SendConfirmationPayloadSchema = z.object({
  orderId: z.string().min(1),
  email: z.email()
})
class SendConfirmationPayload extends Schema.Class<SendConfirmationPayload>(
  'app/SendConfirmationPayload'
)(SendConfirmationPayloadSchema) {}
const SendConfirmationPayloadCodec = Codec.standardSchema({
  schema: SendConfirmationPayload,
  encode: (value) =>
    CoreSchema.encode(SendConfirmationPayload, value).mapError(
      (error) => new JobEncodeFailure({ message: error.message, code: 'schema-encode' })
    )
})
const SendConfirmation = Orders.job('send-confirmation', {
  version: 1,
  payload: SendConfirmationPayloadCodec,
  result: Codec.standardSchema({ schema: z.string() })
})
const ConfirmationWorker = Worker.service('@app/ConfirmationWorker')
const confirmationHandler = Worker.handle(SendConfirmation, (payload) =>
  Effect.fn(async function* () {
    return Result.ok(`sent:${payload.email}`)
  })
)
const ConfirmationWorkerLive = ConfirmationWorker.layer(() => ({
  handlers: [confirmationHandler] as const,
  concurrency: 1,
  pollIntervalMs: 100
}))
const Routes = OutboxRoutes.make({ jobs: JobStore })
const Publisher = OutboxPublisher.service('@app/OutboxPublisher')
const PublisherLive = Publisher.layer(() => ({
  outboxes: [OutboxStore] as const,
  routes: Routes,
  concurrency: 1,
  leaseDurationMs: 30_000,
  heartbeatIntervalMs: 10_000,
  pollIntervalMs: 100
}))

const database = new Database('./data/orders.sqlite')
SqliteMigrator.migrate({ database })
database.exec('CREATE TABLE IF NOT EXISTS orders (id TEXT PRIMARY KEY, email TEXT NOT NULL)')
const AppLive = Layer.complete(
  Layer.merge(
    Layer.merge(
      SqliteJobStore.layer({ database, namespace: 'orders' }),
      SqliteOutboxStore.layer({ database, namespace: 'orders' })
    ),
    Layer.merge(ClockLive, Layer.merge(ConfirmationWorkerLive, PublisherLive))
  )
)
const runtime = await Runtime.make(AppLive)

const prepared = await runtime.run(() =>
  Effect.gen(async function* () {
    const request = yield* SendConfirmation.prepare(
      { orderId: 'order-1', email: 'ada@example.test' },
      { jobId: 'order-confirmation:order-1' }
    )
    return Result.ok(request)
  })
)
if (Result.isError(prepared)) throw prepared.error

const record = makeOutboxRecord({
  id: OutboxId.make('order-confirmation:order-1').unwrap(),
  target: 'jobs',
  request: prepared.value,
  attemptsMax: 5
})
if (Result.isError(record)) throw record.error

const committed = await SqliteOutboxTransactions.transaction(
  database,
  record.value,
  (transaction) => {
    transaction
      .prepare('INSERT INTO orders(id, email) VALUES (?, ?)')
      .run('order-1', 'ada@example.test')
    return Result.ok('order-1')
  }
)
if (Result.isError(committed)) throw committed.error

await runtime.dispose()
database.close()

The callback performs only the domain write. The adapter appends the prepared record automatically, commits both operations together, and rolls back on a nominal Result.err, thrown callback, rejection, append conflict, or validation failure. Publishing remains a separate at-least-once step; see the outbox transaction guide for route composition.

Advanced: caller-owned write contexts

Use SqliteOutboxTransactions.appendIn only in an integration that already owns a compatible write context. The normal application path is SqliteOutboxTransactions.transaction.

better-effect-mq-sqlite

Install and migrate

Quick start: the canonical MQ shape

Transactional outbox

Advanced: caller-owned write contexts

On this page