ebus-proxy

High-performance HTTP + WebSocket proxy for cry-ebus2 services, built on uWebSockets.js.

• HTTP REST — GET (superjson) and POST (msgpack) call any ebus2 service.
• WebSocket — pub/sub channels with prefix matching, plus on-demand service request calls over the same socket.
• Auth — origin allow-list, API keys (header / query / cookie).

Table of Contents

• Installation
• Configuration
  • Security
  • Cookie Authentication
• Usage
• REST API
  • GET — superjson
  • POST — msgpack
  • JavaScript Client Example
• Utility Endpoints
• WebSocket API
  • Message Types
  • Prefix Subscriptions
  • Request/Response Correlation
  • JavaScript Client Example
• TypeScript Support
• API Documentation
• Development

Installation

npm install

Configuration

Create a .env file with the following variables:

| Variable | Description | Default | |----------|-------------|---------| | HOST | Server bind address | 0.0.0.0 | | PORT | Server port | 3033 | | TIMEOUT | Default request timeout in ms | 500 | | DEBUG | Enable debug logging | false | | MAX_BODY_SIZE_MB | Maximum request body size in MB | 1 | | ALLOWED_ORIGINS | Comma-separated list of allowed origin domains (right-matched, e.g. klik.vet also allows demo.klik.vet) | - | | API_KEYS | Comma-separated list of valid API keys | - | | SKIP_COLLECTIONS_FROM_DEBUG | Comma-separated collection names to show minimal debug output for | - |

Security

When ALLOWED_ORIGINS or API_KEYS are configured, requests must satisfy at least one of these conditions:

• Origin check: The Origin header hostname matches or is a subdomain of an entry in ALLOWED_ORIGINS.
• API key: A valid key is provided via:
  • Query parameter: ?apikey=<key> or ?api_key=<key>
  • Header: X-API-Key: <key> or Authorization: Bearer <key>
  • Cookie: apikey=<key> (set via /apikey/<key> endpoint)

If neither ALLOWED_ORIGINS nor API_KEYS are configured, all requests are allowed and a warning is printed on startup.

Cookie Authentication

Set an API key as an HttpOnly cookie by visiting /apikey/<key>. Clear it with /apikey/clear.

Example .env:

ALLOWED_ORIGINS=example.com, localhost
API_KEYS=secret-key-1, secret-key-2

Usage

npm start

To persist across reboots:

pm2 save
pm2 startup

REST API

/ accepts GET and POST, each with its own serialization format. Pick whichever fits your client.

GET — superjson

Request: query parameters

| Param | Required | Description | |-------|----------|-------------| | service | No | Service name. If omitted, the server returns the version banner (see below). | | payload | No | Forwarded to the worker; decoded by the first character (see table below). | | timeout | No | Request timeout in ms (default TIMEOUT env, otherwise 500). |

Payload decoding — the server inspects the first character of the URL-decoded payload:

