dwn-dispatcher

A TypeScript Node.js package for sending push messages in conversational chatbots on the Blip platform. Provides a robust, multi-instance compliant system with automatic retries, scheduling, and shift-based time windows.

Features

• Channel-agnostic message dispatching (WhatsApp, Google Business Messages, etc.)
• Asynchronous message delivery with status monitoring
• Automatic retries with exponential backoff
• Multi-instance deployment support with distributed locking
• Shift-aware scheduling with timezone support
• Template-based messaging with payload filling
• Event publishing for monitoring and analytics

Installation

# npm
npm install @dawntech/dispatcher

# yarn
yarn add @dawntech/dispatcher

# pnpm
pnpm add @dawntech/dispatcher

Usage

Basic Example

import { Dispatcher, Descriptor } from '@dawntech/dispatcher';

// 1. Setup Infrastructure
const redisUrl = process.env.REDIS_URL || 'redis://localhost:6379';
const connection = {
  contract: 'my-blip-contract',
  key: 'my-blip-api-key',
};

const dispatcher = new Dispatcher('my-dispatcher-1', redisUrl, connection);
await dispatcher.setup();

Dispatcher ID Convention

The dispatcher ID supports the : character as a namespace separator. Redis tools (such as Redis Commander and Redis Insight) interpret : as a folder delimiter, creating a navigable tree structure for keys.

Use this convention to organize dispatchers hierarchically:

// Flat ID
const dispatcher = new Dispatcher('main', redisUrl, connection);
// Keys: dwn-dispatcher:main:message:..., dwn-dispatcher:main:queue:..., etc.

// Namespaced ID
const dispatcher = new Dispatcher('acme:onboarding', redisUrl, connection);
// Keys: dwn-dispatcher:acme:onboarding:message:..., dwn-dispatcher:acme:onboarding:queue:..., etc.

This results in the following folder structure in Redis GUI tools:

dwn-dispatcher:
├── acme:
│   └── onboarding:
│       ├── manifest         ← created on setup
│       ├── message:...
│       ├── index:...
│       ├── queue:...
│       └── metrics:...
└── main:
    ├── manifest
    ├── message:...
    ├── index:...
    ├── queue:...
    └── metrics:...

Every dispatcher creates a manifest key (Redis hash) on setup() containing metadata such as the package version and timestamps. This key is used to validate that a dispatcher actually exists.

// 2. Define a Descriptor (Message content + Event listeners)
const welcomeDescriptor = new Descriptor('welcome_message', (payload) => ({
  type: 'text/plain',
  content: `Welcome ${payload.name}!`,
}));

// Optional: listen to events for this specific message
welcomeDescriptor.on('delivered', (msg) => {
  console.log(`Message ${msg.messageId} was delivered!`);
});

// 3. Send message
await dispatcher.send(welcomeDescriptor, '[email protected]', {
  name: 'John Doe',
});

With Scheduling

await dispatcher.send('[email protected]', welcomeDescriptor, payload, {
  schedule: '2024-01-15T14:00:00Z',
});

With Shift Restrictions

await dispatcher.send('[email protected]', welcomeDescriptor, payload, {
  shifts: [
    {
      days: 31, // Mon-Fri (bitmask)
      start: '09:00',
      end: '18:00',
      gmt: '-3',
    },
  ],
});

With Extras (Passthrough Metadata)

Use extras to attach arbitrary metadata that is persisted in Redis and available in callbacks — without triggering any Blip API operations:

await dispatcher.send(descriptor, '[email protected]', payload, {
  meta: { key: 'order-123', webhook: 'https://my-service/delivery' },
});

dispatcher.on('delivered', (message) => {
  const { key, webhook } = (message.options?.meta ?? {}) as { key: string; webhook: string };
  // post delivery confirmation to webhook
});

This is useful when callbacks may fire after a process restart — the data is stored in Redis along with the message and survives across restarts.

Monitoring

You can monitor the Dispatcher metrics and receive alerts for failures or high load using DispatcherMonitor.

import { Dispatcher, DispatcherMonitor } from '@dawntech/dispatcher';

// 1. Create Dispatcher
const dispatcher = new Dispatcher('my-dispatcher', redisUrl, connection);
await dispatcher.setup();

// 2. Create Monitor attached to Dispatcher
const monitor = new DispatcherMonitor(dispatcher, {
  interval: 60000, // Check every minute
  rules: [
    {
      type: 'failure_rate',
      threshold: 0.05, // 5% failure rate
      window: 60000 * 60, // 1 hour window
      debounce: 30 * 60000, // Alert at most every 30 mins
    },
    {
      type: 'queue_size',
      threshold: 1000, // Alert if > 1000 messages pending/scheduled
      debounce: 10 * 60000, // Alert at most every 10 mins
    },
  ],
});

// 3. Listen for Alerts
monitor.on('alert', (alert) => {
  console.error(`[ALERT] ${alert.type}: ${alert.message}`, alert.details);
  // Send to external monitoring (e.g., Slack, PagerDuty, Datadog)
});

monitor.on('resolved', (alert) => {
  console.log(`[RESOLVED] ${alert.type}: ${alert.message}`);
});

// 4. Start Monitoring
monitor.start();

// ... application runs ...

// 5. Cleanup
monitor.stop();