Getting started

Install the package:

bun add effect-mq        # or npm install / pnpm add / yarn add

effect-mq targets Effect v4 (effect@4.0.0-rc). The Postgres and Redis stores have optional peer dependencies; you'll add those in the storage step.

1. Define a job

A job is a class: a tag, a payload schema, and defaults. Producers and runners share this one definition.

import { Job } from "effect-mq"
import { Schema } from "effect"

class SendEmail extends Job.make("SendEmail", {
  payload: { to: Schema.String, subject: Schema.String },
  success: Schema.String,                            // typed result
  idempotencyKey: ({ to, subject }) => `${to}:${subject}`,
  metadata: ({ to }) => ({ to }),                    // queryable context
  queue: "email",
  defaults: {
    attempts: 3,
    backoff: { type: "exponential", delay: "1 second" }
  }
}) {}

Payloads are real schemas. Redacted, DateTimeUtc, branded types, and unions round-trip through storage and come back typed in the handler.

2. Enqueue from anywhere

Producers need the job's store in context, never the worker. Your API server can enqueue without depending on any handler code.

import { Effect } from "effect"

const program = Effect.gen(function*() {
  // fire and forget: returns the JobId
  const jobId = yield* SendEmail.enqueue({ to: "ada@example.com", subject: "hi" })

  // ...or enqueue and await the typed result
  const messageId = yield* SendEmail.execute(
    { to: "grace@example.com", subject: "now" },
    { delay: "5 seconds", priority: 2 }
  )
})

Thanks to idempotencyKey, enqueueing the same email twice is a no-op that returns the existing id.

3. Run a worker

Handlers are Effects; workers are Layers. Compose them wherever you deploy: same process or a dedicated machine.

import { Worker, MemoryJobStore } from "effect-mq"
import { Effect, Layer } from "effect"

const RunnerLive = SendEmail.toLayer(
  (payload) => Effect.map(Worker.CurrentJob, ({ jobId }) => `message-${jobId}`),
  { concurrency: 5 }
).pipe(
  Layer.provideMerge(Worker.layer()),
  Layer.provideMerge(MemoryJobStore.layer)   // swap for Postgres/Redis below
)

Provide RunnerLive to your app and the worker claims, runs, retries, and records jobs. Interrupting the layer's scope triggers a graceful shutdown: the worker releases in-flight jobs back to waiting without consuming an attempt.

4. Pick real storage

MemoryJobStore is for tests and demos. For production, swap the layer; nothing else changes:

Postgres (drizzle)

import { DrizzleJobStore, mqJobs, mqJobAttempts, mqSchedules, mqQueueControl, mqDedupe } from "effect-mq/drizzle-postgres"

// The tables live in YOUR drizzle schema; drizzle-kit owns migrations.
export const jobs = mqJobs()
export const jobAttempts = mqJobAttempts(jobs)
export const schedules = mqSchedules()
export const queues = mqQueueControl()
export const dedupe = mqDedupe()

const StoreLive = DrizzleJobStore.layer({ jobs, attempts: jobAttempts, schedules, queues, dedupe })

Redis

import { RedisJobStore } from "effect-mq/redis"
import { NodeRedis } from "@effect/platform-node"

const StoreLive = RedisJobStore.layer().pipe(
  Layer.provide(NodeRedis.layer({ url: "redis://localhost:6379" }))
)

See Postgres for the migration workflow and Redis for client setup.

5. Check on your jobs

const status = yield* SendEmail.poll(jobId)        // Option<JobStatus>
const runs = yield* SendEmail.attempts(jobId)      // the decoded run ledger

Every attempt (success, retry, failure, stall) is persisted and decodes back to a typed exit.

Where to next

• Defining jobs: everything Job.make accepts.
• Enqueueing: delays, absolute times, priorities, batches.
• Workers & handlers: concurrency, locks, shutdown.
• Testing your app: assert what your services enqueue, without a worker.