@message-queue-toolkit/core

Core library for message-queue-toolkit. Provides foundational abstractions, utilities, and base classes for building message queue publishers and consumers.

Table of Contents

• Installation
• Overview
• Core Concepts
  • Message Schemas
  • Message Type Resolution
  • Handler Configuration
  • Pre-handlers and Barriers
  • Handler Spies
• Key Classes
  • AbstractQueueService
  • MessageHandlerConfigBuilder
  • HandlerContainer
  • MessageSchemaContainer
  • AbstractPublisherManager
  • DomainEventEmitter
• Utilities
  • Error Classes
  • Message Deduplication
  • Payload Offloading
• API Reference
• Links

Installation

npm install @message-queue-toolkit/core zod

Peer Dependencies:

• zod - Schema validation

Overview

The core package provides the foundational building blocks used by all protocol-specific implementations (SQS, SNS, AMQP, Kafka, GCP Pub/Sub). It includes:

• Base Classes: Abstract classes for publishers and consumers
• Handler System: Type-safe message routing and handling
• Validation: Zod schema validation and message parsing
• Utilities: Retry logic, date handling, environment utilities
• Testing: Handler spies for testing async message flows
• Extensibility: Interfaces for payload stores, deduplication stores, and metrics

Core Concepts

Message Schemas

Messages are validated using Zod schemas. The library uses configurable field names:

• messageTypeResolver: Configuration for resolving the message type discriminator (see Message Type Resolution)
• messageIdField (default: 'id'): Field containing the message ID
• messageTimestampField (default: 'timestamp'): Field containing the timestamp

import { z } from 'zod'

const UserCreatedSchema = z.object({
  id: z.string(),
  type: z.literal('user.created'),  // Used for routing
  timestamp: z.string().datetime(),
  userId: z.string(),
  email: z.string().email(),
})

type UserCreated = z.infer<typeof UserCreatedSchema>

Message Type Resolution

What is Message Type?

The message type is a discriminator field that identifies what kind of event or command a message represents. It's used for:

1. Routing: Directing messages to the appropriate handler based on their type
2. Schema validation: Selecting the correct Zod schema to validate the message
3. Observability: Tracking metrics and logs per message type

In a typical event-driven architecture, a single queue or topic may receive multiple types of messages. For example, a user-events queue might receive user.created, user.updated, and user.deleted events. The message type tells the consumer which handler should process each message.

Configuration Options

The messageTypeResolver configuration supports three modes:

Mode 1: Field Path (Simple)

Use when the message type is a field in the parsed message body. Supports dot notation for nested paths:

{
  messageTypeResolver: { messageTypePath: 'type' },  // Extracts from message.type
}

// Nested path example
{
  messageTypeResolver: { messageTypePath: 'metadata.eventType' },  // Extracts from message.metadata.eventType
}

Mode 2: Literal (Constant)

Use when all messages are of the same type:

{
  messageTypeResolver: { literal: 'order.created' },  // All messages treated as this type
}

Mode 3: Custom Resolver (Flexible)

Use for complex scenarios where the type needs to be extracted from message attributes, nested fields, or requires transformation:

import type { MessageTypeResolverConfig } from '@message-queue-toolkit/core'

const resolverConfig: MessageTypeResolverConfig = {
  resolver: ({ messageData, messageAttributes }) => {
    // Your custom logic here
    return 'resolved.type'
  },
}

Important: The resolver function must always return a valid string. If the type cannot be determined, either return a default type or throw an error with a descriptive message.

Real-World Examples by Platform

AWS SQS (Plain)

When publishing your own events directly to SQS, you control the message format:

// Message format you control
{
  "id": "msg-123",
  "type": "order.created",  // Your type field
  "timestamp": "2024-01-15T10:30:00Z",
  "payload": {
    "orderId": "order-456",
    "amount": 99.99
  }
}

// Configuration
{
  messageTypeResolver: { messageTypePath: 'type' },
}

AWS EventBridge → SQS

EventBridge events have a specific structure with detail-type:

// EventBridge event structure delivered to SQS
{
  "version": "0",
  "id": "12345678-1234-1234-1234-123456789012",
  "detail-type": "Order Created",  // EventBridge uses detail-type
  "source": "com.myapp.orders",
  "account": "123456789012",
  "time": "2024-01-15T10:30:00Z",
  "region": "us-east-1",
  "detail": {
    "orderId": "order-456",
    "amount": 99.99
  }
}

// Configuration
{
  messageTypeResolver: { messageTypePath: 'detail-type' },
}

