ts-domain-events

Features:

• Singleton Publisher with subscriber registry
• InMemorySubscriber for testing and event collection
• SnsSubscriber for publishing to AWS SNS topics
• @SnsAttribute decorator for custom SNS MessageAttributes
• DeferredSubscriber base class for transaction-safe publishing
• PrismaDeferredSubscriber for Prisma ORM integration
• EventProcessorCommand for polling SQS queues with exponential backoff retry
• FailoverPublisher for recovering from SNS publish failures
• CLI commands: domain-events process-events and domain-events retry-failures
• Configurable logging interface with NullLogger and ConsoleLogger

Installation

npm install ts-domain-events

For Prisma integration, install the peer dependency:

npm install @prisma/client

AWS SDK clients are bundled as regular dependencies — no separate AWS SDK installation needed.

Requirements

• Node.js >= 18
• TypeScript >= 5.3
• tsconfig.json must enable decorators (required for @SnsAttribute):

{
  "compilerOptions": {
    "experimentalDecorators": true,
    "emitDecoratorMetadata": true
  }
}

Important: If you use @SnsAttribute, add import 'reflect-metadata' at your application entry point before any decorated classes are loaded.

Quick Start

import 'reflect-metadata';
import { Publisher, InMemorySubscriber, DomainEvent } from 'ts-domain-events';

// Define an event
class UserRegisteredEvent implements DomainEvent {
  constructor(private readonly userId: string) {}
  occurredOn(): Date { return new Date(); }
  toArray(): Record<string, any> { return { userId: this.userId }; }
}

// Subscribe and publish
const subscriber = new InMemorySubscriber();
Publisher.instance().subscribe(subscriber);
Publisher.instance().publish(new UserRegisteredEvent('user-123'));

const events = subscriber.popEvents();
console.log(events[0]); // UserRegisteredEvent

Core Concepts

• Publisher — singleton that dispatches events to all registered subscribers. Call Publisher.instance() to get the shared instance.
• Subscriber — abstract base class; implement notify(event) to handle events. Override subscribesToEventType() to filter to a single event type.
• DomainEvent — interface all events must implement: occurredOn() and toArray().

Defining Domain Events

import { DomainEvent } from 'ts-domain-events';

export class OrderCreatedEvent implements DomainEvent {
  constructor(
    private readonly orderId: string,
    private readonly customerId: string,
    private readonly amount: number,
    private readonly timestamp: Date = new Date()
  ) {}

  occurredOn(): Date {
    return this.timestamp;
  }

  toArray(): Record<string, any> {
    return {
      orderId: this.orderId,
      customerId: this.customerId,
      amount: this.amount,
      occurredOn: this.timestamp.toISOString(),
    };
  }
}

toArray() must return a JSON-serializable object — it becomes the event payload in SNS/SQS messages.

Basic Usage

Publisher

import { Publisher } from 'ts-domain-events';

const publisher = Publisher.instance();

// Register subscribers before publishing
publisher.subscribe(mySubscriber);

// Publish an event — all matching subscribers are notified
publisher.publish(new OrderCreatedEvent('order-123', 'customer-456', 99.99));

// Purge stored state from all subscribers
publisher.purge();

// Reset — clears all subscribers (useful in test teardown)
publisher.reset();

InMemorySubscriber

Stores events in memory for inspection. Useful in tests and for temporary buffering.

import { Publisher, InMemorySubscriber } from 'ts-domain-events';

const subscriber = new InMemorySubscriber();
Publisher.instance().subscribe(subscriber);

// Trigger domain logic that publishes events
placeOrder(orderId);

// Retrieve and clear all stored events
const events = subscriber.popEvents();
console.log(events.length); // number of published events

// Inspect without clearing
const peek = subscriber.peekEvents();
console.log(subscriber.count()); // still the same count

// Clear without returning
subscriber.purge();

Custom Subscriber

import { Subscriber, DomainEvent } from 'ts-domain-events';

export class AuditLogSubscriber extends Subscriber {
  // Return an event class name to receive only that type; null = all events
  subscribesToEventType(): string | null {
    return 'OrderCreatedEvent';
  }

