@kido/event-busCopy Anchor Link

Architecture Paradigm: Transactional Outbox Pattern with Atomic DB Commit + Polling Dispatcher with Exponential Backoff + Idempotent Consumer State Guards + Dead-Letter Queue (DLQ) Replay + Redis Streams Consumer Groups.

💥 The Real-World BottleneckCopy Anchor Link

Event-driven microservices and asynchronous message passing frequently suffer from data loss and duplicate processing:

1. The In-Memory EventEmitter Data Loss Disaster:

   Copy Text

   // The Naive Disaster: In-memory async event dispatch
   await db.orders.create({ data: order });
   eventEmitter.emit('order.created', order); // If the container restarts or crashes right now, this event is GONE!

   Standard in-memory event emitters lose critical side-effects (loyalty points, inventory reservations, customer confirmation emails) upon any container restart or crash.
2. The Dual-Write Race Condition: Attempting to publish to RabbitMQ/Kafka directly after database insertion is non-atomic. If the database transaction commits but the network broker rejects the publish, data becomes permanently corrupted across services.
3. Poison Pills & Duplicate Event Re-processing: When a downstream consumer crashes on a malformed event, retry loops re-execute non-idempotent operations repeatedly (e.g., charging credit cards multiple times), eventually taking down downstream workers.

🏛️ 3-Tier Production Evolution LadderCopy Anchor Link

📦 Architecture & Core API ExportsCopy Anchor Link

@kido/event-bus is composed of 5 decoupled modules:

1. Domain Event Bus & Router (src/event-bus.ts)Copy Anchor Link

• DomainEventBus: Type-safe publish-subscribe engine with wildcard matching:
  • Exact: order.created
  • Single segment wildcard: order.*
  • Multi-segment wildcard: order.#
• Middleware support for distributed tracing (traceparent), performance metrics, and payload logging.

2. Transactional Outbox & Poller (src/transactional-outbox.ts)Copy Anchor Link

• TransactionalOutbox: Encapsulates outbox row creation within your existing database transaction (Prisma, Kysely, Drizzle).
• PollingDispatcher: Background daemon polling pending outbox entries with exponential backoff and jitter (exponentialBackoffWithJitter).

3. Idempotent Consumer Guard (src/idempotent-consumer.ts)Copy Anchor Link

• IdempotentConsumer: Wraps message handlers to prevent duplicate execution using idempotency keys (entity_id:version or event_id).
• Automatically caches execution results to return idempotent responses on duplicate deliveries.

4. Dead-Letter Queue & Redrive (src/dead-letter-queue.ts)Copy Anchor Link

• DeadLetterQueue: Stores poisoned or permanently failed events with error traces and payload snapshots.
• dlq.redrive(eventId): Re-publishes resolved events back to the primary event bus after fixing bugs.

5. Distributed Redis Streams Driver (src/drivers/)Copy Anchor Link

• RedisEventBusDriver: Enterprise driver using Redis Streams (XADD, XREADGROUP, consumer groups) for multi-pod container architectures.

💻 Production Implementation & Code SnippetCopy Anchor Link

Transactional Outbox & Idempotent ConsumerCopy Anchor Link

import { 
  DomainEventBus, 
  TransactionalOutbox, 
  IdempotentConsumer,
  InMemoryEventBusDriver 
} from '@kido/event-bus';
import { z } from 'zod';

// 1. Define Strongly-Typed Zod Event Schema
const OrderCreatedEventSchema = z.object({
  orderId: z.string(),
  userId: z.string(),
  totalAmount: z.number().positive(),
});

// 2. Initialize Event Bus
export const eventBus = new DomainEventBus({
  driver: new InMemoryEventBusDriver(),
});

// 3. Register Idempotent Consumer
const inventoryConsumer = new IdempotentConsumer({
  consumerName: 'inventory_deduction_worker',
  storage: myRedisOrDbStorage, // Persists processed event IDs
});

eventBus.subscribe('order.created', async (event) => {
  const payload = OrderCreatedEventSchema.parse(event.payload);

  // Guarantees this block executes EXACTLY ONCE per orderId
  await inventoryConsumer.execute(payload.orderId, async () => {
    console.log(`Deducting stock for order ${payload.orderId}...`);
    await deductStock(payload.orderId);
  });
});

// 4. Atomic Transactional Outbox Publisher
export async function createOrder(orderData: any) {
  // Commit order record and outbox event in ONE atomic SQL transaction
  return await db.$transaction(async (tx) => {
    const order = await tx.order.create({ data: orderData });

    // Store in outbox table
    await tx.outboxEvent.create({
      data: {
        id: crypto.randomUUID(),
        topic: 'order.created',
        payload: {
          orderId: order.id,
          userId: order.userId,
          totalAmount: order.total,
        },
        status: 'PENDING',
      },
    });

    return order;
  });
}

⚡ Benchmark & Verified k6 PerformanceCopy Anchor Link

Benchmarked on Bun 1.3 / Node.js 22 LTS across 100,000 dispatched events:

🚀 Surgical Extraction & CLI InstallationCopy Anchor Link

# Install directly via Kido CLI
npx @kido/cli add @kido/event-bus