// Or with resolver for normalization
{
  messageTypeResolver: {
    resolver: ({ messageData }) => {
      const data = messageData as { 'detail-type'?: string; source?: string }
      const detailType = data['detail-type']
      if (!detailType) throw new Error('detail-type is required')
      // Optionally normalize: "Order Created" → "order.created"
      return detailType.toLowerCase().replace(/ /g, '.')
    },
  },
}

AWS SNS → SQS

SNS messages wrapped in SQS have the actual payload in the Message field (handled automatically by the library after unwrapping):

// After SNS envelope unwrapping, you get your original message
{
  "id": "msg-123",
  "type": "user.signup.completed",
  "userId": "user-789",
  "email": "[email protected]"
}

// Configuration
{
  messageTypeResolver: { messageTypePath: 'type' },
}

Apache Kafka

Kafka typically uses topic-based routing, but you may still need message types within a topic:

// Kafka message value (JSON)
{
  "eventType": "inventory.reserved",
  "eventId": "evt-123",
  "timestamp": 1705312200000,
  "data": {
    "sku": "PROD-001",
    "quantity": 5
  }
}

// Configuration
{
  messageTypeResolver: { messageTypePath: 'eventType' },
}

// Or using Kafka headers (via custom resolver)
{
  messageTypeResolver: {
    resolver: ({ messageData, messageAttributes }) => {
      // Kafka headers are passed as messageAttributes
      if (messageAttributes?.['ce_type']) {
        return messageAttributes['ce_type'] as string  // CloudEvents header
      }
      const data = messageData as { eventType?: string }
      if (!data.eventType) throw new Error('eventType required')
      return data.eventType
    },
  },
}

Google Cloud Pub/Sub (Your Own Events)

When you control the message format in Pub/Sub:

// Your message (base64-decoded from data field)
{
  "type": "payment.processed",
  "paymentId": "pay-123",
  "amount": 150.00,
  "currency": "USD"
}

// Configuration
{
  messageTypeResolver: { messageTypePath: 'type' },
}

Google Cloud Pub/Sub (Cloud Storage Notifications)

Cloud Storage notifications put the event type in message attributes, not the data payload:

// Pub/Sub message structure for Cloud Storage notifications
{
  "data": "eyJraW5kIjoic3RvcmFnZSMgb2JqZWN0In0=",  // Base64-encoded object metadata
  "attributes": {
    "eventType": "OBJECT_FINALIZE",  // Type is HERE, not in data!
    "bucketId": "my-bucket",
    "objectId": "path/to/file.jpg",
    "objectGeneration": "1705312200000"
  },
  "messageId": "123456789",
  "publishTime": "2024-01-15T10:30:00Z"
}

// Configuration - must use resolver to access attributes
{
  messageTypeResolver: {
    resolver: ({ messageAttributes }) => {
      const eventType = messageAttributes?.eventType as string
      if (!eventType) {
        throw new Error('eventType attribute required for Cloud Storage notifications')
      }
      // Map GCS event types to your internal types
      const typeMap: Record<string, string> = {
        'OBJECT_FINALIZE': 'storage.object.created',
        'OBJECT_DELETE': 'storage.object.deleted',
        'OBJECT_ARCHIVE': 'storage.object.archived',
        'OBJECT_METADATA_UPDATE': 'storage.object.metadataUpdated',
      }
      return typeMap[eventType] ?? eventType
    },
  },
}

Google Cloud Pub/Sub (Eventarc / CloudEvents)

Eventarc delivers events in CloudEvents format:

// CloudEvents structured format
{
  "specversion": "1.0",
  "type": "google.cloud.storage.object.v1.finalized",  // CloudEvents type
  "source": "//storage.googleapis.com/projects/_/buckets/my-bucket",
  "id": "1234567890",
  "time": "2024-01-15T10:30:00Z",
  "datacontenttype": "application/json",
  "data": {
    "bucket": "my-bucket",
    "name": "path/to/file.jpg",
    "contentType": "image/jpeg"
  }
}

// Configuration
{
  messageTypeResolver: { messageTypePath: 'type' },  // CloudEvents type is at root level
}

// Or with mapping to simpler types
{
  messageTypeResolver: {
    resolver: ({ messageData }) => {
      const data = messageData as { type?: string }
      const ceType = data.type
      if (!ceType) throw new Error('CloudEvents type required')
      // Map verbose CloudEvents types to simpler names
      if (ceType.includes('storage.object') && ceType.includes('finalized')) {
        return 'storage.object.created'
      }
      if (ceType.includes('storage.object') && ceType.includes('deleted')) {
        return 'storage.object.deleted'
      }
      return ceType
    },
  },
}

Single-Type Queues (Any Platform)