| Pattern | Interpretation | Worker receives | |---------|----------------|-----------------| | starts with { | Parsed as superjson (superjson.parse(payload)). Use encodeURIComponent(superjson.stringify(data)) on the client. Handles Date, Map, Set, BigInt, undefined, RegExp, etc. Malformed → 400. | the decoded value | | Number.isFinite(Number(payload)) is true | Parsed as a finite number. | number | | anything else | Treated as a literal string. Non-finite numeric strings like Infinity / NaN stay as strings. | string |

Response: depends on what the worker returns.

| Worker returns | Content-Type | Body | |----------------|----------------|------| | string | text/plain | the raw string (no quoting, no wrapping) | | finite number | text/plain | String(n) | | anything else (objects, arrays, null, Date, Map, …) | application/json | superjson-encoded — parse with superjson.parse(await res.text()) |

Client tip: branch on Content-Type. If it's application/json, run the body through superjson.parse; otherwise use the text directly. (Type info is lost for scalars on the wire — a numeric body like "234" arrives as a string; coerce on the client if you need a number.)

No service → 200 plain text ebus-proxy <version> — banner / liveness probe.

GET /?service=echo&payload=test       → worker receives "test"   → body "test"   (text/plain)
GET /?service=echo&payload=123.45     → worker receives 123.45   → body "123.45" (text/plain)
GET /?service=echo&payload=%7B...%7D  → worker receives the parsed superjson value → body is superjson (application/json)
GET /                                  → 200 "ebus-proxy <version>" (banner)

Opt-in timing — ?ts. If the query string contains ts (any value, including empty — ?ts works), the response carries timing info:

| Worker returns | Response body | |----------------|---------------| | scalar (string/finite number) | three lines: the value, ts <Date.now()> at response time, rtt <N> ms (integer milliseconds, measured with performance.now()) | | anything else | application/json body containing { ts, rttMs, res } (superjson-encoded) where res is the original worker response |

GET /?service=echo&payload=234&ts        →  234
                                            ts 1715637284000
                                            rtt 7 ms

GET /?service=echo&payload=%7B...%7D&ts  →  application/json: superjson({ ts, rttMs, res: <worker-result> })

POST — msgpack

Request: JSON envelope in the body.

{ "service": "my-service", "payload": "<base64-msgpack>", "timeout": 1000 }

payload is base64-encoded msgpack (structuredClone: true). service is required.

Response: application/msgpack binary. Decode with unpackr.unpack(new Uint8Array(await res.arrayBuffer())).

No service → 400 Missing service parameter.

Opt-in timing — ?ts on the URL. Same flag as GET. When set, the msgpack body is { ts, rttMs, res } (worker response under res); when unset, the body is the worker response directly. Use ?ts as a query parameter even though the request itself is POST.

JavaScript Client Example

import superjson from "superjson";
import { Packr, Unpackr } from "msgpackr";

// --- GET helper that handles both response shapes ---
async function getService(url) {
  const res = await fetch(url);
  if (!res.ok) throw new Error(`${res.status} ${await res.text()}`);
  const text = await res.text();
  const ct = res.headers.get("content-type") || "";
  return ct.startsWith("application/json") ? superjson.parse(text) : text;
}

// Complex payload (use superjson) → object response (application/json)
const getPayload = encodeURIComponent(superjson.stringify({ key: "value", when: new Date() }));
const getResult = await getService(`http://localhost:3033/?service=my-service&payload=${getPayload}&apikey=secret-key-1`);

// Simple string / number payloads — no encoding helper needed
const echoStr = await getService(`http://localhost:3033/?service=echo&payload=test&apikey=secret-key-1`);    // worker sees "test"
const echoNum = await getService(`http://localhost:3033/?service=echo&payload=123.45&apikey=secret-key-1`); // worker sees 123.45

// --- POST (msgpack) ---
const packr = new Packr({ structuredClone: true });
const unpackr = new Unpackr({ structuredClone: true });
const postPayload = Buffer.from(packr.pack({ key: "value" })).toString("base64");
const postRes = await fetch("http://localhost:3033/", {
  method: "POST",
  headers: { "Content-Type": "application/json", "X-API-Key": "secret-key-1" },
  body: JSON.stringify({ service: "my-service", payload: postPayload })
});
const postResult = unpackr.unpack(new Uint8Array(await postRes.arrayBuffer()));

Utility Endpoints

| Endpoint | Auth | Description | |----------|------|-------------| | / | Yes | Service request endpoint (GET superjson / POST msgpack). GET with no service → ebus-proxy <version> plain text. POST with no service → 400. | | /ping | No | Returns pong <ISO-date> for health checks. | | /doc | No | Returns a one-screen cheat sheet of available endpoints. | | /apikey/:key | No | Sets the API key as an HttpOnly cookie (1 week TTL). Use /apikey/clear to delete. | | /spec | Yes | Returns the OpenAPI 3.1 specification (YAML). | | /stat | Yes | Server statistics (debug status, request and WS connection counters). | | /debug | Yes | Returns current debug logging status. | | /debug/on | Yes | Enables debug logging at runtime (returns yes). | | /debug/off | Yes | Disables debug logging at runtime (returns no). |

WebSocket API

Connect to ws://localhost:3033/. Authentication uses the same origin / API key rules as HTTP (pass the key via query param: ws://localhost:3033/?apikey=secret-key-1).

All frames use msgpack with structuredClone: true. The same packer used for POST.

Message Types

Client → Server:

| Type | Fields | Description | |------|--------|-------------| | subscribe | channel, id? | Subscribe to a channel (supports prefix matching). | | unsubscribe | channel, id? | Unsubscribe from a channel. | | publish | channel, data, id? | Publish a message to a channel. | | request | service, payload?, timeout?, id? | Call an ebus2 service over the socket. payload is the raw value — no base64, no superjson; the whole frame is already msgpack. Reply arrives as response. | | ping | id? | Keep-alive ping. |

Server → Client:

| Type | Fields | Description | |------|--------|-------------| | connected | - | Connection established. | | subscribed | channel, id?, already? | Subscription confirmed. already: true if the client was already subscribed. | | unsubscribed | channel, id? | Unsubscription confirmed. | | published | channel, id? | Publish confirmed. | | message | channel, data | Incoming message from a subscribed channel (exact or prefix match). channel is the full original path. | | response | service, data, ts, rttMs, id? | Reply to a request. data is the worker's return value. ts is Date.now() at reply time, rttMs is the ebus round-trip in integer ms (always included — no opt-in flag on WS). | | pong | id? | Ping response. | | error | error, id? | Error reply. |

Prefix Subscriptions

Subscriptions support prefix matching. Subscribing to a parent channel automatically receives messages from all child channels:

subscribe("db/mydb")        → receives messages for db/mydb/users, db/mydb/posts, etc.
subscribe("db/mydb/users")  → receives messages for db/mydb/users only (exact match)

This reduces the number of subscriptions needed when watching multiple sub-channels under a common prefix.

Request/Response Correlation

The optional id field correlates client messages with server replies:

// Client sends with id
ws.send(packr.pack({ type: "subscribe", channel: "events", id: "req-123" }));

// Server replies with the same id
// { type: "subscribed", channel: "events", id: "req-123" }

This applies uniformly to subscribe / unsubscribe / publish / request / ping.

JavaScript Client Example

import { Packr, Unpackr } from "msgpackr";

const packr = new Packr({ structuredClone: true });
const unpackr = new Unpackr({ structuredClone: true });

const ws = new WebSocket("ws://localhost:3033/?apikey=secret-key-1");
ws.binaryType = "arraybuffer";

const pending = new Map();
let reqId = 0;

function send(message) {
  return new Promise((resolve) => {
    const id = String(++reqId);
    pending.set(id, resolve);
    ws.send(packr.pack({ ...message, id }));
  });
}

ws.onmessage = (event) => {
  const msg = unpackr.unpack(new Uint8Array(event.data));
  if (msg.id && pending.has(msg.id)) {
    pending.get(msg.id)(msg);
    pending.delete(msg.id);
  } else if (msg.type === "message") {
    console.log("Channel message:", msg.channel, msg.data);
  }
};

// Pub/sub
await send({ type: "subscribe", channel: "my-channel" });
await send({ type: "publish", channel: "my-channel", data: { hello: "world" } });
await send({ type: "unsubscribe", channel: "my-channel" });

// Service request over the socket
const { data } = await send({ type: "request", service: "echo", payload: { foo: "bar" } });
console.log("worker returned:", data);

TypeScript Support

Type definitions are available for both REST and WebSocket APIs:

import type {
  ServiceRequestQuery,
  ServiceRequestBody,
  WsClientMessage,
  WsServerMessage,
  WsRequestMessage,
  WsResponseMessage,
  WsChannelMessage,
} from "cry-ebus-proxy/contract.types";

For runtime validation with Zod and the ts-rest contract:

import { contract, schemas } from "cry-ebus-proxy/contract.zod";

API Documentation

OpenAPI 3.1 specification is available at openapi.yaml. Serve a live copy from the running proxy via GET /spec.

Development

Build

Compile the TypeScript contracts:

npm run build

This generates dist/ with compiled JavaScript and type declarations.

Test

npm test

The suite spawns isolated child processes for each scenario. Mocha + Chai; see test/ for the layout.

Publish

npm run build
npm publish

The prepublishOnly script runs build automatically.

Published Files

• index.mjs — main server
• dist/contract.types.js + .d.ts — pure TypeScript types
• dist/contract.zod.js + .d.ts — Zod schemas and ts-rest contract
• openapi.yaml — OpenAPI 3.1 specification