oban-ts

Processamento de jobs e transactional outbox para TypeScript, sobre PostgreSQL. É um port do Oban (Elixir) com a mesma arquitetura. O papel do Ecto fica com o TypeORM, e há adaptadores para Knex, Drizzle, Prisma e pg puro.

await dataSource.transaction(async manager => {
	const order = await manager.save(Order, { email });
	// Commit ou rollback juntos: não existe pedido sem e-mail nem e-mail sem pedido.
	await oban.insert(SendReceipt.new({ orderId: order.id }), { tx: manager });
});

O job é uma linha na mesma transação da escrita de negócio. Só depois do commit ele fica visível para os workers, que o executam at-least-once, com retry e backoff.

Documentação, exemplos e skills

• Documentação completa: conceitos, instalação por ORM, workers, outbox, frameworks (NestJS, Express, Next.js), testes, produção, troubleshooting e internals.
• Exemplos executáveis: um por ORM, Express, NestJS, multi-serviço, idempotência, agendamento, cron, fan-out, telemetria e testes.
• Skills para agentes de IA: plugin do Claude Code para instalar o oban-ts em qualquer projeto, refatorar para outbox, escrever workers, testar e depurar.

Mapa Oban → oban-ts

| Oban (Elixir) | oban-ts | | ---------------------------------- | ------------------------------------------------ | | Ecto.Repo | Repo: typeormRepo, knexRepo, drizzleRepo, prismaRepo, pgRepo | | Oban.Migration.up(version: 1) | Migration.up(repo, { version: 1 }) / obanMigration() (TypeORM) / Migration.sql() | | use Oban.Worker, queue: :mailers | defineWorker({ name, queue, perform }) | | MyWorker.new(args) | MyWorker.new(args) | | Oban.Job.new(args, worker: ...) | Job.new(args, { worker }) | | Oban.insert/2, insert_all/2 | oban.insert(cs, { tx }), oban.insertAll(cs, { tx }) | | {:ok, _}, {:error, _}, {:cancel, _}, {:snooze, s} | ok(), error() ou throw, cancel(), snooze(s) | | Oban.Engines.Basic / Inline | BasicEngine / InlineEngine | | Oban.Notifiers.Postgres / Isolated | PostgresNotifier (LISTEN/NOTIFY) / IsolatedNotifier | | Oban.Peers.Postgres / Isolated | PostgresPeer (tabela oban_peers) / IsolatedPeer | | Oban.Stager, Pruner, Lifeline, Cron | Stager, Pruner, Lifeline, Cron | | Oban.Testing (assert_enqueued, perform_job) | @rivia-labs/oban-ts/testing (assertEnqueued, performJob) | | :telemetry | oban.on("job:stop", ...), attachDefaultLogger | | pause_queue, scale_queue, drain_queue, cancel_job, retry_job | mesmos nomes em camelCase |

Instalação

pnpm add @rivia-labs/oban-ts        # + o seu ORM: typeorm | knex | drizzle-orm | @prisma/client | pg

1. Migration

A versão aplicada fica num comentário em oban_jobs, como no Oban. Rodar up de novo não faz nada, e o upgrade aplica só as versões que faltam.

TypeORM (equivalente a Oban.Migration.up dentro de uma migration Ecto):

// src/migrations/1727000000000-AddObanJobsTable.ts
import { obanMigration } from "@rivia-labs/oban-ts/typeorm";
export class AddObanJobsTable1727000000000 extends obanMigration({ version: 1 }) {}

Knex / qualquer adaptador: await Migration.up(knexRepo(knex), { version: 1 }) dentro da migration.

Prisma Migrate / drizzle-kit (só aceitam SQL): gere o arquivo e cole na migration:

import { Migration } from "@rivia-labs/oban-ts";
console.log(Migration.sql({ version: 1 })); // opções: prefix, from, direction: "down"

2. Configuração e start

import { Oban } from "@rivia-labs/oban-ts";
import { typeormRepo } from "@rivia-labs/oban-ts/typeorm";

export const oban = new Oban({
	repo: typeormRepo(dataSource),
	queues: { default: 10, mailers: 20 },     // concorrência por fila, por nó
	workers: [SendReceipt],
	pruner: { maxAge: 60 * 60 * 24 },         // padrão: 60s. `false` desliga
	lifeline: { rescueAfter: 60 * 60 },       // padrão: 1h. `false` desliga
	plugins: [new Cron({ crontab: [["0 8 * * *", DailyReport]], timezone: "America/Sao_Paulo" })],
});

