# tyo-mq
[![NPM module](https://badge.fury.io/js/tyo-mq.png)](https://badge.fury.io/js/tyo-mq)

TYO-MQ is a distributed messaging (pub/sub) service with socket.io. 

[![NPM](https://nodei.co/npm/tyo-mq.png?stars&downloads)](https://nodei.co/npm/tyo-mq/)

TYO-MQ supports fire-and-forget pub/sub by default, plus opt-in durable delivery with ACK, retry, and dead-letter handling for messages that must survive disconnects or failed consumers.

**See it in action:** the [tyo-mq-samples](https://github.com/tyolab/tyo-mq-samples) repo has four runnable mini-apps — a job queue with retry and DLQ, a browser chat, IoT telemetry over wildcard topics, and event-driven microservices — plus a cookbook with one self-contained script per feature. Clone, `npm install`, run.

## The client moved to `tyo-mq-client`

Building an app? Install the client: `npm install tyo-mq-client` — it's
**Apache-2.0** and has no AGPL dependency. Migration is one line:
`require('tyo-mq').Factory` → `require('tyo-mq-client').Factory` (same
API). This package (`tyo-mq`) is the **broker/server**, licensed
**AGPL-3.0-or-later** (a commercial licence is available for organisations
that can't use AGPL — see [COMMERCIAL-LICENSE.md](./COMMERCIAL-LICENSE.md)).
**You only deal with AGPL if you self-host or modify the broker — app
developers use the Apache client and are unaffected.**

This is a cleaner licensing story, not a more complicated one: the part
you `npm install` into your app is permissively licensed, and only the
broker you run as infrastructure carries AGPL's copyleft terms — the same
split as, say, MongoDB's driver vs. server, or Elasticsearch's client vs.
cluster.

## Installation

    npm install tyo-mq

For the client, see [tyo-mq-client](https://github.com/tyolab/tyo-mq-client) instead:

    npm install tyo-mq-client

## Client libraries

The Node.js client (and the browser bundle) live in their own Apache-2.0
package, [tyo-mq-client](https://github.com/tyolab/tyo-mq-client) — see
that repo's README for usage. Clients for other languages live in their
own repositories too:

| Language | Repository | Install |
|----------|------------|---------|
| Node.js / browser | [tyo-mq-client](https://github.com/tyolab/tyo-mq-client) | `npm install tyo-mq-client` |
| Python | [tyo-mq-client-python](https://github.com/tyolab/tyo-mq-client-python) | `pip install tyo-mq-client` |
| Go | [tyo-mq-client-go](https://github.com/tyolab/tyo-mq-client-go) | `go get github.com/tyolab/tyo-mq-client-go` |
| Rust | [tyo-mq-client-rust](https://github.com/tyolab/tyo-mq-client-rust) | crate `tyo-mq-client` (git) |
| Ruby | [tyo-mq-client-ruby](https://github.com/tyolab/tyo-mq-client-ruby) | `gem install tyo-mq-client` |
| C/C++ | [tyo-mq-client-cpp](https://github.com/tyolab/tyo-mq-client-cpp) | CMake `FetchContent` |
| Java | [tyo-mq-client-java](https://github.com/tyolab/tyo-mq-client-java) | Maven `au.com.tyo:tyo-mq-client` |
| C# / .NET | [tyo-mq-client-csharp](https://github.com/tyolab/tyo-mq-client-csharp) | `dotnet add package TYO_MQ_CLIENT` |

All clients speak the same wire protocol; anything a helper doesn't cover is
one `emit`/`on` away.

That claim is verified, not aspirational: the
[tyo-mq-conformance](https://github.com/tyolab/tyo-mq-conformance) suite runs
every client above through the same scenarios against a live server — pub/sub
payload fidelity across every producer/consumer language pair, durable replay,
manual ACK/retry/dead-lettering, topic wildcards, consumer groups, and auth —
and publishes the resulting conformance matrix.

## Creating a messaging server

```javascript
var MessageServer = require("tyo-mq").Server;

var mq = new MessageServer();
mq.start();
```

## Creating a message producer or subscriber

Producer and consumer code lives in `tyo-mq-client` now — install it
alongside (or instead of, if you're not running your own broker) this
package:

```javascript
var Factory = require('tyo-mq-client').Factory,
    producer;

var mq = new Factory();

mq.createProducer('testevent')
.then(function (p) {
    producer = p;

    // produce a default event with data {data: 'test'}
    producer.produce('test text from default event');

    // produce a different kind of event
    producer.produce('event2', {data: 'test text from event2'})
});
```

```javascript
var Factory = require('tyo-mq-client').Factory,
    consumer;

var mq = new Factory();

mq.createConsumer()
.then(function (c) {
    consumer = c;

    // subscribe 'event2'
    consumer.subscribe('event2', (data) => {
        console.log(data);
    });

    // subscribe 'testevent'
    consumer.subscribe('testevent', (data) => {
        console.log(data);
    });
});
```

See [tyo-mq-client](https://github.com/tyolab/tyo-mq-client#readme) for
the full client API (durable delivery, ACK, topics, groups, broadcast).

## Durable Delivery and ACK

Durable subscriptions store matching messages while the consumer is offline. ACK
is not required by default: `durable: true` alone keeps the Phase 2 behavior,
where replayed durable messages are removed immediately after delivery. When a
subscription explicitly asks for ACK with `ack`, `require_ack`, or `manual_ack`,
the server includes a `msgId`, waits for `ACK {msgId}`, retries on timeout, and
moves exhausted messages to the realm DLQ.

```javascript
consumer.subscribe(producer.name, 'task', function (data) {
    // auto-ACK after the handler resolves
}, {
    durable: true,
    ack: true,
    retry: { max_attempts: 3, delay: '5s', backoff: 'exponential' }
});

consumer.subscribe(producer.name, 'task', function (data, from, ack, raw) {
    doWork(data).then(function () {
        ack();
    });
}, {
    durable: true,
    manual_ack: true,
    ack_timeout: '30s',
    retry: { max_attempts: 5, delay: '2s' }
});
```

The storage backends expose `deadLetter(msgId, reason)`, `listDlq(realm)`, and
`discardDlq(msgId)` for management tooling. The in-memory backend keeps DLQ
entries until process exit; SQLite and Redis keep them in their configured
stores.

## Topics, Consumer Groups, and Broadcast

All routing features are opt-in; the default producer/event routing is
unchanged.

**Topic subscriptions** (`{ mode: 'topic' }`) use MQTT-style hierarchical
patterns: `+` matches exactly one level, `#` matches any number of trailing
levels. Topic subscriptions match the produced event name from any producer.
They compose with `durable`, `ack`, and `retry` like any other subscription.

```javascript
consumer.subscribe('org/acme/machine/+/cmd', function (data, from, ack, raw) {
    console.log('command for', raw.event, ':', data);  // raw.event = concrete topic
}, { mode: 'topic' });

consumer.subscribe('org/acme/#', handler, { mode: 'topic', durable: true });

producer.produce('org/acme/machine/m-01/cmd', 'restart');
```

**Consumer groups** (`{ group: 'name' }`) share the load of an event or topic:
each message is delivered to exactly one member of the group (round-robin),
while ungrouped subscribers still receive every message.

```javascript
workerA.subscribe('org/acme/machine/+/cmd', handler, { mode: 'topic', group: 'workers' });
workerB.subscribe('org/acme/machine/+/cmd', handler, { mode: 'topic', group: 'workers' });
```

**Broadcast** sends one message to every connected member of the producer's
realm — including realm members with no matching subscription — or to every
member of a group (one copy each, no load balancing):

```javascript
producer.produce('announcement', data, { broadcast: 'realm' });
producer.produce('rollout', data, { broadcast: 'group', group: 'workers' });
```

Broadcast is fire-and-forget: it does not enqueue for offline durable
subscribers.

## Clustering

Multiple tyo-mq nodes can share one Redis: managed settings (realms,
pre-shared keys, approved tokens) stay in sync across nodes, signed manager
commands cannot be replayed against a peer node, durable subscribers can
reconnect to any node, live messages are relayed so a producer on one node
reaches subscribers on every node, and authorization requests can be decided
from any node. Enable it in the settings file on every node:

```json
{
  "storage": "redis",
  "storage_options": { "url": "redis://10.0.0.5:6379/0" },
  "cluster": { "enabled": true }
}
```

`cluster.redis_url` defaults to `storage_options.url`. Consumer groups are
load-balanced per node, so a group's members should connect to the same node.
See [docs/CLUSTERING.md](docs/CLUSTERING.md) for the full setup guide and
delivery semantics.

## Observability (opt-in HTTP API)

A read-only HTTP surface can be enabled at startup, served on the **same
port** as the socket server (no second port to firewall). It is off by
default — without the option, no HTTP endpoint exists.

```json
{ "http_api": { "enabled": true } }
```

| Endpoint | Auth | Description |
|----------|------|-------------|
| `GET /health` | none | Liveness for load balancers: status, version, uptime, cluster node id |
| `GET /api/metrics` | Bearer admin token* | Prometheus text format: produced/delivered/queued/DLQ/ack-timeout counters, connection counts |
| `GET /api/stats` | Bearer admin token | Per-realm producer/consumer totals and online counts, subscription counts |
| `GET /api/realms/{realm}/dlq` | Bearer admin token | Dead-letter queue contents for the realm |

\* When auth is disabled, no token is needed. `http_api.metrics_auth: false`
opens `/api/metrics` for scrapers that cannot send a token (Prometheus itself
supports `authorization: { credentials: ... }` in the scrape config, so the
default is to require it). Individual endpoints can be switched off with
`health: false`, `metrics: false`, or `stats: false`.

Management (write) operations intentionally stay on the signed socket
command channel — the HTTP surface is read-only by design.

## Rate limits and quotas (opt-in)

For public or multi-tenant deployments, a `limits` settings block turns on
abuse protection — entirely inert without it:

```json
{
  "limits": {
    "trust_proxy": true,
    "messages_per_second": 20, "message_burst": 60,
    "bytes_per_second": 262144,
    "max_connections_per_ip": 20, "connections_per_minute_per_ip": 60,
    "max_registrations_per_socket": 50, "max_registrations_per_realm": 500,
    "max_subscriptions_per_socket": 50, "max_subscriptions_per_realm": 500,
    "max_queued_per_realm": 1000,
    "realms_per_hour_per_ip": 10,
    "max_pending_authorization_requests": 1000,
    "realms": { "org:vip": { "messages_per_second": 200 } }
  }
}
```

Every limit is independent and optional (absent = unlimited), hot-reloads
from the settings file, and can be overridden per realm under
`limits.realms` — the basis for tiered service. Throttled clients receive
`RATE_LIMITED { code: 429, reason, retry_after }` (at most once a second
per socket); connection-level breaches disconnect; every enforcement is
counted in `tyo_mq_rate_limited_total{reason}` on the metrics endpoint.
Behind a reverse proxy, set `trust_proxy: true` so per-IP limits use
`X-Forwarded-For`. The durable-queue cap applies on memory and sqlite
storage (backends that report queue depth).

## Custom namespaces (opt-in)

The broker can host additional socket.io namespaces from operator-supplied
modules — new realtime surfaces without touching tyo-mq itself:

```json
{
  "namespaces": {
    "/collab": { "module": "./namespaces/collab.js", "options": { "max_rooms": 100 } }
  }
}
```

A module exports `attach(io, options, context)` — it calls
`io.of(context.name)` and wires its own handlers; `context.getOptions()`
reads the live options block (hot-reloaded), and any returned API surfaces
as `server.namespaces['/collab']`. The bundled `/remote` streaming
namespace implements this same contract and is the reference
implementation (`lib/remote-namespace.js`). Namespaces attach at startup
and when they appear in a settings reload; a broken module is logged and
skipped, never crashing the broker. `/` and `/remote` are reserved.

## Demo

### Start the TYO-MQ server

```javascript
# Needs to set up the library (module) path
export NODE_PATH=`npm config get prefix`/lib/node_modules/
node -e 'require("tyo-mq/server")'
```

### Customize Server Configuration

You can customize the server configuration including CORS settings by creating your own server file:

```javascript
var Server = require('tyo-mq').Server;

var server = new Server({
    serveClient: false,
    pingInterval: 5000,
    pingTimeout: 10000,
    allowEIO3: true,
    // CORS configuration
    cors: {
        origin: "*",  // Allow all origins, or specify ["http://localhost:3000", "https://yourdomain.com"]
        methods: ["GET", "POST"],
        credentials: true
    },
    // WebSocket compression settings
    perMessageDeflate: {
        threshold: 2048,
        zlibDeflateOptions: {
            chunkSize: 8 * 1024,
        },
        zlibInflateOptions: {
            windowBits: 14,
            memLevel: 7,
        },
        clientNoContextTakeover: true,
        serverNoContextTakeover: true,
        serverMaxWindowBits: 10,
        concurrencyLimit: 20,
    },
});

server.start(8080); // Specify custom port
```

### Authentication and Realms

Authentication is disabled by default for backwards compatibility. When enabled,
clients must send `AUTHENTICATION` before registering as a producer or consumer.
The built-in clients can do this automatically with a configured token:

```javascript
var Server = require('tyo-mq').Server;
var Factory = require('tyo-mq-client').Factory;

var server = new Server({
    auth: {
        enabled: true,
        tokens: [
            { token: 'secret-acme-prod', realm: 'acme', role: 'producer' },
            { token: 'secret-acme-cons', realm: 'acme', role: 'consumer' }
        ]
    }
});
server.start();

var producerMq = new Factory({ auth: { token: 'secret-acme-prod' } });
var consumerMq = new Factory({ auth: { token: 'secret-acme-cons' } });
```

Supported roles are `producer`, `consumer`, `both`, `manager`, and `admin`.
Producers, consumers, and subscriptions are scoped to the authenticated
`realm`, so the same producer or consumer names can exist independently in
different realms. `manager` is a per-realm administration role; `admin` spans
all realms (`realm: "*"`). In addition to configured opaque tokens, the server
can validate HS256 JWTs with `auth.jwt_secret` or delegate validation to an
HTTP endpoint with `auth.auth_url`.

#### Connection authorization

Only `manager` connections always require manual authorization (an approved
token, granted via the authorization request flow or configured directly).
Other roles are governed by per-realm settings:

```json
{
  "auth": {
    "enabled": true,
    "realms": {
      "acme": {
        "key": "consumer-pre-shared-key",
        "require_acceptance": true
      }
    }
  }
}
```

- `consumer` (and `both`) present the realm's pre-shared `key`. Connections
  without a realm, or to a realm without a key (or with `require_key: false`),
  are allowed automatically.
- `producer` (and `both`) must be accepted into the realm through the
  authorization request flow, unless the realm sets
  `require_acceptance: false`.

```javascript
// consumer joins with the realm pre-shared key — no token needed
var consumerMq = new Factory({
    auth: { realm: 'acme', role: 'consumer', key: 'consumer-pre-shared-key' }
});
```

The pre-shared key and acceptance requirement are managed with the signed
management commands `set_realm_key` and `set_realm_acceptance` — available from
the interactive manager (`npm run manager`), the web manager UI
(`npm run manager:web`), or `Authorization.authManagementCommand()`.

Realms come in two forms: **permanent** (the default) and **ephemeral /
disposable** — created with `add_realm {ephemeral: true, ttl: '2h'}` and
disposed of automatically when the TTL lapses (config, scoped tokens, runtime
state, connected sockets, and stored messages are all removed). Convert
between the forms with `set_realm_lifetime`. See [REALM.md](REALM.md) for
details, and [EPHEMERAL-REALMS.md](EPHEMERAL-REALMS.md) for a self-contained
implementation guide to provisioning ephemeral realms programmatically.

When auth is enabled and no `realm: "*", role: "admin"` token is configured,
the server creates one automatically and appends it to `.env` as
`TYO_MQ_ADMIN_TOKEN`. The default `server.js` loads `.env`, so you can enable
auth and let the first server start create the admin token:

```bash
printf 'TYO_MQ_AUTH_ENABLED=true\n' > .env
npm start
```

In another shell, verify the generated token can authenticate:

```bash
npm run auth:admin
```

The helper reads `TYO_MQ_ADMIN_TOKEN` from `.env` and sends `AUTHENTICATION` to
the running server.

Managers do not need to send shared secrets to approve new client tokens. A
client can submit a pending authorization request:

```bash
npm run auth:request -- \
  --realm tyolab \
  --role consumer \
  --client-id tyolab-agent-01 \
  --client-name "Tyolab Agent 01"
```

The command prints a generated `client_token` and `request_id`. A server admin
can retrieve pending requests and approve or reject them with a signed proof
derived from `TYO_MQ_ADMIN_TOKEN`:

```bash
npm run auth:manager -- next
npm run auth:manager -- approve <request_id> --role consumer
npm run auth:manager -- reject <request_id> --reason "unknown client"
```

An org or realm operator can use a scoped realm manager key instead:

```json
{
  "auth": {
    "realms": {
      "tyolab": {
        "required": true,
        "manager_key": "realm-manager-shared-secret"
      }
    }
  }
}
```

```bash
TYO_MQ_REALM_MANAGER_KEY="realm-manager-shared-secret" \
  npm run auth:manager -- next --realm tyolab

TYO_MQ_REALM_MANAGER_KEY="realm-manager-shared-secret" \
  npm run auth:manager -- approve <request_id> --role consumer
```

The manager proof is an HMAC-SHA256 signature over the action, body, timestamp,
and nonce. The admin token or realm manager key stays local to the manager
script. A realm manager key can only poll and decide authorization requests in
its configured realm; server-wide management commands still require the global
admin token. Approved client tokens are added to the server's auth token list.
When `TYO_MQ_SETTINGS_FILE` is configured, approved tokens are persisted to that
file and survive restarts; otherwise they are runtime-only.

Once realm/token data grows beyond hand-edited scale, enable the **SQLite auth
store** (opt-in: `"auth_store": {"filename": "tyo-mq.auth.sqlite"}` in
settings, `--auth-store`, or `TYO_MQ_AUTH_STORE=true`; Node 22+): realms and
tokens are then persisted
row-level in SQLite (WAL, crash-safe) instead of rewriting the JSON file on
every change, with automatic import of existing data on first boot. See
[AUTHENTICATION.md](AUTHENTICATION.md#sqlite-auth-store).

Approved client tokens can be revoked through the interactive manager or the
signed management command `revoke_token`. Revocation can identify a token by
`token_hash`, or by `realm` plus `client_id`.

The same flow is available as library calls:

```javascript
var Authorization = require('tyo-mq-client').Authorization;

await Authorization.submitAuthorizationRequest({
    realm: 'tyolab',
    role: 'consumer',
    client_id: 'tyolab-agent-01',
    client_name: 'Tyolab Agent 01',
    client_token: 'client-secret',
    challenge_response: { ticket: 'INC-123' }
});

var next = await Authorization.nextAuthorizationRequest(process.env.TYO_MQ_ADMIN_TOKEN);
await Authorization.decideAuthorizationRequest(process.env.TYO_MQ_ADMIN_TOKEN, {
    request_id: next.request.request_id,
    approved: true,
    role: 'consumer'
});

var realmNext = await Authorization.nextRealmAuthorizationRequest(
    process.env.TYO_MQ_REALM_MANAGER_KEY,
    'tyolab'
);
await Authorization.decideRealmAuthorizationRequest(process.env.TYO_MQ_REALM_MANAGER_KEY, {
    request_id: realmNext.request.request_id,
    approved: true,
    role: 'consumer'
});
```

For interactive auth and realm management:

```bash
npm run manager
```

With Docker Compose, provide the admin token from the host and persist server
settings through the bundled volume:

```bash
export TYO_MQ_ADMIN_TOKEN="$(openssl rand -hex 32)"
docker compose up -d tyo-mq
docker compose run --rm manager
```

**CORS Options:**
- `origin: "*"` - Allow all origins (development/testing)
- `origin: ["http://localhost:3000"]` - Allow specific origins (production)
- `methods` - Allowed HTTP methods
- `credentials: true` - Allow credentials in requests

### Test Script
```javascript
export NODE_PATH=`npm config get prefix`/lib/node_modules/
node -e 'require("tyo-mq/test")'
```

## Browserify
This package supports being browserified.
In order to browserify, please install two more extra packages:
```
npm install utf-8-validate bufferutil
```

Afterward,
```
browserify web/web.js -o web/client/tyo-mq-client.js
```

## TODO list
* implement the message queuing
* message queuing if intended subscriber is down, resend message when it is up
* message delivery for one or some intended subscribers only

## Maintainer

[Eric Tang](https://twitter.com/_e_tang) @ [TYO LAB](http://tyo.com.au)