Orchestration Core

One pipeline. Any platform. No lock-in.

@moderationos/orchestration-core — a platform-agnostic engine for event-driven automation. Open source under MIT. It's the core engine behind ModerationOS. Telegram is the transport it shipped with first, but it's one of several — Discord, Slack, WhatsApp, or your own webhook source all implement the same Transport contract. The core doesn't know or care which platform it's talking to, and nothing about it is specific to moderation.

Orchestration Core gives you a composable pipeline where every incoming event — a message, a join request, a button press, or something with no platform at all behind it — flows through a chain of stages you define. Each stage does one job: filter spam, look up a user record, check permissions, enqueue a task, route an order. When a stage is done, the next one picks up. The whole thing is testable, observable, and transport-agnostic.

Any platform ──► Transport ──► BotEngine ──► Pipeline
(Telegram,                                      │
 Discord, Slack,                        ┌───────┼───────┐
 your own webhook,                      ▼       ▼       ▼
 an in-memory queue, ...)            Stage1  Stage2  Stage3
                                                         │
                                                    ActionHandler

Table of Contents

• Features
• Requirements
• Installation
• Quick Start
• Project Structure
• Core Concepts
• Configuration
• Examples
• Running Tests
• Documentation
• License

Features

• Pipeline architecture — chain middleware stages that each do one thing well
• Platform-agnostic by contract — every transport implements the same Transport interface; swap Telegram for Discord, Slack, or your own webhook source without touching a stage
• Checked, not duck-typed — a transport that claims a capability it doesn't implement fails loudly at startup, not silently at call time
• Early termination — any stage can halt the pipeline with { stop: true }
• Action pattern — stages declare intent (notify_admin, ban_user); a central handler executes it
• Hook system — observe every stage lifecycle event for logging, metrics, and debugging
• Structured error recovery — per-stage strategies: stop, skip, retry, or fallback
• Multi-transport routing — run a single pipeline across several transports simultaneously (e.g. Telegram's Bot API + MTProto, or any mix of your own)
• Transparent caching — wrap any DB adapter with automatic, per-table-scoped cache invalidation
• No forced dependencies — node-telegram-bot-api, pg, and telegram (gramjs) are optional peers, installed only if you use that adapter; the core and test suite need none of them

Requirements

• Node.js ≥ 16
• The core itself needs nothing else. Everything below is optional and only needed if you use that specific piece:
  • A Telegram Bot Token (get one from @BotFather) — only if you use TelegramAdapter
  • apiId/apiHash from my.telegram.org — only if you use MTProtoAdapter
  • PostgreSQL — only if you use PostgreSQLAdapter
  • Redis (or any cache with get/set) — only if you use CacheAdapter
  • Nothing at all for HttpWebhookTransport — it runs on Node builtins alone

Installation

# npm
npm install @moderationos/orchestration-core

# yarn
yarn add @moderationos/orchestration-core

# pnpm
pnpm add @moderationos/orchestration-core

# bun
bun add @moderationos/orchestration-core

Then import in your project:

import { BotEngine, Pipeline, TelegramAdapter } from '@moderationos/orchestration-core';

Using TelegramAdapter, MTProtoAdapter, or PostgreSQLAdapter? Install their peer package yourself — they're optional, lazy-imported, and not bundled:

npm install node-telegram-bot-api   # for TelegramAdapter
npm install telegram                # for MTProtoAdapter (GramJS)
npm install pg                      # for PostgreSQLAdapter

Using only HttpWebhookTransport, or writing your own transport? Nothing else to install — the core runs on Node builtins alone.

Quick Start

1. Create your bot file

// bot.js
import {
  BotEngine,
  Pipeline,
  TelegramAdapter,
} from '@moderationos/orchestration-core';

// --- Define your pipeline stages ---

async function logMessage(message, context) {
  context.logger.info(`[${message.conversation.id}] ${message.author?.displayName}: ${message.text}`);
}

async function blockBadWords(message, context) {
  const banned = ['spam', 'scam'];
  if (banned.some(w => message.text?.toLowerCase().includes(w))) {
    await context.bot.deleteMessage(message.conversation.id, message.id);
    return { stop: true, reason: 'bad_word' };
  }
}

async function greetNewMembers(message, context) {
  if (message.kind === 'membership.joined') {
    const who = message.target?.participant?.displayName ?? 'friend';
    await context.bot.sendMessage(message.conversation.id, `👋 Welcome, ${who}!`);
    return { stop: true };
  }
}

// --- Wire it together ---

const adapter  = new TelegramAdapter(process.env.BOT_TOKEN);
const pipeline = new Pipeline();

pipeline
  .use(logMessage)
  .use(greetNewMembers)
  .use(blockBadWords);

const engine = new BotEngine(adapter, { pipeline });
await engine.start();

console.log('Bot is running...');

2. Set your token and run

BOT_TOKEN=your_token_here node bot.js

That's it. Messages flow through your three stages in order.

Not a Telegram bot: generic workflow automation

Orchestration Core isn't Telegram-specific, and it isn't moderation-specific either — it's a general event pipeline. This example uses HttpWebhookTransport (Node builtins only, no Telegram involved) to route incoming webhook events, e.g. order-processing:

import { BotEngine, Pipeline, HttpWebhookTransport } from '@moderationos/orchestration-core';

async function validateOrder(message, context) {
  if (message.kind !== 'message') return;
  const order = JSON.parse(message.text);
  if (!order.sku) return { stop: true, reason: 'missing_sku' };
  context.state.order = order;
}

async function enqueueFulfillment(message, context) {
  await context.db?.insert('fulfillment_queue', context.state.order);
  return { action: 'notify_warehouse', data: context.state.order };
}

const transport = new HttpWebhookTransport({ port: 3000 });
const pipeline  = new Pipeline().use(validateOrder).use(enqueueFulfillment);

const engine = new BotEngine(transport, { pipeline });
await engine.start();

console.log(`Listening for webhook events on port ${transport.port}`);

Same BotEngine, same Pipeline, same stage signature — no moderation concept anywhere in sight. This is the point of the Transport contract: the engine doesn't know or care that this transport isn't Telegram.

Project Structure

src/
├── index.js                        # Main entry point — re-exports everything
├── core/
│   ├── BotEngine.js                # Wires transport → pipeline, handles lifecycle
│   ├── BotManager.js               # Manages multiple bot instances dynamically
│   ├── Pipeline.js                 # Runs stages sequentially, handles stop/retry
│   ├── HookManager.js              # Lifecycle event hooks (before/after each stage)
│   ├── ErrorHandler.js             # Per-stage error recovery strategies
│   └── ActionHandler.js            # Dispatches actions returned by stages
└── adapters/
    ├── transports/
    │   ├── Transport.js            # Abstract base contract + capability checker
    │   ├── permissions.js          # Neutral moderation-permission shape (allowAll/denyAll)
    │   ├── TelegramAdapter.js      # Telegram Bot API (polling)
    │   ├── MTProtoAdapter.js       # Telegram MTProto (user-level access)
    │   ├── HttpWebhookTransport.js # Dependency-free reference transport (node:http)
    │   └── CompositeTransport.js   # Fan-out across multiple transports (was TransportAdapter)
    └── databases/
        ├── PostgreSQLAdapter.js    # PostgreSQL with connection pooling
        └── CacheAdapter.js         # Transparent caching layer for any DB adapter

tests/
├── helpers/                        # Shared mocks, a Telegram-free FakeTransport, the
│                                    # reusable conformance suite every transport passes
├── pipeline.test.js
├── hookmanager.test.js
├── errorhandler.test.js
├── actionhandler.test.js
├── botmanager.test.js
├── cacheadapter.test.js
├── transportadapter.test.js        # CompositeTransport (incl. the deprecated alias)
├── transport-conformance.test.js   # One suite run against every transport
├── transport-contract.test.js      # assertTransportContract enforcement
├── httpwebhook.test.js             # Real socket I/O against HttpWebhookTransport
└── moderation-translation.test.js  # Permission/ban translation correctness

docs/
├── architecture.md                 # How everything fits together
├── api.md                          # Full API reference
└── guides/
    ├── writing-stages.md           # How to write pipeline stages
    └── adapters.md                 # Working with transport and DB adapters

MIGRATION.md                        # v1 -> v2 breaking-change list

Core Concepts

Pipeline Stages

A stage is just an async function. It receives the normalized message and a context object, does its work, and optionally returns a result.

async function myStage(message, context) {
  // context.bot    → send messages, ban users, etc.
  // context.db     → database adapter (if configured)
  // context.logger → structured logger
  // context.config → your config object
  // context.state  → shared scratch space for this message
}

Return nothing to continue. Return { stop: true } to halt. Return { action: 'name', data: {} } to dispatch a side effect.

The Context Object

Every stage receives the same context for a given message:

| Property | Type | Description | |---|---|---| | bot | Adapter | The transport adapter — use it to send replies | | db | Adapter | null | Database adapter, if provided to BotEngine | | logger | Logger | Structured logger (pino/winston compatible) | | config | Object | Your configuration, passed to BotEngine options | | state | Object | Empty object — stages can share data via this |

Message Format

Every transport normalizes its native events into one neutral envelope before the pipeline sees them. Nothing platform-specific leaks past this boundary — anything genuinely native lives under raw.

{
  kind:         string,        // neutral event kind — 'message' | 'message.edited'
                               // | 'membership.joined' | 'membership.left'
                               // | 'membership.join_request' | 'interaction.action' | ...
  id:           string,        // opaque event/message id (always a string)
  conversation: { id: string, type, title },  // type: 'direct' | 'group' | 'channel' | 'unknown'
  author:       { id, displayName, handle } | null,
  text:         string,
  spans:        [{ type, start, length, value? }],  // neutral rich-text ranges (replaces entities)
  timestamp:    Date,
  target:       Object | null, // kind-specific data (e.g. { participant } for membership,
                               // { messageId, payload, interactionId } for interaction.action)
  source:       string,        // transport name (multi-transport setups)
  raw:          unknown,       // the untouched native payload — the only escape hatch
}

Migrating from v1? type → kind (new value set), chat → conversation, from → author (displayName/handle), entities → spans, ids are now strings, and callback/join data moved under target. See the full breaking-change list in MIGRATION.md.

Configuration

Pass a config object to BotEngine and it will be available as context.config in every stage:

const engine = new BotEngine(adapter, {
  pipeline,
  db,
  config: {
    adminChatId:    -100123456789,
    warnThreshold:  3,
    allowedDomains: ['example.com'],
  },
  logger: pinoLogger,  // any logger with .info/.warn/.error/.debug
});

Examples

Auto-approve join requests

async function approveJoinRequests(message, context) {
  if (message.kind !== 'membership.join_request') return;

  await context.bot.approveChatJoinRequest(message.conversation.id, message.author.id);
  return { stop: true };
}

Track users in PostgreSQL

async function upsertUser(message, context) {
  if (!message.author) return;

  await context.db.query(
    `INSERT INTO users (id, handle, display_name)
     VALUES ($1, $2, $3)
     ON CONFLICT (id) DO UPDATE
     SET handle = $2, display_name = $3, last_seen = NOW()`,
    [message.author.id, message.author.handle, message.author.displayName]
  );
}

Return an action instead of executing a side effect

// Stage declares intent
async function flagSpammer(message, context) {
  if (isSpam(message.text)) {
    return {
      action: 'ban_and_notify',
      data: { conversationId: message.conversation.id, userId: message.author.id },
    };
  }
}

// ActionHandler executes it
actionHandler.register('ban_and_notify', async ({ conversationId, userId }, context) => {
  await context.bot.banMember(conversationId, userId);
  await context.bot.sendMessage(context.config.adminChatId, `Banned ${userId}`);
});

Add error recovery to a risky stage

import { ErrorHandler } from '@moderationos/orchestration-core';

const errorHandler = new ErrorHandler(logger);

// Retry db lookups up to 3 times, skip on validation failures
errorHandler.registerRecoveryStrategy('fetchUserData', 'retry', { maxRetries: 3, backoffMs: 200 });
errorHandler.registerRecoveryStrategy('validateInput', 'skip');

pipeline.setErrorHandler(errorHandler);

Run two transports through one pipeline

import { CompositeTransport, TelegramAdapter, MTProtoAdapter } from '@moderationos/orchestration-core';

const botApi  = new TelegramAdapter(process.env.BOT_TOKEN);
const mtproto = new MTProtoAdapter({ apiId, apiHash, sessionString });

const transport = new CompositeTransport([botApi, mtproto]);
const engine    = new BotEngine(transport, { pipeline });

CompositeTransport was previously named TransportAdapter. That name is still exported as a deprecated alias, so existing TransportAdapter imports keep working.

Running Tests

npm test

The test suite uses Node's built-in node:test runner with no additional dependencies — no pg, node-telegram-bot-api, or telegram install needed to run it. Almost every test is a pure unit test against mocks or fakes; a handful (httpwebhook.test.js) do real localhost socket I/O against HttpWebhookTransport, with no external service involved. No live Telegram, Postgres, or Redis connection is ever required.

Current pass/fail status: see the Tests badge above, or the Actions tab — it runs on every push and pull request against Node 20/22/24, so the number here would just go stale.

Documentation

Full documentation lives in the docs/ folder:

• Architecture Overview — how all the pieces connect
• Writing Stages Guide — practical patterns for building middleware
• Adapters Guide — transport and database adapters in depth
• API Reference — complete method signatures for every class

License

MIT