await oban.start();
process.on("SIGTERM", () => oban.stop()); // espera os jobs em execução (shutdownGracePeriod: 15s)

| Opção | Padrão | | | --- | --- | --- | | prefix | "public" | schema Postgres das tabelas (multi-tenant / isolamento) | | name | "Oban" | instâncias com nomes diferentes elegem líderes separados | | node | hostname-pid | gravado em attempted_by | | testing | "disabled" | "manual" ou "inline" (veja Testes) | | notifier | PostgresNotifier se o repo suporta LISTEN | IsolatedNotifier no caso contrário | | peer | PostgresPeer | false: o nó nunca lidera (não roda stager, pruner nem cron) | | pollInterval | 1000 ms | rede de segurança caso uma notificação se perca | | shutdownGracePeriod | 15000 ms | |

Workers

import { defineWorker, cancel, snooze } from "@rivia-labs/oban-ts";

export const SendReceipt = defineWorker<{ orderId: number }>({
	name: "SendReceipt",                   // gravado no banco: não renomeie com jobs pendentes
	queue: "mailers",
	maxAttempts: 5,                        // padrão 20
	priority: 0,                           // 0 é a mais alta
	unique: { period: 300, keys: ["orderId"] },
	backoff: job => 2 ** job.attempt,      // segundos. Padrão: 15s + 2^attempt com jitter
	timeout: () => 30_000,                 // ms. Padrão: sem timeout
	async perform(job, { signal }) {
		const order = await orders.find(job.args.orderId);
		if (!order) return cancel("pedido removido"); // vai para `cancelled`, sem retry
		if (!mailer.ready) return snooze(60);         // reagenda sem gastar tentativa
		await mailer.send(order, { signal });         // `throw` → retry com backoff
	},
});

O signal é abortado no timeout, no cancelJob de um job em execução (em qualquer nó) e no shutdown depois do grace period. Repasse-o a fetch, ao driver do banco e similares.

Para enfileirar um job de outro serviço sem importar o worker (outbox entre serviços), use Job.new(args, { worker: "SendReceipt", queue: "mailers" }).

Outbox em cada ORM

Todos os adaptadores recebem o handle de transação nativo do ORM em { tx }:

// TypeORM: EntityManager ou QueryRunner
await dataSource.transaction(async manager => { await oban.insert(job, { tx: manager }); });

// Knex
await knex.transaction(async trx => { await oban.insert(job, { tx: trx }); });

// Drizzle
await db.transaction(async tx => { await oban.insert(job, { tx }); });

// Prisma (interactive transaction)
await prisma.$transaction(async tx => { await oban.insert(job, { tx }); });

// pg puro: um client em que você já deu BEGIN
await oban.insert(job, { tx: client });

| Adaptador | LISTEN/NOTIFY | Observações | | --- | --- | --- | | TypeORM | ✅ (via driver.master) | aceita DataSource, EntityManager ou QueryRunner | | Knex | ✅ (conexão do pool) | clients pg/postgresql | | Drizzle | ✅ com node-postgres (db.$client) | outros drivers usam polling | | Prisma | ➖ polling | para LISTEN: notifier: new PostgresNotifier({ connect: pgListen(DATABASE_URL) }) | | pg | ✅ | pgRepo(pool) |

O NOTIFY é transacional: dentro do tx, o aviso aos workers só sai no commit. Um job de transação revertida nunca é anunciado.

Outro ORM (MikroORM, Sequelize, Kysely...)? Implemente a interface Repo (src/repo.ts): query(sql, params) para um statement com placeholders $n, mais transaction e using. Os parâmetros chegam só como string, número, booleano ou null.

Jobs únicos

unique: {
	period: 60,                       // segundos, ou "infinity"
	fields: ["worker", "queue", "args"],
	keys: ["orderId"],                // compara só essas chaves de args/meta
	states: ["available", "scheduled", "executing", "retryable", "completed"],
}

Se já existe um job igual, insert devolve esse job com conflict: true. Assim como no Oban, a checagem usa advisory lock transacional e é segura com inserts concorrentes. Dentro do seu tx, o lock dura até o commit.

Agendamento, filas e gestão

Worker.new(args, { scheduleIn: 3600 });            // ou scheduledAt: new Date(...)