  async notify(event: DomainEvent): Promise<void> {
    await auditLog.write({
      event: event.constructor.name,
      data: event.toArray(),
      at: event.occurredOn(),
    });
  }
}

Publisher.instance().subscribe(new AuditLogSubscriber());

AWS SNS/SQS Integration

SnsSubscriber

Publishes each event to an SNS topic as a JSON message.

import 'reflect-metadata';
import { SNSClient } from '@aws-sdk/client-sns';
import {
  Publisher,
  SnsSubscriber,
  ConsoleLogger,
} from 'ts-domain-events';

const snsClient = new SNSClient({ region: 'us-east-1' });

const subscriber = new SnsSubscriber(
  snsClient,
  'arn:aws:sns:us-east-1:123456789012:my-topic',
  undefined,    // failureFactory (optional — see Failover section)
  undefined,    // failureRepository (optional — see Failover section)
  new ConsoleLogger()
);

Publisher.instance().subscribe(subscriber);

// Publishes event to SNS
Publisher.instance().publish(new OrderCreatedEvent('order-123', 'customer-456', 99.99));

SNS Message Format

Every event is published as a JSON string with this structure:

{
  "name": "OrderCreatedEvent",
  "data": {
    "orderId": "order-123",
    "customerId": "customer-456",
    "amount": 99.99,
    "occurredOn": "2024-01-15T10:30:00.000Z"
  }
}

When SQS receives this message from an SNS subscription, the SQS message body wraps it in the SNS notification envelope:

{
  "Type": "Notification",
  "MessageId": "abc-123",
  "TopicArn": "arn:aws:sns:us-east-1:123456789012:my-topic",
  "Message": "{\"name\":\"OrderCreatedEvent\",\"data\":{...}}",
  "Timestamp": "2024-01-15T10:30:00.000Z"
}

EventParser.parse() handles this nested structure automatically.

@SnsAttribute Decorator

Add custom SNS MessageAttributes to event classes. These let SNS subscribers filter messages by attribute.

Important: Add import 'reflect-metadata' at your application entry point.

import 'reflect-metadata';
import { SnsAttribute, DomainEvent } from 'ts-domain-events';

@SnsAttribute('source', 'order-service')
@SnsAttribute('version', '1.0')
@SnsAttribute('eventType', 'domain')
export class OrderCreatedEvent implements DomainEvent {
  constructor(
    private readonly orderId: string,
    private readonly amount: number
  ) {}

  occurredOn(): Date { return new Date(); }

  toArray(): Record<string, any> {
    return { orderId: this.orderId, amount: this.amount };
  }
}

When SnsSubscriber publishes this event, the SNS PublishCommand includes:

{
  "MessageAttributes": {
    "source":    { "DataType": "String", "StringValue": "order-service" },
    "version":   { "DataType": "String", "StringValue": "1.0" },
    "eventType": { "DataType": "String", "StringValue": "domain" }
  }
}

All attribute values are stored as SNS String type. Multiple @SnsAttribute decorators are preserved in declaration order (top to bottom).

You can also read attributes programmatically:

import { getSnsAttributes } from 'ts-domain-events';

const attrs = getSnsAttributes(OrderCreatedEvent);
// [
//   { name: 'source',    value: 'order-service' },
//   { name: 'version',   value: '1.0' },
//   { name: 'eventType', value: 'domain' }
// ]

Deferred Publishing (Transaction Safety)

DeferredSubscriber wraps another subscriber and holds events until onCommit() is called. If the transaction rolls back, onRollback() discards all pending events. This ensures events are never published for database changes that didn't persist.

notify(event)  →  stored in pendingEvents (not forwarded yet)
onCommit()     →  all pending events forwarded to decorated subscriber
onRollback()   →  all pending events discarded

Prisma Integration

PrismaDeferredSubscriber uses Prisma Client Extensions to detect write operations and commit/rollback automatically. This requires Prisma 5+ (stable) or 6+.

import 'reflect-metadata';
import { PrismaClient } from '@prisma/client';
import { SNSClient } from '@aws-sdk/client-sns';
import {
  Publisher,
  SnsSubscriber,
  PrismaDeferredSubscriber,
} from 'ts-domain-events';

