@mikara89/cap-nest

Core NestJS package for CAP reliable messaging.

This package provides:

• CapModule
• CapService
• @CapSubscribe
• storage and transport DI tokens
• outbox/inbox models
• retry scheduler wiring
• in-memory mode for tests and local examples

Minimal Usage

import { Module } from '@nestjs/common';
import { CapModule } from '@mikara89/cap-nest';

@Module({
  imports: [CapModule.forInMemory()],
})
export class AppModule {}

forRoot, forRootAsync, and forInMemory accept the core envelope migration option:

CapModule.forRoot({
  imports: [storageModule, transportModule],
  messageEnvelope: { legacyUnversioned: 'reject' },
});

They also accept the optional framework-neutral core diagnostics sink:

import { type CapMessagingDiagnosticsPort } from '@mikara89/cap-core';

const diagnostics: CapMessagingDiagnosticsPort = {
  emit(event) {
    console.log(event.type, event.id, event.at);
  },
};

CapModule.forRoot({
  imports: [storageModule, transportModule],
  diagnostics,
});

Diagnostics are best-effort and non-blocking. Events intentionally exclude message payloads and headers; sink failures are logged and swallowed. They are not a durable audit stream. See the repository diagnostics guide for event and retry semantics.

The default warn mode accepts strict legacy { payload, headers? } bodies and warns once per engine. New bridges that need one body should use createCapMessageEnvelope() re-exported by @mikara89/cap-nest. Ordinary business payloads containing a payload field are not broadly unwrapped.

await capService.publish('user.created', { id: 'u1' });

@CapSubscribe({ topic: 'user.created', group: 'mail-service' })
async handleUserCreated(payload: unknown) {
  // handle message
}

Inbox Recovery

Messaging administration

CapService delegates requeueInbox(id), requeueOutbox(id), and getMessagingSnapshot() directly to core. These APIs are storage-capability dependent and do not add HTTP endpoints. Only failed/dead-letter records can be requeued; the normal scheduler later invokes handlers or emits messages. The snapshot is an operational aggregate and its inbox/outbox halves can represent slightly different instants.

CapModule passes scheduler options to core. scheduler.inboxFallbackWindowMs defaults to 240_000 milliseconds and controls when a pending inbox row may be retried after an interrupted subscriber attempt:

CapModule.forRoot({
  scheduler: { inboxFallbackWindowMs: 300_000 },
});

Recovery is at least once and nontransactional. Keep the window longer than normal handler execution and backlog time, and make handlers idempotent because a slow handler may otherwise be recovered as stale.

Subscription Lifecycle

Nest discovers @CapSubscribe handlers during onModuleInit and registers them without broker I/O. After all modules initialize, CAP's onApplicationBootstrap hook awaits adapter initialization and SubscriberPort.consume() attachment for every registered handler. An initial attachment failure therefore rejects application bootstrap instead of allowing the application to become ready without its consumers.

Use Nest's normal awaited bootstrap before accepting traffic:

const app = await NestFactory.create(AppModule);
await app.listen(3000);

Enable Nest shutdown hooks and await app.close() (or the framework shutdown path) for graceful consumer cleanup. CAP deduplicates shutdown and closes the subscriber during onApplicationShutdown.

Documentation

• Repository overview
• Getting started
• API reference
• Package export surface
• Architecture
• Adapters
• ADRs