await oban.pauseQueue({ queue: "mailers" });        // em todos os nós (localOnly: true para só este)
await oban.resumeQueue({ queue: "mailers" });
await oban.scaleQueue({ queue: "mailers", limit: 50 });
await oban.startQueue({ queue: "reports", limit: 2 });
await oban.stopQueue({ queue: "reports" });
oban.checkQueue("mailers");                         // { limit, paused, running: [ids], ... }

await oban.cancelJob(id);   await oban.cancelAllJobs({ queue: "mailers", states: ["scheduled"] });
await oban.retryJob(id);    await oban.retryAllJobs({ worker: "SendReceipt", states: ["discarded"] });
await oban.deleteJob(id);   await oban.listJobs({ states: ["retryable"], limit: 50 });

Testes

import { assertEnqueued, refuteEnqueued, allEnqueued, performJob } from "@rivia-labs/oban-ts/testing";

// "manual": grava no banco mas não executa nada sozinho
const oban = new Oban({ repo, testing: "manual", workers: [SendReceipt] });
await checkout(order);
await assertEnqueued(oban, { worker: SendReceipt, args: { orderId: order.id } });
await oban.drainQueue({ queue: "mailers", withScheduled: true, withSafety: false });

// "inline": executa na hora do insert, sem banco
const inline = new Oban({ repo, testing: "inline", workers: [SendReceipt] });

// Unitário, sem banco nem Oban
expect(await performJob(SendReceipt, { orderId: 1 })).toEqual({ state: "success" });

Telemetria

oban.on("job:stop", ({ job, outcome, durationMs }) => metrics.timing(`jobs.${job.worker}`, durationMs));
oban.on("job:exception", ({ job, error }) => sentry.captureException(error, { extra: { id: job.id } }));
attachDefaultLogger(oban, logger); // job:start/stop/exception, plugin:exception, error

Eventos: job:start, job:stop, job:exception, plugin:stop, plugin:exception, leader:change, error.

Arquitetura

src/
  oban.ts              Oban: config, ciclo de vida e API pública
  job.ts  worker.ts    Job/changeset, defineWorker e resultados
  repo.ts              contrato Repo (papel do Ecto.Repo)
  adapters/            typeorm, knex, drizzle, prisma, pg
  engine.ts            contrato Engine
  engines/basic.ts     Postgres: fetch com FOR UPDATE SKIP LOCKED, acks, únicos
  engines/inline.ts    testing: "inline"
  migration.ts         migrations versionadas (migrations/postgres/vNN.ts)
  notifier.ts peer.ts  LISTEN/NOTIFY e eleição de líder
  queue/producer.ts    um por fila por nó: concorrência, pause/scale, shutdown
  queue/executor.ts    executa o perform e faz o ack
  plugins/             stager, pruner, lifeline, cron
  cron/expression.ts   parser de cron (5 campos, nomes, @nicknames, timezone)
  telemetry.ts testing.ts

Estados do job, como no Oban: available → executing → completed, ou então retryable (volta a available via Stager), scheduled, cancelled, discarded.

Diferenças deliberadas em relação ao Oban

• timestamptz em vez de timestamp, porque o pg interpreta timestamp sem fuso como hora local.
• errors é jsonb (array) em vez de jsonb[], que nem todo driver decodifica.
• Todo SQL devolve os jobs como to_jsonb(j) e recebe arrays/objetos como JSON. Com isso o mesmo SQL funciona igual em pg, Prisma, Knex e Drizzle, sem depender de como cada driver mapeia bigint e arrays.
• A migration v1 já traz o schema final do Oban (equivalente a V1..V14). As próximas versões entram como v02.ts, v03.ts...
• Único via hash em meta.uniq_key + advisory lock + busca no índice GIN de meta.
• Um timeout aborta o signal, mas JavaScript não consegue matar uma Promise. Um perform que ignora o signal continua rodando depois do timeout (o job já foi marcado como erro).

Desenvolvimento

pnpm db:up              # Postgres 16 em localhost:54330 (docker compose)
pnpm test               # unitários + integração + exemplo de testes (TEST_DATABASE_URL para outro banco)
pnpm examples           # executa todos os exemplos
pnpm typecheck && pnpm build
claude plugin validate . # valida o plugin/marketplace das skills

Os testes de integração rodam o mesmo contrato de outbox em cada adaptador (TypeORM, Knex, Drizzle, Prisma 7 real e pg): rollback descarta o job, o commit o entrega via NOTIFY, e únicos funcionam dentro do tx.

Roadmap

• Engines para MySQL (Dolphin) e SQLite (Lite)
• Reindexer, replace em jobs únicos
• CLI oban-ts migrate / oban-ts gen-sql