const prisma = new PrismaClient();
const snsClient = new SNSClient({ region: 'us-east-1' });

// Inner subscriber that actually publishes to SNS
const snsSubscriber = new SnsSubscriber(
  snsClient,
  'arn:aws:sns:us-east-1:123456789012:my-topic'
);

// Wrap with Prisma deferred subscriber
const deferredSubscriber = new PrismaDeferredSubscriber(snsSubscriber);

// Create an extended client — use `db` everywhere instead of `prisma`
const db = prisma.$extends(deferredSubscriber.extension());

// Register with Publisher
Publisher.instance().subscribe(deferredSubscriber);

Now events are published only after a successful database write:

// Auto-commit (single operation)
Publisher.instance().publish(new OrderCreatedEvent('pending-id'));
const order = await db.order.create({
  data: { customerId: '123', amount: 99.99 }
});
// Event is published after create() succeeds

// Interactive transaction — wrap with withTransaction() for all-or-nothing guarantees
await deferredSubscriber.withTransaction(() =>
  db.$transaction(async (tx) => {
    const order = await tx.order.create({
      data: { customerId: '123', amount: 99.99 }
    });

    // Stored but NOT published yet
    Publisher.instance().publish(new OrderCreatedEvent(order.id));

    await tx.inventory.update({
      where: { productId: order.productId },
      data: { quantity: { decrement: 1 } }
    });
    // If this throws, the event above is discarded
  })
);
// Transaction committed: event is now published

Write operations that trigger onCommit(): create, createMany, createManyAndReturn, update, updateMany, updateManyAndReturn, upsert, delete, deleteMany, executeRaw, queryRaw, runCommandRaw.

Custom ORM Integration

For other ORMs or custom transaction patterns, extend DeferredSubscriber directly:

import { DeferredSubscriber, Subscriber } from 'ts-domain-events';

export class MyOrmDeferredSubscriber extends DeferredSubscriber {
  constructor(decorated: Subscriber) {
    super(decorated);
  }

  // Call this when your transaction commits
  async afterTransactionCommit(): Promise<void> {
    await this.onCommit();
  }

  // Call this when your transaction rolls back
  afterTransactionRollback(): void {
    this.onRollback();
  }
}

Note: A TypeORM adapter (TypeOrmDeferredSubscriber) is not included in the current release. Use the pattern above to add TypeORM support.

Processing Events from SQS

Implement EventProcessor

For each event type you want to handle, implement EventProcessor:

import {
  EventProcessor,
  UnavailableException,
} from 'ts-domain-events';

export class OrderCreatedProcessor implements EventProcessor {
  constructor(private readonly emailService: EmailService) {}

  async processEvent(event: Record<string, any>): Promise<void> {
    const { orderId, customerId, amount } = event.data;

    try {
      await this.emailService.sendConfirmation(customerId, orderId, amount);
    } catch (error) {
      if (error instanceof EmailServiceUnavailableError) {
        // Temporary failure — triggers exponential backoff retry
        const ex = new UnavailableException('Email service unavailable', error);
        ex.setMaximumRetryCount(15);
        ex.setDelaySeconds(60); // Wait at least 60 s between retries
        throw ex;
      }

      // Permanent failure — message goes to DLQ after SQS maxReceiveCount
      throw error;
    }
  }
}

EventProcessorCommand

EventProcessorCommand polls an SQS queue and dispatches messages to processors:

import { SQSClient } from '@aws-sdk/client-sqs';
import {
  EventProcessorCommand,
  ConsoleLogger,
} from 'ts-domain-events';

const sqsClient = new SQSClient({ region: 'us-east-1' });

const processors: Record<string, new () => EventProcessor> = {
  OrderCreatedProcessor,
  OrderCancelledProcessor,
};

const command = new EventProcessorCommand(sqsClient, {
  queueUrl: 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue',
  eventMap: {
    OrderCreatedEvent:   ['OrderCreatedProcessor'],
    OrderCancelledEvent: ['OrderCancelledProcessor', 'RefundProcessor'],
  },
  processorFactory: (className) => new processors[className](),
  maxMessages: 100,
  interactive: false,
  logger: new ConsoleLogger(),
});

