@mantlejs/sync

Cross-instance event synchronisation for Mantle JS. Intercepts service:event emissions, publishes them to a shared message broker, and re-emits received messages from other instances — enabling @mantlejs/socketio to fan out real-time events across every connected process.

---

Installation

npm install @mantlejs/sync

For Redis (ioredis is an optional peer dependency):

npm install @mantlejs/sync ioredis

---

Concepts

How it works

sync() listens for Mantle’s internal service:event bus event (emitted after every create, update, patch, or remove) and publishes a SyncMessage to a shared pub/sub channel. Every other instance subscribed to that channel re-emits the message onto its own local event bus — where @mantlejs/socketio picks it up and fans it out to connected WebSocket clients.

Each instance generates a random instanceId (UUID) at startup. Messages received from the broker that carry the local instanceId as originId are dropped, preventing double-delivery to the originating process’s clients.

Broker failures (publish errors, subscribe failures) are non-fatal: they are logged via app.get('logger') if a logger is configured, and execution continues.

Delivery semantics: at-most-once

Events are delivered at most once — there is no persistence, acknowledgment, or replay. A message is lost when:

• the broker connection is down at publish time (the error is logged and execution continues);
• an instance is restarting, or subscribes after the message was published (pub/sub has no backlog);
• a WebSocket client is disconnected or reconnecting when its instance fans the event out.

Design consequence: clients must not treat the event stream as a source of truth. Treat events as cache-invalidation hints and refetch on reconnect. @mantlejs/client emits a reconnect event for exactly this, and @mantlejs/react’s MantleProvider calls queryClient.invalidateQueries() on it, bounding the staleness a missed event can cause. If you need guaranteed delivery, put a real queue (or an outbox table) behind your service instead of relying on service:event fan-out.

Adapters

Adapter	Source	Transport
redisAdapter()	@mantlejs/sync	Redis pub/sub (ioredis, two connections) — also compatible with DragonflyDB
supabaseAdapter()	@mantlejs/supabase	Supabase Realtime Broadcast — zero additional infrastructure for Supabase users

Custom adapters can be plugged in by implementing the SyncAdapter interface.

---

Quick Start

Redis

import { mantle } from "@mantlejs/mantle";
import { express } from "@mantlejs/express";
import { socketio } from "@mantlejs/socketio";
import { sync, redisAdapter } from "@mantlejs/sync";

const app = mantle()
  .configure(express())
  .configure(socketio())
  .configure(sync({ adapter: redisAdapter({ url: process.env.REDIS_URL }) }));

app.listen(3030);

Supabase Realtime

import { sync } from "@mantlejs/sync";
import { supabase, supabaseAdapter } from "@mantlejs/supabase";

app
  .configure(supabase({ url: process.env.SUPABASE_URL!, key: process.env.SUPABASE_KEY! }))
  .configure(sync({ adapter: supabaseAdapter() }));

---

API

sync(options)

Returns a Mantle plugin that wires the sync adapter to the application event bus.

function sync(options: SyncOptions): MantlePlugin;

Options

Option	Type	Default	Description
adapter	SyncAdapter	—	Required. The pub/sub transport adapter.
channel	string	"mantle:sync"	Shared pub/sub channel name. Must be the same across all instances.

---

redisAdapter(options?)

Creates a Redis pub/sub adapter using ioredis with two dedicated connections (one for publishing, one for subscribing — required by the Redis protocol).

function redisAdapter(options?: RedisAdapterOptions): SyncAdapter;

Options

Option	Type	Default	Description
url	string	—	Full Redis URL (e.g. "redis://user:pass@host:6379/0"). Takes precedence over host/port.
host	string	"127.0.0.1"	Redis hostname. Ignored when url is set.
port	number	6379	Redis port. Ignored when url is set.
password	string	—	AUTH password.
db	number	0	Redis database index.
tls	boolean	false	Enable TLS.

---

Types

import type { SyncMessage, SyncMessageParams, SyncAdapter, SyncOptions, RedisAdapterOptions } from "@mantlejs/sync";
import { sync, redisAdapter } from "@mantlejs/sync";

SyncMessage

The message envelope published and received by adapters.

interface SyncMessage {
  originId: string; // UUID of the originating instance
  path: string; // service path (e.g. "users")
  event: string; // event name (e.g. "created")
  result: unknown; // the service method result
  params: SyncMessageParams;
}

SyncMessageParams

The whitelisted subset of ServiceParams that actually crosses the broker — not the full params object from the originating call:

interface SyncMessageParams {
  provider?: string;
  query?: Record<string, unknown>;
  user?: { id: string | number };
}

sync() reduces every outgoing (and incoming) message to this shape before it touches the adapter. It never carries headers (live credentials), the full user record, or the non-serializable connection object — so request auth tokens and full user documents can never leak across the pub/sub channel, even if a custom adapter logs or persists raw messages.

SyncAdapter

Implement this interface to create a custom pub/sub adapter.

interface SyncAdapter {
  publish(channel: string, message: SyncMessage): Promise<void>;
  subscribe(channel: string, handler: (message: SyncMessage) => void): Promise<void>;
  close(): Promise<void>;
}

---

Development

npx nx build sync    # compile
npx nx test sync     # run tests
npx nx lint sync     # lint

---

Publishing

Build before publishing:

npx nx build sync

First publish (scoped packages require --access public):

cd packages/sync
npm publish --access public

Subsequent releases — bump version in packages/sync/package.json, then:

cd packages/sync
npm publish