@nipigev2/messaging-client

Shared Redis Streams messaging for Nipige services — one source of truth for the stream publisher, the resilient consumer-group utility, and the shared event/intent types. Replaces the per-service notify.publisher.ts copies and the duplicated amqpConsumer.ts, as the platform moves off RabbitMQ onto Redis Streams.

Inject your service's existing ioredis client — the package does no config or connection management of its own.

Publish

import { publishToStream, publishNotification, streamName } from '@nipigev2/messaging-client';
import { redisClient } from '../utils/cache';
import logger from '../utils/logger';

// domain event → its business stream (fire-and-forget; never throws)
await publishToStream(redisClient, streamName('booking_events'), bookingEvent, { logger });

// notification intent → the unified notification bus (notify:stream)
await publishNotification(redisClient, intent, { logger });

Consume

import { startStreamConsumer, streamName } from '@nipigev2/messaging-client';
import { redisClient } from '../utils/cache';
import logger from '../utils/logger';

const handle = startStreamConsumer(redisClient, {
  name: 'dispatch-service.bookingEvents',
  stream: streamName('booking_events'),
  group: 'dispatch-service',      // one group per consuming service (fan-out)
  handler: async (event) => { /* switch on event.eventType */ },
  logger,
});
// on shutdown: await handle.stop();

Semantics: at-least-once (manual XACK), crash recovery (XPENDING + XCLAIM reclaim sweep), poison-pill dead-lettering after maxDeliveries, load-balancing across replicas within a group. Blocking reads run on a duplicated connection.

Conventions

• One stream per former exchange: streamName('<exchange>') → <exchange>:stream.
• One consumer group per consuming service (fan-out). Multiple instances of a service share its group (load-balanced).
• Entries store the JSON payload under the single field payload (STREAM_FIELD).

Build / publish

npm run build (tsc → dist/), then publish to the private registry. Consumers depend on it like @nipigev2/rbac-client.