From 6193f17ea3900cef328478665d8e33289b5c99d8 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 16 Sep 2026 04:00:32 +0100 Subject: [PATCH] feat(core): wire sendRoomMessage's opt-in durable-recording flag sendRoomMessage gains an optional durable parameter; when true, it also records the message as a room-notice in the sender's own oplog via recordRoomSendNotice (#135), alongside the existing wire- authenticated fan-out. Deliberately opt-in per call, not automatic -- a caller decides per message whether it should be catch-up-able, matching this codebase's own established design for the mechanism. MeshStoreIdentity gains a dataStorage field (a KeyValueStorage instance), the same one WireMeshTransport's own frame responder (#136) reads and writes, so a durable send and a peer's later data-request against this device's log operate on one shared oplog. bridge-mesh.ts constructs it once via createNodeFsStorage(oplogDirFor(slot)) and wires it to both the transport and the store's own identity. Every existing fake MeshStoreIdentity fixture across the test suite needed the new required field added -- an opaque placeholder for the many tests that never exercise durable sends, a real in-memory KeyValueStorage for the ones (room-messaging-durable-send.test.ts, test-transport.ts's shared harness) that do. --- src/core/bridge-mesh.ts | 7 +- src/core/mesh-store-shared.ts | 4 +- src/core/room-messaging.ts | 18 ++- src/test/delivery-engine-delivery.test.ts | 1 + src/test/delivery-engine.test.ts | 1 + src/test/identity-restart.integration.test.ts | 2 + src/test/room-join-admission.test.ts | 2 + src/test/room-lifecycle-membership.test.ts | 1 + src/test/room-lifecycle-remote.test.ts | 1 + src/test/room-lifecycle.test.ts | 1 + src/test/room-messaging-durable-send.test.ts | 151 ++++++++++++++++++ src/test/room-messaging.test.ts | 1 + src/test/room-protocol-admission.test.ts | 1 + src/test/room-protocol.test.ts | 1 + src/test/room-send-retry.integration.test.ts | 2 + src/test/test-transport.ts | 6 + 16 files changed, 197 insertions(+), 3 deletions(-) create mode 100644 src/test/room-messaging-durable-send.test.ts diff --git a/src/core/bridge-mesh.ts b/src/core/bridge-mesh.ts index e8ae461d..27bf8b70 100644 --- a/src/core/bridge-mesh.ts +++ b/src/core/bridge-mesh.ts @@ -9,10 +9,11 @@ import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; import { createRevocationView } from "wire-mesh-core/domain/revocation-view"; +import { createNodeFsStorage } from "wire-mesh-core/adapters/node-fs-storage"; import { MeshStore } from "./mesh-store.js"; import { CommsTool } from "./tool.js"; import { WireMeshTransport } from "./wire-mesh-transport.js"; -import { loadOrCreateIdentity } from "./identity-store.js"; +import { loadOrCreateIdentity, oplogDirFor } from "./identity-store.js"; import type { IdentitySlot } from "./identity-store.js"; import { toIdentityPort } from "./wire-mesh-identity.js"; @@ -34,6 +35,8 @@ export function createBridgeMeshSync( const identity = loadOrCreateIdentity(slot); const store = new MeshStore(coordinatorPort); store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); + // One shared dataStorage instance for both the transport's own data-domain frame responder and the store's own durable-send mint path (P5, agent-comms#50) -- oplogDirFor(slot) needs only the slot, not the async identity below, so this can be constructed synchronously right here. + const dataStorage = createNodeFsStorage({ dir: oplogDirFor(slot) }); store.setTransport( new WireMeshTransport( store.events, @@ -43,6 +46,7 @@ export function createBridgeMeshSync( () => store.selfStatus, undefined, () => store.hostedRooms, + dataStorage, ), ); const tool = new CommsTool(store, store.discovery); @@ -56,6 +60,7 @@ export function createBridgeMeshSync( clock: createSystemClock(), slot, revocation, + dataStorage, }); }, }; diff --git a/src/core/mesh-store-shared.ts b/src/core/mesh-store-shared.ts index 2c502cb8..7e69d7e0 100644 --- a/src/core/mesh-store-shared.ts +++ b/src/core/mesh-store-shared.ts @@ -5,14 +5,16 @@ import type { Clock } from "wire-mesh-core/ports/clock"; import type { IdentityPort } from "wire-mesh-core/ports/identity"; import type { RevocationView } from "wire-mesh-core/domain/revocation-view"; +import type { KeyValueStorage } from "wire-mesh-core/ports/storage"; import type { IdentitySlot } from "./identity-store.js"; -/** The identity/clock/persistence collaborators MeshStore mints and persists room-membership grants against. Set via setIdentity(), mirroring the transport's own setTransport() contract. */ +/** The identity/clock/persistence collaborators MeshStore mints and persists room-membership grants against. Set via setIdentity(), mirroring the transport's own setTransport() contract. dataStorage backs this device's own room-notice oplog (P5, agent-comms#50) -- the same KeyValueStorage instance WireMeshTransport's own dataStorage constructor parameter is wired with, so a durable sendRoomMessage and the transport's own data-domain responder read and write the identical log. */ export interface MeshStoreIdentity { identity: IdentityPort; clock: Clock; slot: IdentitySlot; revocation: RevocationView; + dataStorage: KeyValueStorage; } /** diff --git a/src/core/room-messaging.ts b/src/core/room-messaging.ts index aafcb09e..493185c6 100644 --- a/src/core/room-messaging.ts +++ b/src/core/room-messaging.ts @@ -6,6 +6,7 @@ import { bytesFromHex, bytesToHex } from "wire-mesh-core/domain/device-id"; import { dmRoomPath } from "./room-path.js"; import { loadRoomTokens } from "./identity-store.js"; import { randomId } from "./random-id.js"; +import { recordRoomSendNotice } from "./room-notice-log.js"; import { CommsError } from "./store.js"; import type { MeshStoreIdentity } from "./mesh-store-shared.js"; import type { RoomProtocol } from "./room-protocol.js"; @@ -34,6 +35,8 @@ export class RoomMessaging { /** * Sends a room message via a real, wire-authenticated room.send fan-out (P3.5): one directed request per member, each carrying this sender's own persisted room:member token, rather than the legacy broadcastPatch's full-state replication. A member unreachable right now is queued for retry (see sendRoomRequestToMember/flushPendingRoomRequests) instead of blocking or failing the whole send -- delivery to any one recipient is independent of every other. + * + * durable, when true, additionally records this same message as a room-notice in the sender's own oplog via recordRoomSendNotice (P5, agent-comms#50) -- deliberately opt-in per call, not automatic: matching this codebase's own design principle that delivery and durable catch-up are the same artifact only when a caller actually opts a message into it. A caller that wants an offline member to be able to catch up on this specific message later passes true; every other send stays exactly as before. */ async sendRoomMessage( roomId: string, @@ -41,6 +44,7 @@ export class RoomMessaging { content: string, replyTo?: string, streamingBehavior?: StreamingBehavior, + durable?: boolean, ): Promise { const room = this.deps.rooms.get(roomId); if (!room) @@ -48,12 +52,24 @@ export class RoomMessaging { if (!room.members.includes(from)) throw new CommsError(`Not a member of ${roomId}`, "NOT_MEMBER"); - const { slot, clock } = this.deps.requireIdentity(); + const { slot, clock, identity, dataStorage } = this.deps.requireIdentity(); const token = loadRoomTokens(slot)[roomId]; if (token === undefined) { throw new CommsError(`No room:member token for ${roomId}`, "NOT_MEMBER"); } + if (durable === true) { + await recordRoomSendNotice( + { identity, clock, storage: dataStorage }, + { + room: roomId, + token, + contentType: "text/plain", + content: new TextEncoder().encode(content), + }, + ); + } + const messageId = randomId(); const id = bytesToHex(messageId); const message: RoomMessage = { diff --git a/src/test/delivery-engine-delivery.test.ts b/src/test/delivery-engine-delivery.test.ts index 4366de09..8b60a1cd 100644 --- a/src/test/delivery-engine-delivery.test.ts +++ b/src/test/delivery-engine-delivery.test.ts @@ -150,6 +150,7 @@ function makeHarness() { clock: { now: () => NOW_MS }, identity: {} as never, revocation: { record: revocationRecord } as never, + dataStorage: {} as never, }), requireTransport: () => transport, getOnDelivery: () => onDelivery, diff --git a/src/test/delivery-engine.test.ts b/src/test/delivery-engine.test.ts index cf9ee68e..7fe5bd38 100644 --- a/src/test/delivery-engine.test.ts +++ b/src/test/delivery-engine.test.ts @@ -155,6 +155,7 @@ function makeHarness() { clock: { now: () => NOW_MS }, identity: {} as never, revocation: { record: revocationRecord } as never, + dataStorage: {} as never, }), requireTransport: () => transport, getOnDelivery: () => onDelivery, diff --git a/src/test/identity-restart.integration.test.ts b/src/test/identity-restart.integration.test.ts index 7b8565f7..a5fdf3fe 100644 --- a/src/test/identity-restart.integration.test.ts +++ b/src/test/identity-restart.integration.test.ts @@ -11,6 +11,7 @@ import { test, expect } from "vitest"; import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; import { createRevocationView } from "wire-mesh-core/domain/revocation-view"; +import { createMemoryStorage } from "wire-mesh-core/adapters/memory-storage"; import { MeshStore } from "../core/mesh-store.js"; import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import type { PeerIdentity } from "../core/identity.js"; @@ -55,6 +56,7 @@ async function makePeer( clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: createMemoryStorage(), }); const deliveries: DeliveryEvent[] = []; return { store, deliveries }; diff --git a/src/test/room-join-admission.test.ts b/src/test/room-join-admission.test.ts index 31360521..9224199b 100644 --- a/src/test/room-join-admission.test.ts +++ b/src/test/room-join-admission.test.ts @@ -18,6 +18,7 @@ import { } from "wire-mesh-core/domain/device-id"; import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; import { createRevocationView } from "wire-mesh-core/domain/revocation-view"; +import { createMemoryStorage } from "wire-mesh-core/adapters/memory-storage"; import { mintCapabilityToken } from "wire-mesh-core/domain/tokens"; import { MeshStore } from "../core/mesh-store.js"; import { ownerNamedRoomPath } from "../core/room-path.js"; @@ -185,6 +186,7 @@ describe("joinRoom (requester side, remote path)", () => { clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: createMemoryStorage(), }); const ownerId = "f".repeat(DEVICE_ID_HEX_LENGTH); diff --git a/src/test/room-lifecycle-membership.test.ts b/src/test/room-lifecycle-membership.test.ts index 8607dc8d..ce96a7d8 100644 --- a/src/test/room-lifecycle-membership.test.ts +++ b/src/test/room-lifecycle-membership.test.ts @@ -174,6 +174,7 @@ async function makeHarness(): Promise { clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: {} as never, }), requireTransport: () => ({ sendRoomRequest, broadcastRevocation }) as unknown as ReturnType< diff --git a/src/test/room-lifecycle-remote.test.ts b/src/test/room-lifecycle-remote.test.ts index 740d6697..8ab9df49 100644 --- a/src/test/room-lifecycle-remote.test.ts +++ b/src/test/room-lifecycle-remote.test.ts @@ -172,6 +172,7 @@ async function makeHarness(): Promise { clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: {} as never, }), requireTransport: () => ({ sendRoomRequest, broadcastRevocation }) as unknown as ReturnType< diff --git a/src/test/room-lifecycle.test.ts b/src/test/room-lifecycle.test.ts index c3751587..98849ac2 100644 --- a/src/test/room-lifecycle.test.ts +++ b/src/test/room-lifecycle.test.ts @@ -170,6 +170,7 @@ async function makeHarness(): Promise { clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: {} as never, }), requireTransport: () => ({ sendRoomRequest, broadcastRevocation }) as unknown as ReturnType< diff --git a/src/test/room-messaging-durable-send.test.ts b/src/test/room-messaging-durable-send.test.ts new file mode 100644 index 00000000..765c0506 --- /dev/null +++ b/src/test/room-messaging-durable-send.test.ts @@ -0,0 +1,151 @@ +/** + * sendRoomMessage's opt-in durable flag (P5's mint-and-append half, agent-comms#50, wired via recordRoomSendNotice from #135): a caller that passes durable: true also gets the message recorded as a room-notice in the sender's own oplog, verifiable end to end via a real verifyRoomNotice, not just an opaque append. Real identities and a real minted token throughout, matching room-lifecycle.test.ts's own established convention for this class of test -- recordRoomSendNotice genuinely mints and signs, so an opaque placeholder token would fail the mint outright. + */ +import { describe, expect, it } from "vitest"; +import * as fs from "node:fs"; +import * as path from "node:path"; +import { tmpdir } from "node:os"; +import { decode as cborDecode, cdeDecodeOptions } from "cbor2"; +import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; +import { mintCapabilityToken } from "wire-mesh-core/domain/tokens"; +import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; +import { createMemoryStorage } from "wire-mesh-core/adapters/memory-storage"; +import { ownerNamedRoomPath } from "wire-mesh-core/domain/room-path"; +import { verifyRoomNotice } from "wire-mesh-core/domain/room"; +import { createRevocationView } from "wire-mesh-core/domain/revocation-view"; +import { readEntries } from "wire-mesh-core/domain/data-sync"; +import type { RoomNotice } from "wire-mesh-core/generated/protocol"; +import { generateIdentity } from "../core/identity.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; +import { loadOrCreateIdentity, saveRoomToken } from "../core/identity-store.js"; +import { randomId } from "../core/random-id.js"; +import { + RoomMessaging, + type RoomMessagingDeps, +} from "../core/room-messaging.js"; +import type { AgentIdentity, Room } from "../core/types.js"; + +const NO_DELEGATIONS_REMAINING = 0; +const MINUTES_PER_HOUR = 60; +const SECONDS_PER_MINUTE = 60; +const MS_PER_SECOND = 1000; +const ONE_HOUR_MS = MINUTES_PER_HOUR * SECONDS_PER_MINUTE * MS_PER_SECOND; + +async function makeHarness() { + const ownerIdentity = await toIdentityPort(generateIdentity()); + const roomPath = ownerNamedRoomPath( + deviceIdToHex(ownerIdentity.deviceId), + "general", + ); + const clock = createSystemClock(); + const verdict = await mintCapabilityToken({ + identity: ownerIdentity, + clock, + tokenId: randomId(), + bearer: ownerIdentity.deviceId, + capability: "room:member", + scope: { kind: "room", path: roomPath }, + expires: Date.now() + ONE_HOUR_MS, + delegationsRemaining: NO_DELEGATIONS_REMAINING, + }); + if (!verdict.ok) throw new Error(`mint failed: ${verdict.reason}`); + + // saveRoomToken writes into this slot's own persisted identity file, which must already exist on disk -- unrelated to which crypto identity requireIdentity() uses for minting/verifying (room-lifecycle.test.ts's own established convention for this exact precondition). + const slotDir = fs.mkdtempSync( + path.join(tmpdir(), "room-messaging-durable-send-test-"), + ); + const slot = { harness: "test", cwd: "room-messaging", dir: slotDir }; + loadOrCreateIdentity(slot); + saveRoomToken(slot, roomPath, verdict.token); + + const dataStorage = createMemoryStorage(); + const room: Room = { + id: roomPath, + version: 1, + name: "general", + type: "public", + owner: deviceIdToHex(ownerIdentity.deviceId), + createdAt: "2026-01-01T00:00:00.000Z", + description: "", + members: [deviceIdToHex(ownerIdentity.deviceId)], + invited: [], + memberJoins: {}, + memberLeaves: {}, + invitedJoins: {}, + invitedLeaves: {}, + }; + + const deps: RoomMessagingDeps = { + rooms: new Map([[roomPath, room]]), + messages: new Map(), + dms: new Map(), + agents: new Map(), + requireIdentity: () => ({ + slot, + clock, + identity: ownerIdentity, + revocation: createRevocationView(), + dataStorage, + }), + roomProtocol: { sendRoomRequestToMember: async () => undefined }, + federation: { forwardRoomMessage: async () => undefined }, + }; + + return { + messaging: new RoomMessaging(deps), + ownerIdentity, + roomPath, + dataStorage, + }; +} + +describe("RoomMessaging — sendRoomMessage durable flag", () => { + it("records nothing in the oplog when durable is omitted", async () => { + const h = await makeHarness(); + const ownerDeviceHex = deviceIdToHex(h.ownerIdentity.deviceId); + await h.messaging.sendRoomMessage(h.roomPath, ownerDeviceHex, "hello"); + + expect( + await readEntries(h.dataStorage, h.ownerIdentity.deviceId, 0), + ).toEqual([]); + }); + + it("appends a verifiable room-notice to the sender's own oplog when durable is true", async () => { + const h = await makeHarness(); + const ownerDeviceHex = deviceIdToHex(h.ownerIdentity.deviceId); + + await h.messaging.sendRoomMessage( + h.roomPath, + ownerDeviceHex, + "hello, durably", + undefined, + undefined, + true, + ); + + const [entry] = await readEntries( + h.dataStorage, + h.ownerIdentity.deviceId, + 0, + ); + expect(entry).toBeDefined(); + if (entry === undefined) return; + + const decoded = cborDecode(entry, cdeDecodeOptions); + const verdict = await verifyRoomNotice(decoded, { + identity: h.ownerIdentity, + clock: createSystemClock(), + revocation: createRevocationView(), + }); + + expect( + verdict.ok, + `expected the durably-recorded notice to verify, got ${JSON.stringify(verdict)}`, + ).toBeTruthy(); + if (!verdict.ok) return; + expect(verdict.claims.room).toBe(h.roomPath); + expect(new TextDecoder().decode(verdict.claims.content)).toBe( + "hello, durably", + ); + }); +}); diff --git a/src/test/room-messaging.test.ts b/src/test/room-messaging.test.ts index 44ef5bf4..c6393851 100644 --- a/src/test/room-messaging.test.ts +++ b/src/test/room-messaging.test.ts @@ -77,6 +77,7 @@ function makeHarness() { clock: { now: () => NOW_MS }, identity: {} as never, revocation: {} as never, + dataStorage: {} as never, }), roomProtocol: { sendRoomRequestToMember }, federation: { forwardRoomMessage }, diff --git a/src/test/room-protocol-admission.test.ts b/src/test/room-protocol-admission.test.ts index 4ad59436..16d1cd24 100644 --- a/src/test/room-protocol-admission.test.ts +++ b/src/test/room-protocol-admission.test.ts @@ -163,6 +163,7 @@ async function makeHarness(): Promise { clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: {} as never, }), requireTransport: () => ({ sendRoomRequest }) as unknown as ReturnType< diff --git a/src/test/room-protocol.test.ts b/src/test/room-protocol.test.ts index 734f5667..80025756 100644 --- a/src/test/room-protocol.test.ts +++ b/src/test/room-protocol.test.ts @@ -161,6 +161,7 @@ async function makeHarness(): Promise { clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: {} as never, }), requireTransport: () => ({ sendRoomRequest }) as unknown as ReturnType< diff --git a/src/test/room-send-retry.integration.test.ts b/src/test/room-send-retry.integration.test.ts index ca878ddd..d4a912d6 100644 --- a/src/test/room-send-retry.integration.test.ts +++ b/src/test/room-send-retry.integration.test.ts @@ -9,6 +9,7 @@ import { test, expect } from "vitest"; import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; import { createRevocationView } from "wire-mesh-core/domain/revocation-view"; +import { createMemoryStorage } from "wire-mesh-core/adapters/memory-storage"; import { MeshStore } from "../core/mesh-store.js"; import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import { @@ -59,6 +60,7 @@ async function makePeer( clock: createSystemClock(), slot, revocation: createRevocationView(), + dataStorage: createMemoryStorage(), }); const deliveries: DeliveryEvent[] = []; return { store, deliveries }; diff --git a/src/test/test-transport.ts b/src/test/test-transport.ts index 06181ecf..e913ad73 100644 --- a/src/test/test-transport.ts +++ b/src/test/test-transport.ts @@ -7,6 +7,7 @@ import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; import { createSystemClock } from "wire-mesh-core/adapters/system-clock"; import { createRevocationView } from "wire-mesh-core/domain/revocation-view"; +import { createMemoryStorage } from "wire-mesh-core/adapters/memory-storage"; import { loadOrCreateIdentity } from "../core/identity-store.js"; import type { IdentitySlot } from "../core/identity-store.js"; import { toIdentityPort } from "../core/wire-mesh-identity.js"; @@ -31,6 +32,8 @@ export async function wireTestTransport( const identity = loadOrCreateIdentity(resolvedSlot); // Every real bridge sets peerId to deviceIdToHex(identity.deviceId) before wiring the transport (createBridgeMesh) -- WireMeshTransport's own session bookkeeping is keyed by device-id, so a peer's advertised ID and the identity the other side actually authenticates the connection against must be the same value, or introduction/state-sync never recognises the peer as itself. store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); + // One shared dataStorage instance for both the transport's own data-domain frame responder and the store's own durable-send mint path (P5, agent-comms#50) -- memory-backed, matching every other throwaway test identity here, rather than a real createNodeFsStorage a test would need to clean up afterwards. + const dataStorage = createMemoryStorage(); store.setTransport( new WireMeshTransport( store.events, @@ -39,6 +42,8 @@ export async function wireTestTransport( pendingConnectionTimeoutMs, () => store.selfStatus, presenceReadvertiseIntervalMs, + undefined, + dataStorage, ), ); store.setIdentity({ @@ -46,6 +51,7 @@ export async function wireTestTransport( clock: createSystemClock(), slot: resolvedSlot, revocation: createRevocationView(), + dataStorage, }); // Surface transport-level errors instead of leaving them silent — a genuine socket failure during a test run is signal worth seeing even when the test's own assertions still pass, since it can point at a real race the assertions don't happen to catch. store.onError = (e) => {