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) => {