// Graceful shutdown
process.on('SIGINT', () => command.stop());
process.on('SIGTERM', () => command.stop());

await command.execute();

In interactive: true mode the command polls continuously until stop() is called or the process is terminated.

Error handling summary

| Processor throws | Outcome | |---|---| | Returns normally | Message deleted from queue | | UnavailableException (retry < max) | Requeued with exponential backoff delay | | UnavailableException (retry >= max) | Not deleted; moves to DLQ after SQS maxReceiveCount | | Any other Error | Not deleted; moves to DLQ after SQS maxReceiveCount |

Retry Mechanism

When a processor throws UnavailableException, EventProcessorCommand requeues the message with an exponential backoff delay.

How it works

1. Extract RetryCount from SQS message attributes (0 on first attempt)
2. If retryCount < exception.getMaximumRetryCount():
   • Calculate delay = 2^retryCount (in seconds)
   • Apply minimum: delay = max(delay, exception.getDelaySeconds())
   • Cap at 900 seconds (AWS SQS maximum)
   • Delete original message
   • Enqueue new message with DelaySeconds and RetryCount + 1
3. If retryCount >= maximumRetryCount: throw (message stays, goes to DLQ)

Backoff schedule (default, no minimum delay)

| Retry | Delay | |---|---| | 0 | 1 s | | 1 | 2 s | | 3 | 8 s | | 5 | 32 s | | 7 | 128 s | | 9 | 512 s | | ≥10 | 900 s (capped) |

UnavailableException API

import { UnavailableException } from 'ts-domain-events';

// Default: 20 retries, no minimum delay
throw new UnavailableException('Database timeout');

// Custom retry count
const ex = new UnavailableException('Rate limited');
ex.setMaximumRetryCount(10);
throw ex;

// Custom minimum delay (e.g., wait at least 5 minutes)
const ex2 = new UnavailableException('Scheduled maintenance');
ex2.setDelaySeconds(300);
ex2.setMaximumRetryCount(5);
throw ex2;

// Wrap original error
try {
  await externalApi();
} catch (cause) {
  const ex3 = new UnavailableException('API unavailable', cause instanceof Error ? cause : undefined);
  throw ex3;
}

| Property | Default | Setter | |---|---|---| | delaySeconds | 0 | setDelaySeconds(n) | | maximumRetryCount | 20 | setMaximumRetryCount(n) |

RetryCalculator

The utility behind the retry logic is also exported for direct use:

import { RetryCalculator } from 'ts-domain-events';

RetryCalculator.calculateDelay(3);          // 8  (2^3)
RetryCalculator.applyMinimum(8, 60);        // 60 (minimum enforced)
RetryCalculator.capMaximum(1500);           // 900 (capped)
RetryCalculator.calculate(5, 60);           // 60 (32 < 60, uses minimum)
RetryCalculator.calculate(7, 60);           // 128
RetryCalculator.calculate(10);             // 900 (capped)

Failover System

When SnsSubscriber fails to publish, it can store the failure for later retry instead of throwing immediately. Pair SnsSubscriber with a failure factory and repository to enable this behavior.

Wiring SnsSubscriber with failover

import { SNSClient } from '@aws-sdk/client-sns';
import {
  Publisher,
  SnsSubscriber,
  SimplePublishFailureFactory,
  InMemoryFailureRepository,
  ConsoleLogger,
} from 'ts-domain-events';

const snsClient = new SNSClient({ region: 'us-east-1' });
const failureFactory = new SimplePublishFailureFactory();
const failureRepository = new InMemoryFailureRepository();

const subscriber = new SnsSubscriber(
  snsClient,
  'arn:aws:sns:us-east-1:123456789012:my-topic',
  failureFactory,
  failureRepository,
  new ConsoleLogger()
);

Publisher.instance().subscribe(subscriber);

With this setup, if SNS is unreachable, the failure is persisted in the repository rather than throwing SubscriberException.

FailoverPublisher

Retry stored failures in batches (oldest-first):

