@liam-public/node-messaging

Node.js messaging orchestration and channel plugins. A transport-neutral core — a plugin contract, a registry, outbound delivery, and inbound normalization — plus fifteen channel plugins that implement the contract.

The core never imports a channel SDK. Everything channel-specific lives behind ChannelPlugin, so the delivery and inbound paths are written once and a new transport is a new plugin rather than a new branch in shared code.

Exports

Core

• ChannelPluginRegistry — registers plugins by id and resolves them for delivery.
• ChannelPlugin and the adapter types (ChannelOutboundAdapter, ChannelGatewayAdapter, ChannelMessageActionAdapter, …) — the contract a channel implements.
• MessagingConfig, ChannelId, MessagingChannelRuntime — config and host-callback shapes.

Outbound

• deliverOutboundPayloads(params) — chunk, then send text and media through a channel's outbound adapter.
• broadcastMessage(params) — the same delivery across many targets, isolating failures.
• chunkText / chunkByParagraph / chunkMarkdown — three split strategies (below).
• resolveLocalMediaPath / resolveOutboundMediaAsBase64 — containment for outbound media.
• humanDelay(config, abortSignal?) — randomized pause between consecutive replies.

Inbound

• dispatchInboundMessage(options) — dedupe, normalize, hand to your handler, drain replies.
• finalizeInboundContext(ctx) — newline and system-tag normalization.
• shouldSkipDuplicateInbound(ctx) / buildDedupeKey / resetDedupe — redelivery suppression.
• createReplyDispatcher(options) — ordered, back-pressured reply queue.
• normalizeInboundTextNewlines / sanitizeInboundSystemTags — the two text transforms.

Channel plugins

telegramPlugin, slackPlugin, whatsappPlugin, zaloPlugin, discordPlugin, signalPlugin, ircPlugin, imessagePlugin, linePlugin, matrixPlugin, mattermostPlugin, blueBubblesPlugin, synologyChatPlugin, googlechatPlugin, msteamsPlugin.

The plugin contract is mostly optional on purpose

A ChannelPlugin requires only four fields — id, meta, capabilities, and config. Every behavioural adapter (outbound, gateway, inbound wiring, threading, directory, actions, setupWizard, …) is optional.

That is deliberate. The transports differ enormously in what they can do: IRC has no threads, Signal has no reactions API worth the name, Discord has a component model nothing else shares. A contract that demanded all of it would force every plugin to stub methods it cannot honour, and callers would have no way to tell a real implementation from a stub. Instead, absence is the signal — deliverOutboundPayloads throws Channel <id> has no outbound adapter rather than silently doing nothing, and capabilities advertises statically what the plugin supports so a caller can check before it tries.

ChannelId is an open union ("telegram" | … | (string & {})), so a consumer can register a channel this package has never heard of and still get the literal-type autocomplete for the built-ins.

Registration is first-write-wins

const registry = new ChannelPluginRegistry()
registry.register(telegramPlugin)
registry.register(myTelegramFork)   // ignored — 'telegram' is taken

register silently ignores a duplicate id rather than throwing or overwriting. A host that loads plugins from several sources (built-ins, then user config) can therefore register defensively, and ordering expresses precedence: register your overrides before the built-ins. list() returns plugins sorted by meta.order (default 999) then id, which is the order setup and picker UIs display.

Outbound delivery

const results = await deliverOutboundPayloads({
  registry,
  channelId: 'telegram',
  to: '@someone',
  text: longMarkdownReply,
  mediaUrl: '/srv/agent-uploads/chart.png',
  mediaLocalRoots: ['/srv/agent-uploads'],
  cfg,
})

Text is split with the channel's own chunker at its own textChunkLimit, falling back to chunkText at 4000 characters. The fallback is deliberately below Telegram's 4096 and far below Slack's limit: a chunk that is rejected for length costs a round trip and arrives out of order behind its successors, so the default errs small and each plugin raises it to what its transport actually allows.

mediaUrl rides on the first chunk, via sendMedia. Media is the caption-bearing part of the message on most transports, and attaching it to the first chunk keeps the caption with the opening text rather than orphaning it after a wall of continuation messages. Additional mediaUrls are sent after all text, one call each.

bestEffort: true swaps throw-on-first-failure for onError(index, error) and continues. Use it when a partial reply beats no reply — a five-chunk answer losing chunk three is still worth sending. Leave it off when the message is atomic. abortSignal is checked between sends, so an abort stops the remaining chunks without cancelling one in flight.

broadcastMessage isolates failures per target

