diff --git a/apps/cli/src/daemon/db/inbox.ts b/apps/cli/src/daemon/db/inbox.ts index 496e33e..ccdd879 100644 --- a/apps/cli/src/daemon/db/inbox.ts +++ b/apps/cli/src/daemon/db/inbox.ts @@ -111,6 +111,21 @@ export function insertIfNew( return after?.id === row.id ? row.id : null; } +/** 1.38.0: the row a client_message_id already landed as, if any. */ +export function findByClientMessageId( + db: SqliteDb, + clientMessageId: string, +): Pick | null { + return db.prepare( + `SELECT id, body, recipient_pubkey, recipient_kind FROM inbox WHERE client_message_id = ?`, + ).get>(clientMessageId) ?? null; +} + +/** 1.38.0: widen a session-scoped row to the whole member (multicast). */ +export function promoteToMember(db: SqliteDb, id: string, memberPubkey: string): void { + db.prepare(`UPDATE inbox SET recipient_kind = 'member', recipient_pubkey = ? WHERE id = ?`).run(memberPubkey, id); +} + export interface ListInboxParams { since?: number; // received_at >= since topic?: string; diff --git a/apps/cli/src/daemon/events.ts b/apps/cli/src/daemon/events.ts index 9f95525..c842f57 100644 --- a/apps/cli/src/daemon/events.ts +++ b/apps/cli/src/daemon/events.ts @@ -98,6 +98,8 @@ export function shouldDeliver(e: DaemonEvent, f: SseFilterOptions): boolean { return !!f.sessionPubkey && f.sessionPubkey.toLowerCase() === recipientPubkey; } if (recipientKind === "member") { + const excluded = typeof e.data.exclude_session_pubkey === "string" ? e.data.exclude_session_pubkey.toLowerCase() : null; + if (excluded && f.sessionPubkey && f.sessionPubkey.toLowerCase() === excluded) return false; return !!f.memberPubkey && f.memberPubkey.toLowerCase() === recipientPubkey; } return true; diff --git a/apps/cli/src/daemon/inbound.ts b/apps/cli/src/daemon/inbound.ts index 424ad2c..51e63ff 100644 --- a/apps/cli/src/daemon/inbound.ts +++ b/apps/cli/src/daemon/inbound.ts @@ -5,7 +5,7 @@ import { randomUUID } from "node:crypto"; import type { SqliteDb } from "./db/sqlite.js"; -import { insertIfNew } from "./db/inbox.js"; +import { findByClientMessageId, insertIfNew, promoteToMember } from "./db/inbox.js"; import type { EventBus } from "./events.js"; import { decryptDirect } from "~/services/crypto/facade.js"; import { decodeBase64Utf8 } from "~/utils/text.js"; @@ -118,6 +118,41 @@ export async function handleBrokerPush(msg: Record, ctx: Inboun } } + // 1.38.0 (spec 2026-09-26 §3): a multicast (broadcast / group / topic) + // reaches this daemon twice — on the member-WS (visible to every session) + // and on a session-WS (visible to that session). Dedupe used to keep + // whichever landed first, so a session copy winning hid the message from + // every other session. Converge on the member-wide row instead. + const existing = ctx.recipientKind + ? findByClientMessageId(ctx.db, clientMessageId) + : null; + if (existing && existing.recipient_kind && existing.recipient_kind !== ctx.recipientKind) { + ctx.ackClientMessage?.(clientMessageId, brokerMessageId); + if (ctx.recipientKind === "member" && existing.recipient_kind === "session" && ctx.recipientPubkey) { + promoteToMember(ctx.db, existing.id, ctx.recipientPubkey); + ctx.bus.publish("message", { + id: existing.id, + mesh: ctx.meshSlug, + client_message_id: clientMessageId, + broker_message_id: brokerMessageId, + sender_pubkey: senderPubkey, + sender_member_pubkey: senderMemberPk, + sender_name: senderName, + topic, + reply_to_id: replyToId, + priority, + ...(subtype ? { subtype } : {}), + body: existing.body, + created_at: createdAt, + recipient_pubkey: ctx.recipientPubkey, + recipient_kind: "member", + // the session that already rendered it must not see it twice + ...(existing.recipient_pubkey ? { exclude_session_pubkey: existing.recipient_pubkey } : {}), + }); + } + return; + } + const id = randomUUID(); const inserted = insertIfNew(ctx.db, { id, diff --git a/apps/cli/tests/unit/inbound-multicast.test.ts b/apps/cli/tests/unit/inbound-multicast.test.ts new file mode 100644 index 0000000..f523e39 --- /dev/null +++ b/apps/cli/tests/unit/inbound-multicast.test.ts @@ -0,0 +1,51 @@ +/** Spec 2026-09-26 §3: a broadcast arriving on both the member-WS and a + * session-WS must reach every session exactly once, whatever lands first. */ + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { openSqlite, type SqliteDb } from "~/daemon/db/sqlite.js"; +import { migrateInbox } from "~/daemon/db/inbox.js"; +import { EventBus, shouldDeliver, type DaemonEvent } from "~/daemon/events.js"; +import { handleBrokerPush } from "~/daemon/inbound.js"; + +const MEMBER = "a".repeat(64); +const S1 = "1".repeat(64); +const S2 = "2".repeat(64); +const push = { type: "push", messageId: "b1", client_message_id: "cid-1", ciphertext: Buffer.from("hola").toString("base64") }; + +describe("inbound: multicast convergence", () => { + let dir: string; + let db: SqliteDb; + let events: DaemonEvent[]; + let bus: EventBus; + beforeEach(async () => { + dir = mkdtempSync(join(tmpdir(), "cmh-mc-")); + db = await openSqlite(join(dir, "inbox.db")); + migrateInbox(db); + events = []; + bus = new EventBus(); + bus.subscribe((e) => { if (e.kind === "message") events.push(e); }); + }); + afterEach(() => { try { db.close(); } catch { /* ignore */ } rmSync(dir, { recursive: true, force: true }); }); + + const seenBy = (session: string) => + events.filter((e) => shouldDeliver(e, { sessionPubkey: session, memberPubkey: MEMBER, meshSlug: "m" })).length; + + it("session copy first, member copy second → every session sees it once", async () => { + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: S1, recipientKind: "session" }); + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: MEMBER, recipientKind: "member" }); + expect(seenBy(S1)).toBe(1); + expect(seenBy(S2)).toBe(1); + }); + + it("member copy first → the session copy is dropped", async () => { + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: MEMBER, recipientKind: "member" }); + await handleBrokerPush(push, { db, bus, meshSlug: "m", recipientPubkey: S1, recipientKind: "session" }); + expect(seenBy(S1)).toBe(1); + expect(seenBy(S2)).toBe(1); + expect(db.prepare("SELECT COUNT(*) AS n FROM inbox").get()).toEqual({ n: 1 }); + }); +});