import { SNSClient } from '@aws-sdk/client-sns';
import { FailoverPublisher, ConsoleLogger } from 'ts-domain-events';

const failoverPublisher = new FailoverPublisher(
  snsClient,
  failureRepository,
  new ConsoleLogger()
);

// Returns true when all failures are retried; false if batch limit reached
const allDone = await failoverPublisher.retryNotifyOfPublishFailures(100);

InMemoryFailureRepository

Suitable for development, testing, and single-process deployments where persistence across restarts is not required.

import { InMemoryFailureRepository } from 'ts-domain-events';

const repo = new InMemoryFailureRepository();

console.log(repo.count());         // number of stored failures
repo.clear();                      // remove all failures (useful in tests)

For production, implement DomainEventPublishFailureRepository backed by DynamoDB, PostgreSQL, or another durable store.

import {
  DomainEventPublishFailure,
  DomainEventPublishFailureRepository,
} from 'ts-domain-events';

export class DynamoDbFailureRepository implements DomainEventPublishFailureRepository {
  add(failure: DomainEventPublishFailure): void { /* stage */ }
  async save(): Promise<void> { /* persist to DynamoDB */ }
  async getOldestMessage(): Promise<DomainEventPublishFailure | null> { /* query */ }
  async deleteMessageOfIdentity(identity: string): Promise<void> { /* delete */ }
}

CLI Commands

After installing the package, the domain-events binary is available in ./node_modules/.bin/ or via npx.

npx domain-events --help
npx domain-events process-events --help
npx domain-events retry-failures --help

domain-events process-events

Polls an SQS queue and dispatches events to processor classes.

npx domain-events process-events \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/my-queue \
  --config ./event-map.config.json \
  --max-messages 100

For continuous polling (e.g., as a worker process):

npx domain-events process-events \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/my-queue \
  --config ./event-map.config.json \
  --interactive

| Option | Required | Default | Description | |---|---|---|---| | --queue-url <url> | yes | — | SQS queue URL | | --config <path> | no | — | Path to event map JSON config | | --interactive | no | false | Poll continuously until SIGINT/SIGTERM | | --max-messages <n> | no | 100 | Max messages to process before exiting |

Config file format (event-map.config.json)

{
  "eventMap": {
    "OrderCreatedEvent": ["OrderCreatedProcessor"],
    "OrderCancelledEvent": ["OrderCancelledProcessor", "RefundProcessor"]
  },
  "processors": {
    "OrderCreatedProcessor":   "./processors/order-created-processor",
    "OrderCancelledProcessor": "./processors/order-cancelled-processor",
    "RefundProcessor":         "./processors/refund-processor"
  }
}

Each processor module must export the class by its name (or as default). Paths are resolved relative to process.cwd().

domain-events retry-failures

Retries failed SNS publishes from the failure repository.

npx domain-events retry-failures \
  --repository memory \
  --batch-size 1000

| Option | Required | Default | Description | |---|---|---|---| | --repository <type> | yes | — | Repository type: memory | | --batch-size <n> | no | 1000 | Max failures to retry per run | | --config <path> | no | — | Path to repository config JSON |

Currently supported repository types: memory. DynamoDB and PostgreSQL types are reserved for future releases — implement DomainEventPublishFailureRepository and wire it directly in code for now.

Exit codes: 0 on success (all retried or batch complete), 1 on error.

Environment Variables & Configuration

AWS Region

The CLI commands read AWS_REGION (or AWS_DEFAULT_REGION) from the environment. When constructing SDK clients in your own code, pass the region explicitly:

const snsClient = new SNSClient({ region: process.env.AWS_REGION ?? 'us-east-1' });
const sqsClient = new SQSClient({ region: process.env.AWS_REGION ?? 'us-east-1' });

AWS Credentials

The SDK uses standard AWS credential resolution (in order):

1. Environment variables: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
2. Shared credentials file (~/.aws/credentials)
3. IAM role attached to the execution environment (EC2, ECS, Lambda)

No SDK-specific credential configuration is needed.

Required IAM Permissions

For the publishing side (SnsSubscriber):

