better-effect-mq-redis

better-effect-mq-redis is a storage extension for better-effect-mq. It provides durable queue state, optional events, schedules, and flow persistence while the core package keeps ownership of Jobs, handlers, retries, and Workers.

Install

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

The redis peer is needed for config-based factories. If your application already owns compatible clients, pass them to the Layer and keep their lifecycle at the application boundary.

Quick start: the canonical MQ shape

This example uses the adapter-created connections. The queue descriptor, handler, Worker Service, Layer composition, and Runtime are the same as the core MQ guide.

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 { RedisJobStore } from 'better-effect-mq-redis'
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 = RedisJobStore.layerFromConfig({
  url: process.env.REDIS_URL,
  namespace: 'emails',
  prefix: 'better-effect-mq'
})
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()
}

layerFromConfig owns the command and subscriber connections. For caller-owned clients, use RedisJobStore.layer or RedisJobStore.layerWithEvents and close those clients outside the Runtime. Redis notifications are wake hints; bounded polling remains the recovery path. This quick start 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

Use layerWithEventsFromConfig when queue transitions need a bounded event feed. Schedule and flow providers follow the same namespace and named-token conventions as JobStore. Redis and Valkey delivery is at least once, so configure persistence, replication, memory limits, and idempotent handlers for the failure modes of the service you operate.

Redis-native transactional outbox

When the domain write is also a Redis write, use RedisOutbox.transaction. Prepare the request and record first; the adapter creates the MULTI context, appends the record automatically, executes the commands, and discards the context when the callback fails:

import { Effect } from 'better-effect'
import { OutboxId, makeOutboxRecord } from 'better-effect-mq-outbox'
import { RedisClient, RedisOutbox } from 'better-effect-mq-redis'
import { Result } from 'better-result'

const redisUrl = process.env.REDIS_URL
if (redisUrl === undefined) throw new Error('REDIS_URL is required')
const redis = await RedisClient.fromConfig({ url: redisUrl, namespace: 'orders' })
await redis.initialize()
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 RedisOutbox.transaction(redis, record.value, (transaction) => {
  transaction.sendCommand(['HSET', 'orders:message-123', 'recipient', 'ada@example.test'])
  return Result.ok(undefined)
})
if (Result.isError(committed)) throw committed.error

This boundary is atomic only for Redis-native writes in the same namespace. It cannot include a PostgreSQL, MySQL, SQLite, MongoDB, or remote-service write in MULTI/EXEC; use that database's transaction helper for a database-owned domain write. After commit, compose Continue with the SendEmail, runtime, and WorkerLive values from the quick start. Compose RedisOutboxStore.layer* with a Runtime-owned publisher and that Worker:

import { JobStore } from 'better-effect-mq'
import { OutboxPublisher, OutboxRoutes, OutboxStore } from 'better-effect-mq-outbox'
import { RedisOutboxStore } from 'better-effect-mq-redis'

const Routes = OutboxRoutes.make({ emails: JobStore })
const Publisher = OutboxPublisher.service('@app/EmailOutboxPublisher')
const RedisOutboxLive = RedisOutboxStore.layerFromConfig({ url: redisUrl, namespace: 'orders' })
const PublisherLive = Publisher.layer(() => ({
  outboxes: [OutboxStore] as const,
  routes: Routes,
  concurrency: 2,
  pollIntervalMs: 500
}))
// Continue with `WorkerLive`, `redisUrl`, and `runtime` from the earlier
// sections. Add PublisherLive and RedisOutboxLive 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 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 and route setup.

better-effect-mq-redis

Install

Quick start: the canonical MQ shape

Optional capabilities

Redis-native transactional outbox

On this page