Broadcast sends sequentially and collects { results, failures } rather than rejecting: one dead target must not suppress delivery to the rest, and the caller needs to know which ones missed.

mediaLocalRoots is forwarded to every target, so the containment rules below apply to a broadcast exactly as they do to a single send — including failing closed when the roots are unset. A refused target lands in failures with the reason; the rest still deliver.

Outbound media fails closed

resolveLocalMediaPath is the reason mediaLocalRoots exists, and it is the one default in this package that is restrictive rather than convenient.

On an agent-driven channel, a reply payload's mediaUrl is model output. An unchecked path turns the bot's own media upload into a file-exfiltration primitive: anyone who can influence the model's reply names /root/.ssh/id_rsa and has it delivered into the chat thread. So:

• No configured roots means no local media, not "all paths allowed". An operator who has not thought about this gets a refusal with a message naming the setting, not a silent hole.
• Candidate and roots are both resolved through realpath, so a symlink planted inside an allowed root cannot step outside it.
• Containment is checked with path.relative, not a string prefix — /srv/uploads-private is not a child of /srv/uploads.
• Remote URLs are rejected outright. Download first, then send the local file; the alternative is an SSRF surface reachable from model output.

Enforcement lives here rather than in each plugin so that a new channel cannot forget it. resolveOutboundMediaAsBase64 passes data: URIs through untouched — they are already content, so no filesystem is involved — and sends everything else through the same check.

Inbound

await dispatchInboundMessage({
  ctx,
  dispatcher: createReplyDispatcher({ deliver, humanDelay: { minMs: 400, maxMs: 1200 } }),
  onMessage: async (finalized, dispatcher) => {
    dispatcher.sendFinalReply({ text: await runAgent(finalized.bodyForAgent) })
  },
})

dispatchInboundMessage dedupes, finalizes, runs your handler, and then — in a finally — marks the dispatcher complete and waits for the queue to drain. The finally matters: a handler that throws has usually already queued replies, and dropping them would lose the error message the user needs to see.

Deduplication

Reconnecting transports redeliver. The cache keys on channelId|accountId|from|threadId|messageId with a 20-minute TTL and a 5000-entry ceiling, swept lazily when full.

Two consequences worth knowing: a message with no messageId is never deduped (there is nothing stable to key on, and suppressing on content would eat legitimate repeats), and the cache is module-level, shared by every channel in the process. resetDedupe() exists so tests do not leak state into each other.

Inbound text is treated as untrusted

sanitizeInboundSystemTags rewrites [system message] → (System Message) and a line-leading System: → System (untrusted):. Inbound text is attacker-controlled and, on an agent-driven channel, goes into a prompt — the transform exists so a sender cannot forge framing that reads as the host's own system voice. finalizeInboundContext reports whether it fired via sanitized, and normalizes CRLF/CR to LF so chunking and prompt assembly see one newline form.

The reply dispatcher preserves order

Replies queue onto a single promise chain, so they arrive in the order they were queued — typically tool results, then block replies, then the final answer — regardless of how long each send takes. The dispatcher does not reorder by kind; kind is passed through to deliver so the caller can format each differently.

The pending counter starts at 1, as a reservation. Without it a handler that awaits before queuing its first reply would look idle and waitForIdle() would return immediately; the reservation is released by markComplete(), which dispatchInboundMessage calls for you.

humanDelay is applied between subsequent block replies only — never before the first, which would just add latency to every response. Empty payloads are dropped through onSkip rather than sent, but still counted in getQueuedCounts(), so a caller can tell "produced nothing" from "produced something unsendable". Delivery errors go to onError and never break the chain.

Chunking

| Function | Splits on | Use when | | --- | --- | --- | | chunkText | line boundaries, then hard-cut | plain text; the safe default | | chunkByParagraph | blank lines, falling back to chunkText | prose, to avoid mid-paragraph breaks | | chunkMarkdown | paragraphs, keeping fenced code blocks atomic | agent output containing code |

chunkMarkdown is the one worth reaching for on an agent channel. A code fence split across two messages renders as two broken blocks — the opening fence unterminated, the remainder as literal backticks — so fences are held together whole and only hard-cut when a single block exceeds the limit on its own.

All three take an explicit limit rather than reading channel config, so they stay pure and testable; deliverOutboundPayloads resolves the limit from the plugin.

Tests

pnpm --filter @liam-public/node-messaging test

Unit tests only — no network, no Docker. Channel gateways are exercised through fakes, so the suite covers reconnect and error-boundary behaviour without a live account.