{
  "Effect": "Allow",
  "Action": ["sns:Publish"],
  "Resource": "arn:aws:sns:us-east-1:123456789012:your-topic"
}

For the processing side (EventProcessorCommand):

{
  "Effect": "Allow",
  "Action": [
    "sqs:ReceiveMessage",
    "sqs:DeleteMessage",
    "sqs:SendMessage",
    "sqs:GetQueueAttributes"
  ],
  "Resource": "arn:aws:sqs:us-east-1:123456789012:your-queue"
}

SQS Queue Configuration Recommendations

• Enable long polling (ReceiveMessageWaitTimeSeconds: 20) on the queue — the SDK already requests it per call.
• Set visibility timeout higher than your longest expected processing time to avoid duplicate processing.
• Configure a Dead Letter Queue (maxReceiveCount: 3–5) for unprocessable messages.

Logging

All SDK classes accept an optional Logger instance. The default is NullLogger (no output).

Built-in loggers

import { NullLogger, ConsoleLogger } from 'ts-domain-events';

// No output (default)
const quiet = new NullLogger();

// Logs to console.debug/info/warn/error
const verbose = new ConsoleLogger();

Custom logger (e.g., Winston)

import { Logger } from 'ts-domain-events';
import winston from 'winston';

class WinstonLogger implements Logger {
  constructor(private readonly logger: winston.Logger) {}

  debug(message: string, context?: Record<string, any>): void {
    this.logger.debug(message, context);
  }
  info(message: string, context?: Record<string, any>): void {
    this.logger.info(message, context);
  }
  warn(message: string, context?: Record<string, any>): void {
    this.logger.warn(message, context);
  }
  error(message: string, context?: Record<string, any>): void {
    this.logger.error(message, context);
  }
}

const subscriber = new SnsSubscriber(
  snsClient,
  topicArn,
  undefined,
  undefined,
  new WinstonLogger(winston.createLogger({ /* ... */ }))
);

Error Handling

SubscriberException

Thrown by SnsSubscriber when SNS publish fails and no failureRepository is configured.

import { SubscriberException } from 'ts-domain-events';

try {
  await subscriber.notify(event);
} catch (error) {
  if (error instanceof SubscriberException) {
    console.error('Subscriber failed:', error.message);
    console.error('Cause:', error.cause?.message);
  }
}

UnavailableException

Thrown by EventProcessor implementations to signal a temporary failure that should be retried with exponential backoff.

import { UnavailableException } from 'ts-domain-events';

// In your EventProcessor
async processEvent(event: Record<string, any>): Promise<void> {
  if (!this.service.isAvailable()) {
    throw new UnavailableException('Service temporarily unavailable');
  }
  // ...
}

ParsingException

Thrown by EventParser.parse() when the SQS message body is malformed, missing required fields, or contains invalid JSON.

import { EventParser, ParsingException } from 'ts-domain-events';

try {
  const event = EventParser.parse(messageBody);
} catch (error) {
  if (error instanceof ParsingException) {
    // Invalid message format — log and skip or send to DLQ manually
    console.error('Parse failed:', error.message);
  }
}

All custom exceptions expose a cause property holding the original error (if any) and set name to match the class name for reliable instanceof checks across module boundaries.

Migration from PHP Module

If your team is coming from researchsquare/domain-events-module (PHP), the concepts map directly:

| PHP | TypeScript | |---|---| | Publisher::instance() | Publisher.instance() | | Publisher::publish() | Publisher.instance().publish() | | Publisher::subscribe() | Publisher.instance().subscribe() | | Subscriber::notify() | Subscriber.notify() (abstract method) | | Subscriber::subscribesToEventType() | Subscriber.subscribesToEventType() (optional override) | | InMemorySubscriber::popEvents() | InMemorySubscriber.popEvents() | | SnsSubscriber | SnsSubscriber | | @SnsAttribute | @SnsAttribute decorator (requires reflect-metadata) | | DeferredSubscriber::onCommit() | DeferredSubscriber.onCommit() | | TypeOrmDeferredSubscriber | Not available — extend DeferredSubscriber manually | | PrismaDeferredSubscriber | PrismaDeferredSubscriber (with extension() + prisma.$extends()) | | EventProcessor::processEvent() | EventProcessor.processEvent() | | UnavailableException | UnavailableException | | FailoverPublisher | FailoverPublisher |

