@forgeframework/eventqos

Kafka event streaming with Quality of Service guarantees for the Forge Framework.

Part of the Forge Framework — a TypeScript framework for backend microservices and web applications.

Installation

npm install @forgeframework/eventqos

Usage

import {
  ForgeEventQoSFactory,
  QoSLevel,
  type ForgeEvent,
  type ProducerConfig,
  type ConsumerConfig,
} from '@forgeframework/eventqos';
import { ForgeLogger } from '@forgeframework/logging';

const logger = new ForgeLogger({ level: 'info' });
const factory = new ForgeEventQoSFactory(logger);

const producerConfig: ProducerConfig = {
  brokers: ['localhost:9092'],
  topic: 'orders',
  clientId: 'order-service',
  compression: 'snappy',
  retry: { maxRetries: 3, initialDelayMs: 100, maxDelayMs: 5000, factor: 2 },
  dlq: {
    topic: 'orders.dlq',
    maxRetries: 5,
    backoff: { initialDelayMs: 500, maxDelayMs: 30000, factor: 2, jitter: true },
  },
};

const producer = factory.createAtLeastOnce(producerConfig);

const event: ForgeEvent<{ orderId: string; total: number }> = {
  eventId: '550e8400-e29b-41d4-a716-446655440000',
  type: 'order.created',
  timestamp: new Date().toISOString(),
  correlationId: 'req-abc-123',
  source: 'order-service',
  data: { orderId: 'ORD-001', total: 99.99 },
  metadata: { version: 1, region: 'us-east-1' },
};

await producer.send(event);
await producer.flush();

const consumerConfig: ConsumerConfig = {
  brokers: ['localhost:9092'],
  topic: 'orders',
  groupId: 'inventory-group',
  clientId: 'inventory-service',
  qos: QoSLevel.AT_LEAST_ONCE,
  batch: { batchSize: 10, flushIntervalMs: 2000 },
  dlq: {
    topic: 'orders.dlq',
    maxRetries: 3,
    backoff: { initialDelayMs: 200, maxDelayMs: 10000, factor: 2 },
  },
};

const consumer = factory.createConsumer(consumerConfig);

await consumer.subscribeBatch('orders', async (events) => {
  for (const evt of events) {
    logger.info('Processing order', { orderId: evt.data.orderId, correlationId: evt.correlationId });
  }
}, { batchSize: 10, flushIntervalMs: 2000 });

const health = consumer.health();
const metrics = await health.getOffsetMetrics();
logger.info('Consumer health', { healthy: health.isHealthy(), metrics });

await producer.disconnect();
await consumer.disconnect();

Documentation

Full API reference and guides: https://carbonforge.io/framework/docs/events/eventqos

(Documentation site launching soon.)

License

MIT © Carbon Forge