better-effect-mq-mysql

better-effect-mq-mysql is the MySQL extension for better-effect-mq. It adds durable storage while leaving the application-facing queue, Job, Worker, Layer, and Runtime model intact.

Install and migrate

bun add better-effect-mq-mysql better-effect-mq better-effect better-result better-effect-schema zod mysql2

Use MySQL 8.0.16 or newer with InnoDB and strict SQL mode. Run the migrator before workers start; the storage Layer validates the existing layout but does not run migrations for you.

import { createPool } from 'mysql2/promise'
import { MySqlMigrator } from 'better-effect-mq-mysql'

const pool = createPool(process.env.MYSQL_URL ?? '')
await MySqlMigrator.run(pool)
await MySqlMigrator.validate(pool)

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 { MySqlJobStore } from 'better-effect-mq-mysql'
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 = MySqlJobStore.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 example borrows the pool and closes it after Runtime disposal. Use layerFromConfig when the adapter should own a mysql2/promise pool. Schedules, events, flows, and outbox storage use the same namespace and Layer-first composition. MySQL delivery is at least once; design handlers and downstream effects for safe repetition. 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

Use MySqlOutbox.transaction when a MySQL domain write and a prepared Job request must commit together. Prepare the request and record first; the adapter then owns the connection, transaction, append, and cleanup. The callback performs only the domain write:

import { Effect } from 'better-effect'
import { OutboxId, makeOutboxRecord } from 'better-effect-mq-outbox'
import { MySqlOutbox } from 'better-effect-mq-mysql'
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 MySqlOutbox.transaction(
  pool,
  record.value,
  async (connection) => {
    await connection.query('INSERT INTO emails (id, recipient) VALUES (?, ?)', [
      'message-123',
      'ada@example.test'
    ])
    return Result.ok(undefined)
  },
  { namespace: 'emails' }
)
if (Result.isError(committed)) throw committed.error

The adapter appends the supplied record only after the callback succeeds and commits both writes as one MySQL transaction. A thrown, rejected, or nominal Result.err callback rolls back the domain write and append. 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 { MySqlOutboxStore } from 'better-effect-mq-mysql'

const Routes = OutboxRoutes.make({ emails: JobStore })
const Publisher = OutboxPublisher.service('@app/EmailOutboxPublisher')
const MySqlOutboxLive = MySqlOutboxStore.layer({ pool, 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 MySqlOutboxLive 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. See the outbox transaction guide for the complete order-to-confirmation workflow.

better-effect-mq-mysql

Install and migrate

Quick start: the canonical MQ shape

Transactional outbox

On this page