swap-event-transport
Move domain events from BullMQ queues to a log-based broker (Kafka, Redpanda, NATS JetStream). Use when the user decides event volume, ordering, replay or fan-out has outgrown Redis queues, or asks ab
Files1
View on GitHubSKILL.md.claude/skills/swap-event-transport/SKILL.md
GitHubSwap the event transport
- Interface:
EventBus(abstract class,publish(events): Promise<void>) inapps/worker/src/outbox/event-bus.ts. - Today:
BullMqEventBuscopies each event into every queue whose filter ineventSubscribers(packages/jobs/src/queues.ts) accepts it, with jobId = event id. At least once, unordered. - Bound in:
apps/worker/src/outbox/outbox.module.ts. - Env: none beyond
REDIS_URLtoday.
Swap
- A class extending
EventBusthat produces eachEventEnvelopeto one topic keyed byevent.key(per-aggregate order), idempotently (the relay may republish a batch after a crash). Bind it inoutbox.module.ts. The relay,emitEventand every producer stay as they are. - Consumers are what move. Each is a BullMQ
@Processoron anevents-*queue today:events-audit(apps/worker/src/audit/audit.processor.ts),events-realtime(apps/worker/src/realtime/realtime.processor.ts),events-webhooks(apps/webhooks/src/outbound/fanout.processor.ts),events-notifications(apps/notifications/src/events/events.processor.ts),events-billing(apps/api/src/modules/billing/billing-events.processor.ts). Each becomes a consumer group applying itseventSubscribersfilter itself, still deduping on the event id. Moving one consumer at a time works if the bus publishes to both for a while. - The broker's address in
apps/worker/src/env.tsand each consumer'ssrc/env.ts,.env.example,docs/environment.md; KEDA triggers indeploy/charts/stackchange from queue length to consumer lag.
Tests
The outbox relay tests in apps/worker/test/worker.integration.test.ts, and each
consumer's integration suite. A broker needs a local container for tests; add it to
docker-compose.yml with a memory limit like the others.
Finish
- The verify skill.
- Ask the
revieweragent to review the change, and thesecurity-revieweragent: a new implementation brings its own credentials and sends data somewhere new. If you wrote a migration, themigration-revieweragent too. - Update the seam's row in the README's "Scaling path" table if what's "Now" changed.