From bdfea8ee370e96becb931f7534f791f801aa0113 Mon Sep 17 00:00:00 2001 From: "rayson951005@gmail.com" Date: Fri, 26 Jun 2026 09:47:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(routing):=20Envelope=20=E4=B8=89=E6=A1=A3?= =?UTF-8?q?=E8=B7=AF=E7=94=B1=20+=20traceId/hop=20=E9=98=B2=E7=8E=AF=20+?= =?UTF-8?q?=20=E7=A6=BB=E7=BA=BF=E8=A1=A5=E6=8A=95=20(PR6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按 spec §3.2 把 broker 从纯 topic 扇出升级为身份感知路由，消费 Store(pending/getMembers)。 - 三档路由：per-ws handler 的 shouldDeliver 过滤——broadcast(无 to → 全发)、@提及(全发， 高亮 client 侧 mentions[])、DM(to:[id] → 仅命名身份，present to 即 DM、空 to → nobody，绝不退化广播)。 - loop prevention：from-skip(绝不回声发送者) + hop<=0 drop(多 hop 泛化)。 - 防伪：publish 入口运行期 envelope 校验(from 必对象/to 必数组/idempotencyKey/roomId 必非空 string) + 无条件 stamp env.from.agentId=认证身份(覆盖客户端伪造)。 - 离线补投：store_if_offline 对 intended(DM=to / broadcast=getMembers) 中不可达身份 enqueuePending； hello(welcome) 与 subscribe 后都 drainPendingTo 补收(覆盖「已连接未订阅」gap 窗口)。 - topicMembers Map> 跟踪每 topic 可达身份(支持同身份多连接)。 - 双 hello 守卫(防身份重绑泄漏 topicMembers)；drain 移出 auth try/catch(防 store 错误误踢连接)。 测试：broker-routing.test 12 用例(DM 隔离/broadcast 跳发送者/hop-drop/离线补投/防伪/double-hello/ gap-drain/drain-error/空-to/missing-key/**SqliteStore 离线**堵 InMemoryStore 测试盲点)；现有测试 改双身份(同身份 pub/sub 被 from-skip 是预期)。full check exit 0：1763 pass / 0 fail。 Cross-review：5 轮(轮1 thorough + 4 lean)，抓出并修复 6 个真实 issue——HIGH 缺失/非对象 from 绕过 防伪+防环失效、HIGH drain 在 auth catch 内致 store 错误误踢连接、MED gap 窗口离线丢、MED 双 hello 重绑泄漏、LOW string-to 子串匹配 DM 泄漏、LOW idempotencyKey/roomId 未校验致 SqliteStore 静默丢(§6.4 跨实现分叉)。连续两轮 0 收敛。 Backlog：topic==roomId 双源校验、to 元素类型校验、from.name/agentType 防伪、异步 Store 的 offline+live 原子性(§8.2/PR11)、BrokerClient 暴露 onError、广播离线混合成员测试。 --- feat(routing): three-tier Envelope routing + traceId/hop loop prevention + offline replay (PR6) Upgrade the broker from plain topic fan-out to identity-aware routing (§3.2): broadcast / @mention / DM (by subscriber identity), loop prevention (never echo to sender + hop<=0 drop), anti-spoof (runtime envelope validation + stamp the authenticated sender), and store_if_offline replay (enqueue for unreachable recipients, drain on (re)connect + subscribe). Converged after 5 adversarial rounds that caught + fixed 6 real issues (2 HIGH). 🤖 Generated with [Claude Code](https://claude.com/claude-code) --- plugins/agentbridge/server/bridge-server.js | 4 +- plugins/agentbridge/server/daemon.js | 4 +- src/broker.ts | 144 +++++++- src/integration-test/broker-client.test.ts | 16 +- src/integration-test/broker-routing.test.ts | 358 ++++++++++++++++++++ src/integration-test/broker.test.ts | 12 +- 6 files changed, 511 insertions(+), 27 deletions(-) create mode 100644 src/integration-test/broker-routing.test.ts diff --git a/plugins/agentbridge/server/bridge-server.js b/plugins/agentbridge/server/bridge-server.js index 5352ddf..71b2b0a 100755 --- a/plugins/agentbridge/server/bridge-server.js +++ b/plugins/agentbridge/server/bridge-server.js @@ -14707,10 +14707,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.24", "0.0.0-source"), - commit: defineString("3b71186", "source"), + commit: defineString("cdb0dcc", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("d4792cf81ed0", "source") + codeHash: defineString("3908e9b66df5", "source") }); function sameRuntimeContract(a, b) { if (!a || !b) diff --git a/plugins/agentbridge/server/daemon.js b/plugins/agentbridge/server/daemon.js index 38be71e..2da647b 100755 --- a/plugins/agentbridge/server/daemon.js +++ b/plugins/agentbridge/server/daemon.js @@ -30,10 +30,10 @@ function defineNumber(value, fallback) { } var BUILD_INFO = Object.freeze({ version: defineString("0.1.24", "0.0.0-source"), - commit: defineString("3b71186", "source"), + commit: defineString("cdb0dcc", "source"), bundle: defineBundle("plugin"), contractVersion: defineNumber(1, CONTRACT_VERSION), - codeHash: defineString("d4792cf81ed0", "source") + codeHash: defineString("3908e9b66df5", "source") }); function daemonStatusBuildInfo() { return { ...BUILD_INFO }; diff --git a/src/broker.ts b/src/broker.ts index d87ad43..b1797e2 100644 --- a/src/broker.ts +++ b/src/broker.ts @@ -43,12 +43,19 @@ export interface BrokerOptions { * daemon (independent failure domain) and binds a CONFIGURABLE host (default * loopback; Tailscale uses the 100.x address, never 0.0.0.0 — §7.3). * - * Loop prevention / dedup (traceId/hop) and three-tier routing (broadcast / - * @mention / DM) are layered on top in a later PR; this PR is plain topic fan-out. + * Routing (§3.2): three-tier — broadcast (no `to` → all room subscribers), + * @mention (delivered to all, highlight is client-side via `mentions[]`), and DM + * (`to: [agentId]` → only those identities). Loop prevention: never echo to the + * sender (`from.agentId`) + a `hop<=0` drop (the multi-hop generalisation of the + * v1 binary-source guard). Offline replay: a `store_if_offline` envelope is + * persisted (Store.pending) for any intended recipient with no live subscription, + * and drained to them on their next (re)connect. */ export class Broker { private server: ReturnType | null = null; private nextConnId = 0; + /** topic → (identityId → live-subscription count) — who is reachable per topic. */ + private readonly topicMembers = new Map>(); private readonly transport: MessageTransport; private readonly log: (msg: string) => void; @@ -87,6 +94,8 @@ export class Broker { }); }, close(ws) { + const me = ws.data.identity?.id; + if (me) for (const topic of ws.data.subs.keys()) self.removeTopicMember(topic, me); for (const unsub of ws.data.subs.values()) unsub(); ws.data.subs.clear(); self.log(`conn #${ws.data.connId} closed`); @@ -129,15 +138,33 @@ export class Broker { const msg = parsed as ClientMessage; if (msg.type === "hello") { + // Reject a second hello on an already-authenticated socket: re-binding the + // identity would strand the old identity's subscriptions + topicMembers + // counts (close removes under the NEW identity), leaving the old one + // permanently "reachable" and silently dropping its store_if_offline. + if (ws.data.identity) { + this.send(ws, { type: "error", reason: "already authenticated" }); + return; + } + let identity: Identity; try { - const identity = await this.opts.identityProvider.authenticate(msg.token); - ws.data.identity = identity; - this.send(ws, { type: "welcome", identity }); - this.log(`conn #${ws.data.connId} authenticated as ${identity.id}`); + identity = await this.opts.identityProvider.authenticate(msg.token); } catch { // Never echo the presented token or the underlying reason. this.send(ws, { type: "auth_error", reason: "invalid token" }); ws.close(CLOSE_AUTH_FAILED, "auth failed"); + return; + } + ws.data.identity = identity; + this.send(ws, { type: "welcome", identity }); + this.log(`conn #${ws.data.connId} authenticated as ${identity.id}`); + // Reconnect replay (§3.2) — OUTSIDE the auth try/catch: a transient store + // error during drain must NOT be misreported as an auth failure (which + // would close a connection that was already welcomed + resolved client-side). + try { + await this.drainPendingTo(ws, identity.id); + } catch (e) { + this.log(`pending drain failed for ${identity.id}: ${String(e)}`); } return; } @@ -147,16 +174,25 @@ export class Broker { this.send(ws, { type: "error", reason: "not authenticated (send hello first)" }); return; } + const me = ws.data.identity.id; switch (msg.type) { case "subscribe": { - if (ws.data.subs.has(msg.topic)) return; // idempotent const topic = msg.topic; + if (ws.data.subs.has(topic)) { + this.send(ws, { type: "subscribed", topic }); // re-ack so a re-subscribe never hangs + return; + } const unsub = this.transport.subscribe(topic, (envelope) => { - this.send(ws, { type: "event", topic, envelope }); + if (this.shouldDeliver(me, envelope)) this.send(ws, { type: "event", topic, envelope }); }); ws.data.subs.set(topic, unsub); - this.send(ws, { type: "subscribed", topic }); // ack: safe to publish now + this.addTopicMember(topic, me); + this.send(ws, { type: "subscribed", topic }); + // Drain anything queued during the connected-but-not-yet-subscribed gap + // (between hello's drain and this subscribe). Safe: drainPending removes, + // so an already-drained message is never re-delivered. + await this.drainPendingTo(ws, me); return; } case "unsubscribe": { @@ -164,13 +200,51 @@ export class Broker { if (unsub) { unsub(); ws.data.subs.delete(msg.topic); + this.removeTopicMember(msg.topic, me); } return; } case "publish": { // CONTROL PLANE ONLY: forward the structured Envelope. No filesystem, - // no repo access — code sync is git's job (§2.6). - await this.transport.publish(msg.topic, msg.envelope); + // no repo access — code sync is git's job (§2.6). The broker is a trust + // boundary (§7.3), so validate the envelope SHAPE at runtime — TS types + // are compile-time only; a client can send anything over the wire. + const env = msg.envelope as Envelope | null | undefined; + if (!env || typeof env !== "object" || Array.isArray(env)) { + this.send(ws, { type: "error", reason: "malformed envelope" }); + return; + } + if (!env.from || typeof env.from !== "object" || Array.isArray(env.from)) { + this.send(ws, { type: "error", reason: "envelope.from must be an object" }); + return; + } + if (env.to !== undefined && !Array.isArray(env.to)) { + // A string `to` would make `to.includes(me)` a SUBSTRING match (DM leak). + this.send(ws, { type: "error", reason: "envelope.to must be an array" }); + return; + } + // Load-bearing for offline replay: a missing idempotencyKey hits the + // NOT NULL + OR IGNORE in SqliteStore → silently dropped (≠ InMemoryStore, + // a §6.4 divergence); a missing roomId drains as `topic: undefined`. + if (typeof env.idempotencyKey !== "string" || env.idempotencyKey === "") { + this.send(ws, { type: "error", reason: "envelope.idempotencyKey must be a non-empty string" }); + return; + } + if (typeof env.roomId !== "string" || env.roomId === "") { + this.send(ws, { type: "error", reason: "envelope.roomId must be a non-empty string" }); + return; + } + // Anti-spoof + reliable loop prevention: stamp the authenticated sender + // unconditionally (from is now guaranteed a plain object). + env.from.agentId = me; + // hop<=0 → drop (multi-hop loop guard, §3.2). + if (typeof env.hop === "number" && env.hop <= 0) return; + // Offline replay: persist for intended recipients with no live subscription. + if (env.deliveryMode === "store_if_offline") { + await this.storeForOfflineRecipients(msg.topic, env, me); + } + // Live fan-out; each subscriber's handler applies shouldDeliver (DM / from-skip). + await this.transport.publish(msg.topic, env); return; } default: { @@ -178,4 +252,52 @@ export class Broker { } } } + + /** Should `me` receive `env`? Loop prevention (skip sender) + DM filter. */ + private shouldDeliver(me: string, env: Envelope): boolean { + if (env.from?.agentId === me) return false; // never echo to the sender + if (Array.isArray(env.to)) return env.to.includes(me); // DM: present `to` ⇒ only named targets ([] ⇒ nobody, never a broadcast) + return true; // broadcast / @mention (highlight is client-side via mentions[]) + } + + private addTopicMember(topic: string, id: string): void { + let m = this.topicMembers.get(topic); + if (!m) { + m = new Map(); + this.topicMembers.set(topic, m); + } + m.set(id, (m.get(id) ?? 0) + 1); + } + + private removeTopicMember(topic: string, id: string): void { + const m = this.topicMembers.get(topic); + if (!m) return; + const n = (m.get(id) ?? 0) - 1; + if (n <= 0) m.delete(id); + else m.set(id, n); + if (m.size === 0) this.topicMembers.delete(topic); + } + + private isReachable(topic: string, id: string): boolean { + return (this.topicMembers.get(topic)?.get(id) ?? 0) > 0; + } + + /** Persist a store_if_offline envelope for intended recipients with no live subscription (§3.2). */ + private async storeForOfflineRecipients(topic: string, env: Envelope, from?: string): Promise { + const intended = Array.isArray(env.to) ? env.to : await this.opts.store.getMembers(topic); + for (const id of intended) { + if (id === from) continue; // never store for the sender + if (!this.isReachable(topic, id)) { + await this.opts.store.enqueuePending(id, env); + } + } + } + + /** Reconnect replay (§3.2): drain + deliver everything queued for this identity. */ + private async drainPendingTo(ws: ServerWebSocket, id: string): Promise { + const pending = await this.opts.store.drainPending(id); + for (const env of pending) { + this.send(ws, { type: "event", topic: env.roomId, envelope: env }); + } + } } diff --git a/src/integration-test/broker-client.test.ts b/src/integration-test/broker-client.test.ts index dc4580d..7fe3dd2 100644 --- a/src/integration-test/broker-client.test.ts +++ b/src/integration-test/broker-client.test.ts @@ -12,7 +12,9 @@ async function startBroker() { const store = new InMemoryStore(); const svc = new IdentityService(store); await svc.registerIdentity("alice@x.com", "Alice"); + await svc.registerIdentity("bob@x.com", "Bob"); const token = await svc.issueToken("alice@x.com"); + const tokenB = await svc.issueToken("bob@x.com"); const broker = new Broker({ store, identityProvider: new StorePskIdentityProvider(store), @@ -21,11 +23,11 @@ async function startBroker() { log: () => {}, }); const { port } = broker.start(); - return { broker, store, token, url: `ws://127.0.0.1:${port}/ws` }; + return { broker, store, token, tokenB, url: `ws://127.0.0.1:${port}/ws` }; } describe("BrokerClient ↔ real Broker", () => { - test("connects, authenticates, and round-trips an event", async () => { + test("connects + authenticates; does NOT echo a client's own broadcast (loop prevention)", async () => { const { broker, store, token, url } = await startBroker(); const c = new BrokerClient({ url, token, log: () => {} }); try { @@ -36,7 +38,7 @@ describe("BrokerClient ↔ real Broker", () => { await sleep(60); // let subscribe + ack land c.publish("room-1", makeEnvelope({ messageId: "e1" })); await sleep(60); - expect(got).toEqual(["e1"]); + expect(got).toEqual([]); // the sender never receives its own broadcast } finally { c.close(); broker.stop(); @@ -44,10 +46,10 @@ describe("BrokerClient ↔ real Broker", () => { } }); - test("fans out across clients: A publishes, B receives", async () => { - const { broker, store, token, url } = await startBroker(); - const a = new BrokerClient({ url, token, log: () => {} }); - const b = new BrokerClient({ url, token, log: () => {} }); + test("fans out across clients: A publishes, B (different identity) receives", async () => { + const { broker, store, token, tokenB, url } = await startBroker(); + const a = new BrokerClient({ url, token, log: () => {} }); // alice + const b = new BrokerClient({ url, token: tokenB, log: () => {} }); // bob try { await a.connect(); await b.connect(); diff --git a/src/integration-test/broker-routing.test.ts b/src/integration-test/broker-routing.test.ts new file mode 100644 index 0000000..b8fcef0 --- /dev/null +++ b/src/integration-test/broker-routing.test.ts @@ -0,0 +1,358 @@ +import { describe, test, expect } from "bun:test"; +import { Broker } from "../broker"; +import { InMemoryStore } from "../backbone/store/memory-store"; +import { SqliteStore } from "../backbone/store/sqlite-store"; +import { IdentityService } from "../backbone/identity-service"; +import { StorePskIdentityProvider } from "../backbone/identity/store-psk-identity-provider"; +import { makeEnvelope } from "../unit-test/backbone-fixtures"; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +/** Buffering WS client so no inbound message is lost between awaits. */ +class WsClient { + ws!: WebSocket; + private q: any[] = []; + private waiters: ((m: any) => void)[] = []; + static async connect(url: string): Promise { + const c = new WsClient(); + c.ws = new WebSocket(url); + c.ws.onmessage = (ev) => { + const m = JSON.parse(ev.data as string); + const w = c.waiters.shift(); + if (w) w(m); + else c.q.push(m); + }; + await new Promise((res, rej) => { + c.ws.onopen = () => res(); + c.ws.onerror = () => rej(new Error("connect failed")); + }); + return c; + } + next(): Promise { + const m = this.q.shift(); + if (m !== undefined) return Promise.resolve(m); + return new Promise((r) => this.waiters.push(r)); + } + send(m: unknown) { + this.ws.send(JSON.stringify(m)); + } + /** All currently-buffered messages (no wait). */ + drainNow(): any[] { + const all = this.q; + this.q = []; + return all; + } + close() { + this.ws.close(); + } +} + +async function start() { + const store = new InMemoryStore(); + const svc = new IdentityService(store); + const ids = ["alice@x.com", "bob@x.com", "carol@x.com"] as const; + const token: Record = {}; + for (const id of ids) { + await svc.registerIdentity(id, id); + token[id] = await svc.issueToken(id); + } + const broker = new Broker({ + store, + identityProvider: new StorePskIdentityProvider(store), + host: "127.0.0.1", + port: 0, + log: () => {}, + }); + const { port } = broker.start(); + return { broker, store, token, url: `ws://127.0.0.1:${port}/ws` }; +} + +async function join(url: string, token: string, topic: string): Promise { + const c = await WsClient.connect(url); + c.send({ type: "hello", token }); + await c.next(); // welcome + c.send({ type: "subscribe", topic }); + await c.next(); // subscribed + return c; +} + +describe("Broker routing (§3.2): DM / broadcast / hop / offline replay", () => { + test("DM (`to`) reaches only the named target, not other room members", async () => { + const { broker, store, token, url } = await start(); + const bob = await join(url, token["bob@x.com"]!, "r"); + const carol = await join(url, token["carol@x.com"]!, "r"); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + alice.send({ + type: "publish", + topic: "r", + envelope: makeEnvelope({ messageId: "dm1", to: ["bob@x.com"], deliveryMode: "online_only" }), + }); + const got = await bob.next(); + expect(got).toMatchObject({ type: "event", envelope: { messageId: "dm1" } }); + await sleep(40); + expect(carol.drainNow()).toEqual([]); // carol (a room member) does NOT see the DM + } finally { + bob.close(); + carol.close(); + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("broadcast reaches all room members EXCEPT the sender (loop prevention)", async () => { + const { broker, store, token, url } = await start(); + const bob = await join(url, token["bob@x.com"]!, "r"); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + alice.send({ + type: "publish", + topic: "r", + envelope: makeEnvelope({ messageId: "bc1", deliveryMode: "online_only" }), + }); + expect(await bob.next()).toMatchObject({ type: "event", envelope: { messageId: "bc1" } }); + await sleep(40); + expect(alice.drainNow()).toEqual([]); // sender never receives its own broadcast + } finally { + bob.close(); + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("hop<=0 is dropped (multi-hop loop guard)", async () => { + const { broker, store, token, url } = await start(); + const bob = await join(url, token["bob@x.com"]!, "r"); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + alice.send({ + type: "publish", + topic: "r", + envelope: makeEnvelope({ messageId: "hop0", hop: 0, deliveryMode: "online_only" }), + }); + await sleep(40); + expect(bob.drainNow()).toEqual([]); // dropped, nobody receives + } finally { + bob.close(); + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("store_if_offline DM is queued for an offline target and drained on reconnect", async () => { + const { broker, store, token, url } = await start(); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + // bob is NOT connected — the DM must be persisted for him. + alice.send({ + type: "publish", + topic: "r", + envelope: makeEnvelope({ + roomId: "r", + messageId: "queued1", + to: ["bob@x.com"], + deliveryMode: "store_if_offline", + }), + }); + await sleep(40); + // bob connects → drains the queued DM on welcome + const bob = await WsClient.connect(url); + bob.send({ type: "hello", token: token["bob@x.com"]! }); + expect(await bob.next()).toMatchObject({ type: "welcome" }); + expect(await bob.next()).toMatchObject({ type: "event", envelope: { messageId: "queued1" } }); + bob.close(); + } finally { + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("rejects an envelope with a missing/non-object from (anti-spoof guard)", async () => { + const { broker, store, token, url } = await start(); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + const env = makeEnvelope({ messageId: "noFrom", deliveryMode: "online_only" }); + delete (env as { from?: unknown }).from; + alice.send({ type: "publish", topic: "r", envelope: env }); + expect(await alice.next()).toMatchObject({ type: "error" }); // rejected, never fanned out + } finally { + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("rejects a string `to` (would degrade DM to a substring match → leak)", async () => { + const { broker, store, token, url } = await start(); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + const env = makeEnvelope({ messageId: "strTo", deliveryMode: "online_only" }); + (env as { to?: unknown }).to = "bob@x.com"; // a string, not an array + alice.send({ type: "publish", topic: "r", envelope: env }); + expect(await alice.next()).toMatchObject({ type: "error" }); + } finally { + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("rejects a second hello on an authenticated socket (no identity rebind)", async () => { + const { broker, store, token, url } = await start(); + const c = await WsClient.connect(url); + try { + c.send({ type: "hello", token: token["alice@x.com"]! }); + expect(await c.next()).toMatchObject({ type: "welcome" }); + c.send({ type: "hello", token: token["bob@x.com"]! }); + expect(await c.next()).toMatchObject({ type: "error" }); + } finally { + c.close(); + broker.stop(); + await store.close(); + } + }); + + test("a DM sent while connected-but-not-subscribed is drained on subscribe (no gap loss)", async () => { + const { broker, store, token, url } = await start(); + const alice = await join(url, token["alice@x.com"]!, "r"); + const bob = await WsClient.connect(url); + bob.send({ type: "hello", token: token["bob@x.com"]! }); + await bob.next(); // welcome (drain empty — bob not subscribed yet) + try { + alice.send({ + type: "publish", + topic: "r", + envelope: makeEnvelope({ + roomId: "r", + messageId: "gap1", + to: ["bob@x.com"], + deliveryMode: "store_if_offline", + }), + }); + await sleep(40); + bob.send({ type: "subscribe", topic: "r" }); // now reachable → drains the gap-window DM + expect(await bob.next()).toMatchObject({ type: "subscribed" }); + expect(await bob.next()).toMatchObject({ type: "event", envelope: { messageId: "gap1" } }); + } finally { + alice.close(); + bob.close(); + broker.stop(); + await store.close(); + } + }); + + test("rejects an envelope missing idempotencyKey (offline-replay load-bearing field)", async () => { + const { broker, store, token, url } = await start(); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + const env = makeEnvelope({ + roomId: "r", + messageId: "noKey", + to: ["bob@x.com"], + deliveryMode: "store_if_offline", + }); + delete (env as { idempotencyKey?: unknown }).idempotencyKey; + alice.send({ type: "publish", topic: "r", envelope: env }); + expect(await alice.next()).toMatchObject({ type: "error" }); + } finally { + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("an empty `to: []` delivers to nobody (never degrades to a broadcast)", async () => { + const { broker, store, token, url } = await start(); + const bob = await join(url, token["bob@x.com"]!, "r"); + const alice = await join(url, token["alice@x.com"]!, "r"); + try { + alice.send({ + type: "publish", + topic: "r", + envelope: makeEnvelope({ roomId: "r", messageId: "emptyTo", to: [], deliveryMode: "online_only" }), + }); + await sleep(40); + expect(bob.drainNow()).toEqual([]); // an empty DM target list reaches nobody + } finally { + bob.close(); + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("offline DM round-trips through the production SqliteStore (closes the InMemoryStore blind spot)", async () => { + const store = new SqliteStore(":memory:"); + const svc = new IdentityService(store); + await svc.registerIdentity("alice@x.com", "Alice"); + await svc.registerIdentity("bob@x.com", "Bob"); + const tokA = await svc.issueToken("alice@x.com"); + const tokB = await svc.issueToken("bob@x.com"); + const broker = new Broker({ + store, + identityProvider: new StorePskIdentityProvider(store), + host: "127.0.0.1", + port: 0, + log: () => {}, + }); + const { port } = broker.start(); + const url = `ws://127.0.0.1:${port}/ws`; + const alice = await join(url, tokA, "r"); + try { + alice.send({ + type: "publish", + topic: "r", + envelope: makeEnvelope({ + roomId: "r", + messageId: "sql1", + idempotencyKey: "sk1", + to: ["bob@x.com"], + deliveryMode: "store_if_offline", + }), + }); + await sleep(40); + const bob = await WsClient.connect(url); + bob.send({ type: "hello", token: tokB }); + expect(await bob.next()).toMatchObject({ type: "welcome" }); + expect(await bob.next()).toMatchObject({ type: "event", envelope: { messageId: "sql1" } }); + bob.close(); + } finally { + alice.close(); + broker.stop(); + await store.close(); + } + }); + + test("a store error during reconnect-drain does NOT close the already-welcomed connection", async () => { + const store = new InMemoryStore(); + const svc = new IdentityService(store); + await svc.registerIdentity("alice@x.com", "Alice"); + const token = await svc.issueToken("alice@x.com"); + (store as { drainPending: unknown }).drainPending = async () => { + throw new Error("boom"); + }; + const broker = new Broker({ + store, + identityProvider: new StorePskIdentityProvider(store), + host: "127.0.0.1", + port: 0, + log: () => {}, + }); + const { port } = broker.start(); + const c = await WsClient.connect(`ws://127.0.0.1:${port}/ws`); + try { + c.send({ type: "hello", token }); + expect(await c.next()).toMatchObject({ type: "welcome" }); // welcomed despite drain error + c.send({ type: "subscribe", topic: "r" }); + expect(await c.next()).toMatchObject({ type: "subscribed" }); // still usable, not closed + } finally { + c.close(); + broker.stop(); + await store.close(); + } + }); +}); diff --git a/src/integration-test/broker.test.ts b/src/integration-test/broker.test.ts index fbac71c..96a4141 100644 --- a/src/integration-test/broker.test.ts +++ b/src/integration-test/broker.test.ts @@ -56,7 +56,9 @@ async function startBroker() { const store = new InMemoryStore(); const svc = new IdentityService(store); await svc.registerIdentity("alice@x.com", "Alice"); + await svc.registerIdentity("bob@x.com", "Bob"); const token = await svc.issueToken("alice@x.com"); + const tokenB = await svc.issueToken("bob@x.com"); const broker = new Broker({ store, identityProvider: new StorePskIdentityProvider(store), @@ -65,7 +67,7 @@ async function startBroker() { log: () => {}, }); const { port } = broker.start(); - return { broker, store, token, url: `ws://127.0.0.1:${port}/ws` }; + return { broker, store, token, tokenB, url: `ws://127.0.0.1:${port}/ws` }; } describe("Broker — WSS + PSK auth + transport fan-out", () => { @@ -91,15 +93,15 @@ describe("Broker — WSS + PSK auth + transport fan-out", () => { } }); - test("publish fans out the envelope to a subscriber (control-plane only)", async () => { - const { broker, store, token, url } = await startBroker(); + test("publish fans out the envelope to a (different-identity) subscriber", async () => { + const { broker, store, token, tokenB, url } = await startBroker(); try { const pub = await WsClient.connect(url); - pub.send({ type: "hello", token }); + pub.send({ type: "hello", token }); // alice publishes await pub.next(); // welcome const sub = await WsClient.connect(url); - sub.send({ type: "hello", token }); + sub.send({ type: "hello", token: tokenB }); // bob subscribes await sub.next(); // welcome sub.send({ type: "subscribe", topic: "room-1" }); expect(await sub.next()).toEqual({ type: "subscribed", topic: "room-1" });