Key differences

• Decorators require setup: Import reflect-metadata at your entry point and enable experimentalDecorators + emitDecoratorMetadata in tsconfig.json.
• Async support: Subscriber.notify() and EventProcessor.processEvent() can return Promise<void>.
• AWS SDK v3: Clients are constructed separately and injected into SnsSubscriber and EventProcessorCommand — the SDK does not create clients internally.
• No TypeORM adapter: The TypeScript SDK does not ship a TypeOrmDeferredSubscriber. Extend DeferredSubscriber with TypeORM lifecycle hooks.
• Module format: The package ships both ESM and CommonJS. Most modern TypeScript setups will use ESM automatically.

Troubleshooting

AWS credential errors

CredentialsProviderError: Could not load credentials from any providers

Set credentials explicitly or ensure the execution environment has an IAM role:

export AWS_ACCESS_KEY_ID=AKIA...
export AWS_SECRET_ACCESS_KEY=...
export AWS_REGION=us-east-1

Or create ~/.aws/credentials:

[default]
aws_access_key_id = AKIA...
aws_secret_access_key = ...

reflect-metadata not found / decorators not working

You must import reflect-metadata once at your application entry point, before any decorated classes are loaded:

// app.ts  (entry point — must be first)
import 'reflect-metadata';
import { Publisher } from 'ts-domain-events';
// ...

Also verify tsconfig.json contains:

{
  "compilerOptions": {
    "experimentalDecorators": true,
    "emitDecoratorMetadata": true
  }
}

SNS attributes are empty

SnsSubscriber reads decorator metadata via reflect-metadata. If attributes are not appearing in SNS messages, confirm:

1. import 'reflect-metadata' is at your entry point.
2. emitDecoratorMetadata: true is in tsconfig.json.
3. Decorators are on the concrete class, not an interface or abstract base.

Events not being published

If notify() is never called on your subscriber, check that:

• Publisher.instance().subscribe(yourSubscriber) was called before publish().
• subscribesToEventType() returns the correct class name (exact match against event.constructor.name), or null to receive all events.
• You are not accidentally creating multiple Publisher instances — always use Publisher.instance().

SQS message visibility timeout issues

If you see the same message processed more than once (before max retries are reached):

• Increase your SQS queue's Visibility Timeout to be longer than your maximum processor execution time.
• The EventProcessorCommand uses 20-second long polling per receive call; ensure the timeout accounts for this.

Messages going to DLQ too quickly

If messages move to the DLQ before being fully retried via UnavailableException:

• Increase the SQS queue's maxReceiveCount (Redrive Policy) to be higher than your maximumRetryCount.
• UnavailableException manages its own retry counter in MessageAttributes, separate from SQS's ApproximateReceiveCount.

ParsingException on valid messages

SNS-to-SQS subscriptions must use raw message delivery disabled (the default). If raw delivery is enabled, the SNS envelope is stripped and EventParser.parse() will fail because it expects the {"Type":"Notification","Message":"..."} wrapper.

Development

Setup

npm install

Commands

# Build (ESM + CJS + type declarations)
npm run build

# Run all tests
npm test

# Watch mode
npm run test:watch

# Test coverage report
npm run test:coverage

# Type check (no emit)
npm run typecheck

# Lint
npm run lint

Project structure

src/
  core/           # Publisher, Subscriber, DomainEvent
  subscribers/    # InMemorySubscriber, SnsSubscriber, DeferredSubscriber
  adapters/       # PrismaDeferredSubscriber
  processor/      # EventProcessor, EventProcessorCommand, EventParser, EventFormatter
  failover/       # FailoverPublisher, InMemoryFailureRepository, failure interfaces
  exceptions/     # SubscriberException, UnavailableException, ParsingException
  utils/          # Logger, SnsAttribute, RetryCalculator
  cli/            # CLI commands (domain-events.ts, process-events.command.ts, retry-failures.command.ts)
bin/
  domain-events   # CLI entry point

License

MIT