When a queue/subscription only ever receives one type of message, use literal:

// Dedicated queue for order.created events only
{
  messageTypeResolver: {
    literal: 'order.created',
  },
}

This is useful for:

• Dedicated queues/subscriptions filtered to a single event type
• Legacy systems where messages don't have a type field
• Simple integrations where you know exactly what you're receiving

Handler Configuration

Use MessageHandlerConfigBuilder to configure handlers for different message types:

import { MessageHandlerConfigBuilder } from '@message-queue-toolkit/core'

const handlers = new MessageHandlerConfigBuilder<SupportedMessages, ExecutionContext>()
  .addConfig(
    UserCreatedSchema,
    async (message, context, preHandlingOutputs) => {
      await context.userService.createUser(message.userId)
      return { result: 'success' }
    }
  )
  .addConfig(
    UserUpdatedSchema,
    async (message, context, preHandlingOutputs) => {
      await context.userService.updateUser(message.userId, message.changes)
      return { result: 'success' }
    }
  )
  .build()

Pre-handlers and Barriers

Pre-handlers are middleware functions that run before the main handler:

const preHandlers = [
  (message, context, output, next) => {
    // Enrich context, validate prerequisites, etc.
    output.logger = context.logger.child({ messageId: message.id })
    next({ result: 'success' })
  },
]

Barriers control whether a message should be processed or retried later:

const preHandlerBarrier = async (message, context, preHandlerOutput) => {
  const prerequisiteMet = await checkPrerequisite(message)
  return {
    isPassing: prerequisiteMet,
    output: { ready: true },
  }
}

Handler Spies

Handler spies enable testing of async message flows:

// Enable in consumer/publisher options
{ handlerSpy: true }

// Wait for specific messages in tests
const result = await consumer.handlerSpy.waitForMessageWithId('msg-123', 'consumed', 5000)
expect(result.userId).toBe('user-456')

Key Classes

AbstractQueueService

Base class for all queue services. Provides:

• Message serialization/deserialization
• Schema validation
• Retry logic with exponential backoff
• Payload offloading support
• Message deduplication primitives

MessageHandlerConfigBuilder

Fluent builder for configuring message handlers:

import { MessageHandlerConfigBuilder } from '@message-queue-toolkit/core'

const handlers = new MessageHandlerConfigBuilder<
  SupportedMessages,
  ExecutionContext,
  PrehandlerOutput
>()
  .addConfig(Schema1, handler1)
  .addConfig(Schema2, handler2, {
    preHandlers: [preHandler1, preHandler2],
    preHandlerBarrier: barrierFn,
    messageLogFormatter: (msg) => ({ id: msg.id }),
  })
  .build()

Handler Configuration Options

The third parameter to addConfig accepts these options:

| Option | Type | Description | |--------|------|-------------| | messageType | string | Explicit message type for routing. Required when using custom resolver. | | messageLogFormatter | (message) => unknown | Custom formatter for logging | | preHandlers | Prehandler[] | Middleware functions run before the handler | | preHandlerBarrier | BarrierCallback | Barrier function for out-of-order message handling |

Explicit Message Type

When using a custom resolver function (messageTypeResolver: { resolver: fn }), the message type cannot be automatically extracted from schemas at registration time. You must provide an explicit messageType for each handler:

const handlers = new MessageHandlerConfigBuilder<SupportedMessages, Context>()
  .addConfig(
    STORAGE_OBJECT_SCHEMA,
    handleObjectCreated,
    { messageType: 'storage.object.created' }  // Required for custom resolver
  )
  .addConfig(
    STORAGE_DELETE_SCHEMA,
    handleObjectDeleted,
    { messageType: 'storage.object.deleted' }  // Required for custom resolver
  )
  .build()

const container = new HandlerContainer({
  messageHandlers: handlers,
  messageTypeResolver: {
    resolver: ({ messageAttributes }) => {
      // Map external event types to your internal types
      const eventType = messageAttributes?.eventType as string
      if (eventType === 'OBJECT_FINALIZE') return 'storage.object.created'
      if (eventType === 'OBJECT_DELETE') return 'storage.object.deleted'
      throw new Error(`Unknown event type: ${eventType}`)
    },
  },
})

Priority for determining handler message type:

1. Explicit messageType in handler options (highest priority)
2. Literal type from messageTypeResolver: { literal: 'type' }
3. Extract from schema's literal field using messageTypePath

If the message type cannot be determined, an error is thrown during container construction.

HandlerContainer

Routes messages to appropriate handlers based on message type:

import { HandlerContainer } from '@message-queue-toolkit/core'

const container = new HandlerContainer({
  messageHandlers: handlers,
  messageTypeResolver: { messageTypePath: 'type' },
})

const handler = container.resolveHandler(message.type)

MessageSchemaContainer

Manages Zod schemas and validates messages:

import { MessageSchemaContainer } from '@message-queue-toolkit/core'

const container = new MessageSchemaContainer({
  messageSchemas: [{ schema: Schema1 }, { schema: Schema2 }],
  messageDefinitions: [],
  messageTypeResolver: { messageTypePath: 'type' },
})

const result = container.resolveSchema(message)
if ('error' in result) {
  // Handle error
} else {
  const schema = result.result
}

AbstractPublisherManager

Factory pattern for spawning publishers on demand:

import { AbstractPublisherManager } from '@message-queue-toolkit/core'

// Automatically spawns publishers and fills metadata
await publisherManager.publish('user-events-topic', {
  type: 'user.created',
  userId: 'user-123',
})

DomainEventEmitter

Event emitter for domain events:

import { DomainEventEmitter } from '@message-queue-toolkit/core'

const emitter = new DomainEventEmitter()

emitter.on('user.created', async (event) => {
  console.log('User created:', event.userId)
})

await emitter.emit('user.created', { userId: 'user-123' })

Utilities

Error Classes

import {
  MessageValidationError,
  MessageInvalidFormatError,
  DoNotProcessMessageError,
  RetryMessageLaterError,
} from '@message-queue-toolkit/core'

// Validation failed
throw new MessageValidationError(zodError.issues)

// Message format is invalid (cannot parse)
throw new MessageInvalidFormatError({ message: 'Invalid JSON' })

// Do not process this message (skip without retry)
throw new DoNotProcessMessageError({ message: 'Duplicate message' })

// Retry this message later
throw new RetryMessageLaterError({ message: 'Dependency not ready' })

Message Deduplication

Interfaces for implementing deduplication stores:

import type { MessageDeduplicationStore, ReleasableLock } from '@message-queue-toolkit/core'

// Implement custom deduplication store
class MyDeduplicationStore implements MessageDeduplicationStore {
  async keyExists(key: string): Promise<boolean> { /* ... */ }
  async setKey(key: string, ttlSeconds: number): Promise<void> { /* ... */ }
  async acquireLock(key: string, options: AcquireLockOptions): Promise<ReleasableLock> { /* ... */ }
}

Payload Offloading

Interfaces for implementing payload stores:

import type { PayloadStore, PayloadStoreConfig } from '@message-queue-toolkit/core'

// Implement custom payload store
class MyPayloadStore implements PayloadStore {
  async storePayload(payload: Buffer, messageId: string): Promise<PayloadRef> { /* ... */ }
  async retrievePayload(ref: PayloadRef): Promise<Buffer> { /* ... */ }
}

API Reference

Types

// Handler result type
type HandlerResult = Either<'retryLater', 'success'>

// Pre-handler signature
type Prehandler<Message, Context, Output> = (
  message: Message,
  context: Context,
  output: Output,
  next: (result: PrehandlerResult) => void
) => void

// Barrier signature
type BarrierCallback<Message, Context, PrehandlerOutput, BarrierOutput> = (
  message: Message,
  context: Context,
  preHandlerOutput: PrehandlerOutput
) => Promise<BarrierResult<BarrierOutput>>

// Barrier result
type BarrierResult<Output> =
  | { isPassing: true; output: Output }
  | { isPassing: false; output?: never }

// Message type resolver context
type MessageTypeResolverContext = {
  messageData: unknown
  messageAttributes?: Record<string, unknown>
}

// Message type resolver function
type MessageTypeResolverFn = (context: MessageTypeResolverContext) => string

// Message type resolver configuration
type MessageTypeResolverConfig =
  | { messageTypePath: string }  // Extract from field at root of message data
  | { literal: string }          // Constant type for all messages
  | { resolver: MessageTypeResolverFn }  // Custom resolver function

Utility Functions

// Environment utilities
isProduction(): boolean
reloadConfig(): void

// Date utilities
isRetryDateExceeded(timestamp: string | Date, maxRetryDuration: number): boolean

// Message parsing
parseMessage<T>(data: unknown, schema: ZodSchema<T>): ParseMessageResult<T>

// Wait utilities
waitAndRetry<T>(fn: () => Promise<T>, options: WaitAndRetryOptions): Promise<T>

// Object utilities
objectMatches(obj: unknown, pattern: unknown): boolean
isShallowSubset(subset: object, superset: object): boolean

Links

• Main Repository
• SQS Package
• SNS Package
• AMQP Package
• GCP Pub/Sub Package
• Kafka Package