diff --git a/ts/packages/core/src/domain/mesh-session.ts b/ts/packages/core/src/domain/mesh-session.ts index 9a8fc689..4e037501 100644 --- a/ts/packages/core/src/domain/mesh-session.ts +++ b/ts/packages/core/src/domain/mesh-session.ts @@ -800,6 +800,11 @@ export function createMeshSession( addresses: readonly string[] = [], /** Fired for every peer-advert entry as it's applied to the directory, regardless of source -- the hook a gossip-expansion consumer (wire-mesh#187, gossip-expansion.ts's own createGossipExpansion) uses to observe newly-gossiped peers without becoming a second, competing consumer of this session's own single-reader events stream (each emitted SessionEvent wakes at most one waiter, so a second for-await loop over events would silently steal events from whichever consumer already reads it). Omit for a caller with no use for it, exactly today's behaviour. */ onPeerAdvert?: (advert: PeerAdvert) => void, + /** Fired once per frame received on this session's connection, with the Connection it arrived on: a new Connection after a reconnect is how a caller notices it has to announce itself again, and a core/data reply is how it receives a log it asked a hub for. Omit for a caller with no use for it, exactly today's behaviour. */ + onFrame?: ( + connection: Readonly, + frame: Frame, + ) => void | Promise, ): MeshSession { const { session } = createSessionCore( identity, @@ -808,6 +813,7 @@ export function createMeshSession( async (address) => transport.connect(address), onPeerAdvert, addresses, + onFrame, ); return session; } diff --git a/ts/packages/core/src/domain/notice-board.ts b/ts/packages/core/src/domain/notice-board.ts index 7a213447..559d7144 100644 --- a/ts/packages/core/src/domain/notice-board.ts +++ b/ts/packages/core/src/domain/notice-board.ts @@ -175,11 +175,12 @@ export function createNoticeBoard( async function readLog( peer: DeviceId, room: string, + dmRootPolicy: "self" | "either-participant", ): Promise { const entries = await readEntries(storage, peer, 0); const out: NoticeBoardEntry[] = []; for (const entry of entries) { - out.push(await readOne(entry, room)); + out.push(await readOne(entry, room, dmRootPolicy)); } return out; } @@ -187,6 +188,7 @@ export function createNoticeBoard( async function readOne( entry: Uint8Array, room: string, + dmRootPolicy: "self" | "either-participant", ): Promise { const notice = decodeNotice(entry); if (notice === undefined) { @@ -197,6 +199,7 @@ export function createNoticeBoard( clock, revocation, expectedRoom: room, + dmRootPolicy, }); if (!verdict.ok) { return { verified: false, reason: verdict.reason }; @@ -246,8 +249,10 @@ export function createNoticeBoard( return { postEncryptedNotice, ingestPeerEntry, - readOwnNotices: async (room: string) => readLog(identity.deviceId, room), + // A device's own token in a DM is rooted at the other participant, so its own log is read under the either-participant policy; another device's notices must be rooted at this identity. + readOwnNotices: async (room: string) => + readLog(identity.deviceId, room, "either-participant"), readPeerNotices: async (peer: DeviceId, room: string) => - readLog(peer, room), + readLog(peer, room, "self"), }; } diff --git a/ts/packages/core/src/domain/room-rekey.ts b/ts/packages/core/src/domain/room-rekey.ts index 3028fdd7..142bd866 100644 --- a/ts/packages/core/src/domain/room-rekey.ts +++ b/ts/packages/core/src/domain/room-rekey.ts @@ -44,7 +44,7 @@ export function buildRoomRekeyCommand( /** Sends a room.rekey for roomPath. targetDevice/token forward directly to MeshSession.sendManageRequest's own identically-named parameters, exactly as sendRoomMessage's own doc comment already describes. */ export async function sendRoomRekey( - session: Readonly, + session: Readonly>, roomPath: string, keyEpoch: number, wrappedKey: Uint8Array | readonly Uint8Array[], diff --git a/ts/packages/core/src/domain/room.ts b/ts/packages/core/src/domain/room.ts index 95b59285..affec857 100644 --- a/ts/packages/core/src/domain/room.ts +++ b/ts/packages/core/src/domain/room.ts @@ -90,6 +90,10 @@ export interface VerifyRoomNoticeOptions { * When given, refuses any notice not claiming exactly this room -- "is this the room I actually asked to read", checked against the notice's own self-declared `room` field before any cryptographic work, so a caller scanning a mixed stream of notices can cheaply skip ones for other rooms. Independent of, and layered on top of, this function's own unconditional internal self-consistency check (the embedded token's scope.path MUST equal the notice's own `room` field regardless of whether expectedRoom is given at all) -- a notice can be internally self-consistent yet still be for a room other than the one a caller expected, and this is the option that catches that case. Mirrors verifyCapabilityToken's own optional expectedBearer for the same "also assert it matches what I expected" shape. */ expectedRoom?: RoomPath; + /** + * How a DM's embedded token must be rooted, passed to verifyRoomToken (see its own `dmRootPolicy`). The default requires the root to be the verifying identity, which is right for a notice another device posted to me. A device reading its own notice holds a token rooted at the other participant, so only "either-participant" can pass there; it is safe because nothing about an own notice rests on the root rule, the device wrote it. + */ + dmRootPolicy?: "self" | "either-participant"; } /** @@ -170,6 +174,9 @@ export async function verifyRoomNotice( expectedBearer: claims.poster, // Obligation 3 (scope.kind MUST be "room" and scope.path MUST equal this notice's own `room` field): this notice's internal self-consistency, independent of options.expectedRoom above, which instead asserts the room against what the CALLER expected. roomPath: claims.room, + ...(options.dmRootPolicy !== undefined + ? { dmRootPolicy: options.dmRootPolicy } + : {}), }); if (!tokenVerdict.ok) { return { ok: false, reason: tokenVerdict.reason }; diff --git a/ts/packages/core/test/mesh-session-create.unit.test.ts b/ts/packages/core/test/mesh-session-create.unit.test.ts index 724f1cff..475d7bae 100644 --- a/ts/packages/core/test/mesh-session-create.unit.test.ts +++ b/ts/packages/core/test/mesh-session-create.unit.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from "vitest"; import type { DataHaveFrame, + DataRequestFrame, GossipFrame, HandshakeFrame, PeerAdvert, @@ -642,4 +643,31 @@ describe("createMeshSession", () => { expect(result.done).toBe(false); await session.close(); }); + + it("hands every received frame to onFrame", async () => { + const { transport, connection } = fakeTransport(); + const onFrame = vi.fn(async (): Promise => Promise.resolve()); + const session = createMeshSession( + transport, + testIdentity, + testClock, + null, + [], + undefined, + onFrame, + ); + await session.connect("ws://node", ["core/data"]); + + const request: DataRequestFrame = { + type: "data-request", + peer: deviceA, + "from-seq": 0, + }; + connection.push(request); + + await vi.waitFor(() => { + expect(onFrame).toHaveBeenCalledWith(expect.anything(), request); + }); + await session.close(); + }); }); diff --git a/ts/packages/core/test/notice-board.unit.test.ts b/ts/packages/core/test/notice-board.unit.test.ts index a772f012..c17ed9de 100644 --- a/ts/packages/core/test/notice-board.unit.test.ts +++ b/ts/packages/core/test/notice-board.unit.test.ts @@ -377,3 +377,93 @@ describe("createNoticeBoard", () => { expect(encryptedContentType("text/plain")).toBe("text/plain+aes256gcm"); }); }); + +describe("notices in a DM", () => { + async function dm(): Promise<{ + a: IdentityPort; + b: IdentityPort; + roomPath: string; + }> { + const first = await generateEs256Identity(); + const second = await generateEs256Identity(); + const firstIsLower = + deviceIdToHex(first.deviceId) < deviceIdToHex(second.deviceId); + const a = firstIsLower ? first : second; + const b = firstIsLower ? second : first; + return { + a, + b, + roomPath: `${deviceIdToHex(a.deviceId)}+${deviceIdToHex(b.deviceId)}`, + }; + } + + async function boardFor( + identity: IdentityPort, + roomPath: string, + key: Uint8Array, + ): Promise<{ + board: ReturnType; + storage: KeyValueStorage; + }> { + const keys = memoryKeyStore(); + await keys.set(roomPath, FIRST_EPOCH, key); + const storage = createMemoryStorage(); + return { + board: createNoticeBoard({ + identity, + storage, + clock: fixedClock(NOW_MS), + revocation: createRevocationView(), + roomKeys: keys, + }), + storage, + }; + } + + it("reads its own notice although the token it holds is rooted at the other participant", async () => { + const { a, b, roomPath } = await dm(); + const { board } = await boardFor(a, roomPath, generateContentKey()); + await board.postEncryptedNotice({ + room: roomPath, + token: await mintRoomMemberToken(b, a, roomPath), + contentType: "text/plain", + plaintext: new TextEncoder().encode("mine"), + }); + + const read = await board.readOwnNotices(roomPath); + + expect(read[0]?.verified).toBe(true); + expect(read[0]?.plaintext).toEqual(new TextEncoder().encode("mine")); + }); + + it("reads the peer's notice when its token was granted by the reader, and refuses one the poster issued itself", async () => { + const { a, b, roomPath } = await dm(); + const key = generateContentKey(); + const granted = await boardFor(a, roomPath, key); + const selfIssued = await boardFor(a, roomPath, key); + await granted.board.postEncryptedNotice({ + room: roomPath, + token: await mintRoomMemberToken(b, a, roomPath), + contentType: "text/plain", + plaintext: new TextEncoder().encode("granted"), + }); + await selfIssued.board.postEncryptedNotice({ + room: roomPath, + token: await mintRoomMemberToken(a, a, roomPath), + contentType: "text/plain", + plaintext: new TextEncoder().encode("self-issued"), + }); + + for (const [poster, expected] of [ + [granted, true], + [selfIssued, false], + ] as const) { + const reader = await boardFor(b, roomPath, key); + for (const entry of await readEntries(poster.storage, a.deviceId, 0)) { + await reader.board.ingestPeerEntry(a.deviceId, new Uint8Array(entry)); + } + const read = await reader.board.readPeerNotices(a.deviceId, roomPath); + expect(read[0]?.verified).toBe(expected); + } + }); +}); diff --git a/ts/packages/web-console/README.md b/ts/packages/web-console/README.md index ddde530e..e6b8ebb2 100644 --- a/ts/packages/web-console/README.md +++ b/ts/packages/web-console/README.md @@ -43,7 +43,7 @@ The `webrtc:signal` capability itself carries no target-device field -- a `manag ## Messaging -Messages travel over `core/room`. A conversation is identified by its room path: a direct message's path names the two participating devices, so one peer is one conversation. `src/conversations.ts` holds the state of every conversation and the pure reducer that evolves it, and `src/hooks/use-room-messaging.ts` feeds it from live sessions. A conversation reaches its peer by up to two routes. Clicking Message opens it at once over the hub the peer was found through (a relay pairing and a secure channel, which needs only the connection both sides already have), and a WebRTC connection is negotiated in the background; when it opens, the conversation moves onto it, and when it closes the conversation falls back to the hub. `src/request-demux.ts` splits a hub connection's incoming requests between the negotiator and the conversations reached through that hub. A conversation restored from `src/message-store.ts` at start-up has no route, is shown as offline, and takes one when a peer reconnects. Durable notices sync with data frames, which only a direct connection carries, so they are unavailable while a conversation is on the hub. Received messages in a conversation that is not on screen count as unread; the conversation on screen is read as messages arrive. Both sent and received messages are persisted per room path. A message being sent shows in the conversation as sending; if delivery fails it stays there as not sent, with the reason, and can be retried or dismissed. +Messages travel over `core/room`. A conversation is identified by its room path: a direct message's path names the two participating devices, so one peer is one conversation. `src/conversations.ts` holds the state of every conversation and the pure reducer that evolves it, and `src/hooks/use-room-messaging.ts` feeds it from live sessions. A conversation reaches its peer by up to two routes. Clicking Message opens it at once over the hub the peer was found through (a relay pairing and a secure channel, which needs only the connection both sides already have), and a WebRTC connection is negotiated in the background; when it opens, the conversation moves onto it, and when it closes the conversation falls back to the hub. `src/request-demux.ts` splits a hub connection's incoming requests between the negotiator and the conversations reached through that hub. A conversation restored from `src/message-store.ts` at start-up has no route, is shown as offline, and takes one when a peer reconnects. Durable notices are kept in the console's storage together with the conversation's room keys and tokens, so a reload keeps them, and they sync with data frames over whichever route is connected: a peer announces its log directly, and a hub that holds logs for absent devices (`--mailbox-dir`) is sent the own log when the connection opens and after each post and is asked for the conversation peers' logs, so a notice posted while the peer was offline is read after the peer reconnects. A hub does not announce a new notice to a peer that is already connected, so over a hub alone a notice appears at the peer's next sync, which a message from the author, a reconnect or opening the conversation triggers. Received messages in a conversation that is not on screen count as unread; the conversation on screen is read as messages arrive. Both sent and received messages are persisted per room path. A message being sent shows in the conversation as sending; if delivery fails it stays there as not sent, with the reason, and can be retried or dismissed. ## What is deliberately deferred diff --git a/ts/packages/web-console/playwright.config.ts b/ts/packages/web-console/playwright.config.ts index 0909cb51..b237ff1b 100644 --- a/ts/packages/web-console/playwright.config.ts +++ b/ts/packages/web-console/playwright.config.ts @@ -3,6 +3,10 @@ import { defineConfig } from "@playwright/test"; // A fixed, dedicated port pair for the e2e run (never the dev-time 8787/5173 pair) so a developer's own `pnpm dev` session never collides with CI or a concurrent local test run. export const RELAY_PORT = 8790; export const VITE_PORT = 8798; +// A second hub that also holds other devices' logs while they are offline, for the spec that has one device away while the other writes. +export const MAILBOX_RELAY_PORT = 8791; +export const MAILBOX_RELAY_ADDRESS = `ws://127.0.0.1:${String(MAILBOX_RELAY_PORT)}`; +const MAILBOX_DIR = "test-results/mailbox"; export const RELAY_ADDRESS = `ws://127.0.0.1:${String(RELAY_PORT)}`; const RELAY_HEALTH_URL = `http://127.0.0.1:${String(RELAY_PORT)}/`; const HARNESS_URL = `http://localhost:${String(VITE_PORT)}/live-check/harness.html`; @@ -26,6 +30,12 @@ export default defineConfig({ reuseExistingServer: process.env.CI === undefined, timeout: 10_000, }, + { + command: `node ${nodePackageDist} --bind 127.0.0.1:${String(MAILBOX_RELAY_PORT)} --mailbox-dir ${MAILBOX_DIR}`, + port: MAILBOX_RELAY_PORT, + reuseExistingServer: process.env.CI === undefined, + timeout: 10_000, + }, { command: `npx vite --port ${String(VITE_PORT)} --strictPort`, url: HARNESS_URL, diff --git a/ts/packages/web-console/src/App.tsx b/ts/packages/web-console/src/App.tsx index 64b27f88..138e0d82 100644 --- a/ts/packages/web-console/src/App.tsx +++ b/ts/packages/web-console/src/App.tsx @@ -57,7 +57,10 @@ import type { PreferencesStore } from "./preferences-store.js"; import type { NameStore } from "./name-store.js"; import { ConversationList } from "./components/ConversationList.js"; import { RoomPanel } from "./components/RoomPanel.js"; -import { useRoomMessaging } from "./hooks/use-room-messaging.js"; +import { + useRoomMessaging, + type HubEndpoint, +} from "./hooks/use-room-messaging.js"; import { createRequestDemux } from "./request-demux.js"; import { ROOM_MEMBER_CAPABILITY } from "wire-mesh-core/domain/room-token-verification"; import { WEBRTC_SIGNAL_VERB } from "wire-mesh-core/domain/webrtc-signaling"; @@ -104,6 +107,8 @@ interface ConnectionFailure { interface ConnectionEntry { address: string; session: ReturnType; + /** The session as the conversations reach it: one stable object per connection, so a conversation's route can be matched when the connection closes. */ + hub: HubEndpoint; negotiator: WebrtcNegotiator; } @@ -168,6 +173,11 @@ export function App({ }, [selectedRoomPath, selectedUnread, markRead]); // A negotiator's own onIncomingConnection callback is registered once, at construction, and must still call whatever the *latest* attach is -- attach itself is a fresh function every time useRoomMessaging's own session map changes, so a ref (updated every render, read from the callback) is what keeps that call from closing over a stale, since-superseded attach. + const hubsBySession = useRef(new Map()); + const onHubFrameRef = useRef(roomMessaging.onHubFrame); + useEffect(() => { + onHubFrameRef.current = roomMessaging.onHubFrame; + }, [roomMessaging.onHubFrame]); const watchHubRef = useRef(roomMessaging.watchHub); useEffect(() => { watchHubRef.current = roomMessaging.watchHub; @@ -238,13 +248,16 @@ export function App({ } } })(); - watchHubRef.current({ + const hub: HubEndpoint = { sendManageRequest: async (...args) => session.sendManageRequest(...args), + sendDataFrame: async (frame) => session.sendDataFrame(frame), incomingManageRequests: demux.stream(ROOM_MEMBER_CAPABILITY), - }); + }; + hubsBySession.current.set(session, hub); + watchHubRef.current(hub); setConnections((current) => [ ...current, - { address: entryAddress, session, negotiator }, + { address: entryAddress, session, hub, negotiator }, ]); } @@ -287,6 +300,7 @@ export function App({ const expansionRef: { current: GossipExpansion | null } = { current: null, }; + const sessionRef: { current: MeshSession | null } = { current: null }; const session = createMeshSession( createDialTransport(trust.memory), identity, @@ -296,7 +310,18 @@ export function App({ (advert) => { expansionRef.current?.considerAdvert(advert); }, + (connection, frame) => { + // Frames that arrive before the connection is attached have no hub to go to yet; the notice logs are exchanged on the first frame after it is. + const hub = + sessionRef.current === null + ? undefined + : hubsBySession.current.get(sessionRef.current); + if (hub !== undefined) { + onHubFrameRef.current(hub, connection, frame); + } + }, ); + sessionRef.current = session; expansionRef.current = createGossipExpansion({ selfDeviceId: identity.deviceId, shouldExpand: async (candidate) => confirmExpansion(candidate, via), @@ -425,7 +450,7 @@ export function App({ } function handleClose(target: Readonly): void { - roomMessaging.dropHub(target.session); + roomMessaging.dropHub(target.hub); void target.session.close(); setConnections((current) => current.filter((entry) => entry.session !== target.session), @@ -437,7 +462,7 @@ export function App({ device: DeviceId, ): void { // The conversation opens at once over the hub the peer was found through, and a direct connection is negotiated in the background: it takes over when it opens, and its failure costs nothing but the direct route. - roomMessaging.openRelay(entry.session, device); + roomMessaging.openRelay(entry.hub, device); setSelectedPath( dmRoomPath(deviceIdToHex(identity.deviceId), deviceIdToHex(device)), ); diff --git a/ts/packages/web-console/src/components/RoomPanel.tsx b/ts/packages/web-console/src/components/RoomPanel.tsx index 7a98b86f..7a85f562 100644 --- a/ts/packages/web-console/src/components/RoomPanel.tsx +++ b/ts/packages/web-console/src/components/RoomPanel.tsx @@ -107,7 +107,7 @@ export function RoomPanel({ labelPosition="center" /> - {view.via === "direct" ? ( + {view.via !== undefined ? ( ) : ( - Durable notices need a direct connection. + Durable notices can be posted while the conversation is connected. )} diff --git a/ts/packages/web-console/src/conversations.ts b/ts/packages/web-console/src/conversations.ts index e8fcce01..8c69b5d8 100644 --- a/ts/packages/web-console/src/conversations.ts +++ b/ts/packages/web-console/src/conversations.ts @@ -31,8 +31,11 @@ export type PendingOutgoing = { | { status: "failed"; error: string } ); -/** What a conversation needs from a hub connection to send through it: the relay peer is addressed by device-id on each request. */ -export type RelaySender = Pick; +/** What a conversation needs from a hub connection to send through it: the relay peer is addressed by device-id on each request, and notice logs are exchanged with the hub as data frames. */ +export type RelaySender = Pick< + MeshSession, + "sendManageRequest" | "sendDataFrame" +>; /** A conversation's route through a hub, where the peer is reached by a relay pairing and a secure channel instead of a connection of its own. */ export interface RelayRoute { diff --git a/ts/packages/web-console/src/hooks/use-room-messaging.ts b/ts/packages/web-console/src/hooks/use-room-messaging.ts index a1aaba00..b1543095 100644 --- a/ts/packages/web-console/src/hooks/use-room-messaging.ts +++ b/ts/packages/web-console/src/hooks/use-room-messaging.ts @@ -23,7 +23,10 @@ import type { DeviceId, Frame, } from "wire-mesh-core/generated/protocol"; -import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; +import { + deviceIdFromHex, + deviceIdToHex, +} from "wire-mesh-core/domain/device-id"; import { dmRoomPath } from "wire-mesh-core/domain/room-path"; import { createRoomRouter, @@ -42,10 +45,10 @@ import { } from "../conversations.js"; import type { KeyValueStorage } from "wire-mesh-core/ports/storage"; import { + createPersistentRoomKeyDelivery, createPersistentRoomKeyStore, createPersistentRoomTokenStore, } from "../persistent-room-state.js"; -import type { RoomKeyStore } from "wire-mesh-core/domain/notice-board"; import { createRoomRekeyHandler } from "wire-mesh-core/domain/room-rekey"; import { bootstrapDmEpoch1, createNoticeWiring } from "../notices.js"; @@ -73,6 +76,12 @@ export interface RoomMessaging { watchHub: (hub: Readonly) => void; /** Forgets a hub as a route, as when its connection is closed. */ dropHub: (hub: Readonly) => void; + /** Feeds one frame received on a hub connection. The first frame on a connection, which is also the first after a reconnect, announces this side's notice log to the hub and asks it for the logs of every peer this console talks to. */ + onHubFrame: ( + hub: Readonly, + connection: Readonly, + frame: Frame, + ) => void; /** Wires a WebRTC-negotiated Connection up as the direct route of the peer's DM conversation -- either side: this console's own initiate() (knownPeerDevice supplied), or an incoming offer accepted via onIncomingConnection (peer device discovered from AcceptedMeshSession's own peerDeviceId). A no-op if that conversation already has a live session; a conversation restored from storage, or whose session has closed, takes the new one. */ attach: ( connection: Readonly, @@ -84,7 +93,7 @@ export interface RoomMessaging { retry: (roomPath: string, localId: string) => Promise; /** Dismisses an outgoing message without sending it. */ discard: (roomPath: string, localId: string) => void; - /** Posts one durable encrypted notice to the DM room, over the direct connection only (the notice board syncs with data frames, which a hub route does not carry): joins if no token yet, bootstraps epoch 1 when this side is the lower participant and no epoch exists, then posts and announces via the notice wiring. */ + /** Posts one durable encrypted notice to the DM room, over whichever route the conversation has: joins if no token yet, bootstraps epoch 1 when this side is the lower participant and no epoch exists, then posts. The notice is announced to every connected route, and a hub that holds it delivers it to a peer that was offline. */ postNotice: (roomPath: string, text: string) => Promise; /** Acknowledges a conversation's unread messages. */ markRead: (roomPath: string) => void; @@ -129,39 +138,75 @@ export function useRoomMessaging( }); }, []); + /** The latest conversations and hubs, for callbacks that outlive the render that created them. */ + const sessionsRef = useRef(sessions); useEffect(() => { - const unmounted = new AbortController(); - void (async (): Promise => { - for (const roomPath of await messageStore.roomPaths()) { - const messages = await messageStore.list(roomPath); - if (unmounted.signal.aborted) return; - dispatch({ - type: "restored", - roomPath, - participants: participantsOf(roomPath, ownDeviceHex), - messages, - }); + sessionsRef.current = sessions; + }, [sessions]); + const hubsRef = useRef(new Set>()); + const seenConnectionsRef = useRef(new WeakSet()); + const roomKeyStore = useMemo( + () => createPersistentRoomKeyStore(roomStorage), + [roomStorage], + ); + + /** The other members of every conversation this console holds. */ + function knownPeers(): DeviceId[] { + return [...sessionsRef.current.values()].flatMap((entry) => + entry.participants.map((hex) => deviceIdFromHex(hex)), + ); + } + + const refreshAllRef = useRef<() => void>(() => undefined); + const noticeWiring = useMemo( + () => + createNoticeWiring({ + senders: () => [ + ...new Set([ + ...[...sessionsRef.current.values()].flatMap((entry) => [ + ...(entry.direct === undefined ? [] : [entry.direct]), + ...(entry.relay === undefined ? [] : [entry.relay.hub]), + ]), + ...hubsRef.current, + ]), + ], + storage: roomStorage, + identity, + clock, + revocation, + roomKeys: roomKeyStore, + onChange: () => { + refreshAllRef.current(); + }, + }), + [identity, clock, roomStorage, roomKeyStore, revocation], + ); + + /** Re-reads one conversation's notices into its state. */ + function refreshRoom(roomPath: string): void { + const participants = sessionsRef.current.get(roomPath)?.participants; + if (participants === undefined) return; + void noticeWiring + .readRoom( + roomPath, + participants.map((hex) => deviceIdFromHex(hex)), + ) + .then((notices) => { + dispatch({ type: "notices", roomPath, notices: [...notices] }); + }); + } + useEffect(() => { + refreshAllRef.current = (): void => { + for (const roomPath of sessionsRef.current.keys()) { + refreshRoom(roomPath); } - })(); - return () => { - unmounted.abort(); }; - }, [messageStore, ownDeviceHex]); - /** Per-session notice machinery, outside the reducer: mutable wiring state the frame flow drives directly, with reducer-visible changes surfaced through "notices" actions. */ - const noticeState = useRef( - new Map< - string, - { - wiring: ReturnType; - roomKeys: RoomKeyStore; - refresh: () => void; - } - >(), - ); + }); + /** The latest held token per session, for the lazily-built rekey handler -- a ref, not reducer state, because the handler closes over it at dispatch time and must see the current value. */ const tokenRef = useRef(new Map()); - const roomKeyStore = useMemo( - () => createPersistentRoomKeyStore(roomStorage), + const keyDelivery = useMemo( + () => createPersistentRoomKeyDelivery(roomStorage), [roomStorage], ); const tokenStore = useMemo( @@ -188,51 +233,64 @@ export function useRoomMessaging( return restored; } - function makeNoticeLifecycle( - session: MeshSession, - roomPath: string, - ): { - wiring: ReturnType; - roomKeys: RoomKeyStore; - refresh: () => void; - } { - // The wiring's onChange needs refresh; refresh needs the wiring. An - // object holder breaks the cycle without a let-reassignment: refresh - // reads lifecycle.wiring, which is populated immediately after - // construction and can only be called from the wiring's own events (or - // externally, after construction) -- never before it exists. - const lifecycle: { - wiring?: ReturnType; - } = {}; - const refresh = (): void => { - const wiring = lifecycle.wiring; - if (wiring === undefined) { - return; + useEffect(() => { + const unmounted = new AbortController(); + void (async (): Promise => { + const roomPaths = new Set([ + ...(await messageStore.roomPaths()), + ...(await tokenStore.rooms()), + ]); + for (const roomPath of roomPaths) { + const messages = await messageStore.list(roomPath); + if (unmounted.signal.aborted) return; + const participants = participantsOf(roomPath, ownDeviceHex); + dispatch({ type: "restored", roomPath, participants, messages }); + // The reducer has not applied the action yet, so the notices are read for the participants directly. + void noticeWiring + .readRoom( + roomPath, + participants.map((hex) => deviceIdFromHex(hex)), + ) + .then((notices) => { + dispatch({ type: "notices", roomPath, notices: [...notices] }); + }); } - void wiring.readRoom(roomPath).then((notices) => { - dispatch({ type: "notices", roomPath, notices: [...notices] }); - }); + })(); + return () => { + unmounted.abort(); }; - const wiring = createNoticeWiring({ - session, - storage: roomStorage, + }, [messageStore, tokenStore, noticeWiring, ownDeviceHex]); + + /** The first-epoch deliveries under way, by room: a second call for a room joins the first instead of starting its own, because two that both found no key would each mint one and leave the two sides holding different keys. */ + const firstEpochInFlight = useRef(new Map>()); + + /** Hands the peer the DM's first epoch key if this side is the one that mints it and the peer has not yet acknowledged it. Rejects when the peer cannot take it yet. */ + async function deliverFirstEpoch( + roomPath: string, + token: CapabilityToken, + route: { session: RelaySender; target: DeviceId | undefined }, + ): Promise { + const running = firstEpochInFlight.current.get(roomPath); + if (running !== undefined) return running; + const delivery = bootstrapDmEpoch1({ + session: route.session, + target: route.target, identity, clock, revocation, + ownRoomMemberToken: token, + roomPath, roomKeys: roomKeyStore, - onChange: () => { - refresh(); - }, + delivery: keyDelivery, + }).finally(() => { + firstEpochInFlight.current.delete(roomPath); }); - lifecycle.wiring = wiring; - return { wiring, roomKeys: roomKeyStore, refresh }; + firstEpochInFlight.current.set(roomPath, delivery); + return delivery; } - /** What to do with each room request that arrives for a conversation, whichever route it came over. `notice` is the conversation's notice board, which only a direct connection has: a rekey that arrives without one is refused, since there is nowhere to keep the key. */ - function roomHandlers( - roomPath: string, - notice: ReturnType | undefined, - ): RoomRouterHandlers { + /** What to do with each room request that arrives for a conversation, whichever route it came over. */ + function roomHandlers(roomPath: string): RoomRouterHandlers { return { onMessage: (message) => { const stored: StoredMessage = { @@ -243,6 +301,12 @@ export function useRoomMessaging( }; persist(messageStore.append(roomPath, stored)); dispatch({ type: "message", roomPath, message: stored }); + // A peer that is writing may have written notices too, and a hub does not announce them to the other side. + void noticeWiring.pull( + (sessionsRef.current.get(roomPath)?.participants ?? []).map((hex) => + deviceIdFromHex(hex), + ), + ); }, onGrantIssued: (token) => { persist(grants.record("issued", token, clock.now())); @@ -256,6 +320,13 @@ export function useRoomMessaging( async decide(decision): Promise { dispatch({ type: "join-request-settled", roomPath }); await event.decide(decision); + // Once the peer holds its token it can take the first epoch key, which a rekey sent before then could not. + void (async (): Promise => { + const token = await heldToken(roomPath); + const entry = sessionsRef.current.get(roomPath); + if (token === undefined || entry === undefined) return; + await deliverFirstEpoch(roomPath, token, routeOf(entry)); + })().catch(() => undefined); }, }, }); @@ -267,13 +338,6 @@ export function useRoomMessaging( // the peer retries, which the DM bootstrap's lower-mints ordering // makes the steady state anyway. onRekey: async (incoming) => { - if (notice === undefined) { - await incoming.respond({ - result: "error", - code: "unsupported_route", - }); - return; - } const ownToken = await heldToken(roomPath); if (ownToken === undefined) { await incoming.respond({ @@ -289,13 +353,13 @@ export function useRoomMessaging( ownRoomMemberToken: ownToken, onRekey: async (event) => { for (const [i, key] of event.contentKeys.entries()) { - await notice.roomKeys.set( + await roomKeyStore.set( roomPath, event.keyEpoch - event.contentKeys.length + 1 + i, key, ); } - notice.refresh(); + refreshRoom(roomPath); }, }); await handler(incoming); @@ -308,11 +372,11 @@ export function useRoomMessaging( connection: Readonly, knownPeerDevice?: DeviceId, ): Promise => { - // The wiring needs the session (sendDataFrame) and the session's own - // onFrame hook needs the wiring -- but the peer's device-id (and with - // it the DM room path) only resolves after the handshake. A late-bound - // hook bridges the gap: early frames (handshake/gossip) find nothing - // to observe; once the lifecycle exists every frame reaches it. + // The peer's device-id (and with it the DM room path) only resolves + // after the handshake, but the session's onFrame hook exists from the + // start; a late-bound sink bridges the gap: early frames (handshake, + // gossip) find nothing to observe, and once the sink is set every frame + // reaches the notice wiring. const frameSink: { current?: (frame: Frame) => void } = {}; const accepted = await acceptMeshSession( connection, @@ -333,10 +397,8 @@ export function useRoomMessaging( await accepted.close(); return; } - const notice = makeNoticeLifecycle(accepted, roomPath); - noticeState.current.set(roomPath, notice); frameSink.current = (frame) => { - notice.wiring.handleFrame(frame); + noticeWiring.handleFrame(frame, accepted); }; dispatch({ type: "direct-opened", @@ -350,6 +412,7 @@ export function useRoomMessaging( participants: [peerHex], messages: await messageStore.list(roomPath), }); + void noticeWiring.sync(accepted, [peerDevice]).catch(() => undefined); createRoomRouter( accepted, @@ -360,7 +423,7 @@ export function useRoomMessaging( peerDevice, currentMembers: () => [identity.deviceId, peerDevice], }, - roomHandlers(roomPath, notice), + roomHandlers(roomPath), ); void (async (): Promise => { @@ -371,7 +434,7 @@ export function useRoomMessaging( } })(); }, - [identity, clock, ownDeviceHex, sessions, messageStore], + [identity, clock, ownDeviceHex, sessions, messageStore, noticeWiring], ); /** One delivery attempt: joins the room if this side holds no token yet, sends, and persists the message. Reports each phase it is in through onPhase, since the join waits on the other side's consent. Throws on any failure; the caller decides how that is surfaced. */ @@ -394,20 +457,10 @@ export function useRoomMessaging( dispatch({ type: "token", roomPath, token }); // Opportunistic DM bootstrap: once this side holds its join-grant, // the lower participant mints epoch 1 for the noticeboard. Fire-and- - // observe rather than awaited -- messaging must not block on it. Only - // a direct connection has a notice board to bootstrap. - const notice = noticeState.current.get(roomPath); - if (entry?.direct !== undefined && notice !== undefined) { - void bootstrapDmEpoch1({ - session: entry.direct, - identity, - clock, - revocation: revocation, - ownRoomMemberToken: token, - roomPath, - roomKeys: notice.roomKeys, - }).catch(() => undefined); - } + // observe rather than awaited -- messaging must not block on it. + void deliverFirstEpoch(roomPath, token, { session, target }).catch( + () => undefined, + ); } const outcome = await sendRoomMessage( session, @@ -428,7 +481,7 @@ export function useRoomMessaging( await messageStore.append(roomPath, stored); return stored; }, - [sessions, clock, messageStore, identity], + [sessions, clock, messageStore, identity, roomKeyStore], ); /** Runs one delivery attempt for a pending outgoing message. Success moves it into the history; failure keeps it in the list as failed, with the reason, so the user can retry or dismiss it instead of losing the text. */ @@ -489,14 +542,10 @@ export function useRoomMessaging( const postNotice = useCallback( async (roomPath: string, text: string): Promise => { const entry = sessions.get(roomPath); - const session = entry?.direct; - const notice = noticeState.current.get(roomPath); - if (session === undefined || notice === undefined) { - throw new Error("durable notices need a direct connection"); - } + const { session, target } = routeOf(entry); let token = entry?.token ?? (await heldToken(roomPath)); if (token === undefined) { - const joined = await requestToJoin(session, roomPath); + const joined = await requestToJoin(session, roomPath, target); token = joined.token; await rememberToken(roomPath, token); persist(grants.record("held", token, clock.now())); @@ -506,24 +555,16 @@ export function useRoomMessaging( // without a key would throw inside the wiring -- the bootstrap must // complete first, and for the lower participant it is exactly one // wrap + one send. - await bootstrapDmEpoch1({ - session, - identity, - clock, - revocation: revocation, - ownRoomMemberToken: token, - roomPath, - roomKeys: notice.roomKeys, - }); - await notice.wiring.post({ + await deliverFirstEpoch(roomPath, token, { session, target }); + await noticeWiring.post({ room: roomPath, token, contentType: "text/plain", plaintext: new TextEncoder().encode(text), }); - notice.refresh(); + refreshRoom(roomPath); }, - [sessions, identity, clock], + [sessions, identity, clock, roomKeyStore, noticeWiring], ); const markRead = useCallback((roomPath: string): void => { @@ -543,14 +584,16 @@ export function useRoomMessaging( void messageStore.list(roomPath).then((messages) => { dispatch({ type: "restored", roomPath, participants, messages }); }); + void noticeWiring.sync(hub, [peer]).catch(() => undefined); }, - [ownDeviceHex, messageStore], + [ownDeviceHex, messageStore, noticeWiring], ); const watchHub = useCallback( (hub: Readonly): void => { // One reader for the hub's room requests, splitting them by the device each channel authenticated: a conversation's router reads only its own peer's, and the first request from a peer opens that peer's conversation. const perPeer = new Map>(); + hubsRef.current.add(hub); void (async (): Promise => { for await (const incoming of hub.incomingManageRequests) { const peer = incoming.fromDevice; @@ -571,7 +614,7 @@ export function useRoomMessaging( peerDevice: peer, currentMembers: () => [identity.deviceId, peer], }, - roomHandlers(roomPath, undefined), + roomHandlers(roomPath), ); } queue.push(incoming); @@ -582,15 +625,32 @@ export function useRoomMessaging( ); const dropHub = useCallback((hub: Readonly): void => { + hubsRef.current.delete(hub); dispatch({ type: "hub-closed", hub }); }, []); + const onHubFrame = useCallback( + ( + hub: Readonly, + connection: Readonly, + frame: Frame, + ): void => { + if (!seenConnectionsRef.current.has(connection)) { + seenConnectionsRef.current.add(connection); + void noticeWiring.sync(hub, knownPeers()).catch(() => undefined); + } + noticeWiring.handleFrame(frame, hub); + }, + [noticeWiring], + ); + return { conversations: [...sessions.values()], persistenceFailure, openRelay, watchHub, dropHub, + onHubFrame, attach, send, retry, diff --git a/ts/packages/web-console/src/notices.ts b/ts/packages/web-console/src/notices.ts index b5580c68..e6eee98e 100644 --- a/ts/packages/web-console/src/notices.ts +++ b/ts/packages/web-console/src/notices.ts @@ -32,24 +32,38 @@ import type { KeyValueStorage } from "wire-mesh-core/ports/storage"; /** How many entries one catch-up data-entries frame may carry -- a bounding constant, not a protocol limit; a larger log simply takes more rounds. */ const CATCH_UP_BATCH_LIMIT = 32; +/** Where a data frame can be sent: a direct peer session or a hub. */ +export type DataFrameSender = Pick; + export interface NoticeWiring { - /** Observes one inbound frame (pass from the session's own onFrame hook). Data-have, data-entries, and data-request are consumed here; everything else is ignored. */ - handleFrame: (frame: Frame) => void; - /** Posts an encrypted notice and immediately announces it with a data-have on the wire. */ + /** Observes one inbound frame (pass from a session's own onFrame hook) together with the sender it arrived through, which any reply goes back to. Data-have, data-entries, and data-request are consumed here; everything else is ignored. */ + handleFrame: (frame: Frame, from: Readonly) => void; + /** Posts an encrypted notice and announces it, with a data-have, to every sender connected at that moment. A sender that is not connected is skipped: the notice is in the own log, and `sync` announces it when that sender connects. */ post: (options: { room: string; token: CapabilityToken; contentType: string; plaintext: Uint8Array; }) => Promise; - /** Verified (and where this side holds the epoch key, decrypted) notices this side holds for one room, across every tracked peer log plus the own log. */ - readRoom: (room: string) => Promise; + /** Verified (and where this side holds the epoch key, decrypted) notices this side holds for one room: the own log plus the logs of `peers`, the room's other members. */ + readRoom: ( + room: string, + peers: readonly DeviceId[], + ) => Promise; + /** Announces the own log to `sender` and asks it for the logs of `peers`, as when a connection opens: whatever was posted or written while the two were apart is exchanged. Rejects when the sender is not connected. */ + sync: ( + sender: Readonly, + peers: readonly DeviceId[], + ) => Promise; + /** Asks every connected sender for the logs of `peers`, as when something suggests a peer has written: a message from it arrived. */ + pull: (peers: readonly DeviceId[]) => Promise; /** The key store createRoomRekeyHandler's onRekey feeds -- exposed so the rekey wiring and this module share one store. */ roomKeys: RoomKeyStore; } export interface CreateNoticeWiringOptions { - session: MeshSession; + /** Every sender that is connected right now; consulted on each post. */ + senders: () => readonly DataFrameSender[]; storage: KeyValueStorage; identity: IdentityPort; clock: Clock; @@ -62,7 +76,7 @@ export interface CreateNoticeWiringOptions { export function createNoticeWiring( options: Readonly, ): NoticeWiring { - const { session, storage, identity, clock, revocation, roomKeys, onChange } = + const { senders, storage, identity, clock, revocation, roomKeys, onChange } = options; const board = createNoticeBoard({ identity, @@ -71,22 +85,32 @@ export function createNoticeWiring( revocation, roomKeys, }); - /** Logs this side tracks: hex device-id key (bytewise device-ids are reference-compared by Map, so equal bytes from separately decoded frames would duplicate), value the peer's device id. Seeded once the peer's first data-have arrives. */ - const trackedPeers = new Map(); - async function requestCatchUp(peer: DeviceId): Promise { + async function requestCatchUp( + from: Readonly, + peer: DeviceId, + ): Promise { const head = await headSeqFor(storage, peer); - await session.sendDataFrame({ + await from.sendDataFrame({ type: "data-request", peer, "from-seq": head, }); } - function handleFrame(frame: Frame): void { + async function announceOwnLog(to: Readonly): Promise { + const head = await headSeqFor(storage, identity.deviceId); + if (head === 0) return; + await to.sendDataFrame({ + type: "data-have", + peer: identity.deviceId, + "head-seq": head, + }); + } + + function handleFrame(frame: Frame, from: Readonly): void { if (frame.type === "data-have") { - trackedPeers.set(deviceIdToHex(frame.peer), frame.peer); - void requestCatchUp(frame.peer).catch(() => undefined); + void requestCatchUp(from, frame.peer).catch(() => undefined); return; } if (frame.type === "data-entries") { @@ -95,7 +119,7 @@ export function createNoticeWiring( if (!result.ok) return; // A full batch means the sender may have had more; a short batch means its log ended there. One more round per full batch keeps catch-up bounded without a second bookkeeping mechanism. if (frame.entries.length >= CATCH_UP_BATCH_LIMIT) { - await requestCatchUp(frame.peer).catch(() => undefined); + await requestCatchUp(from, frame.peer).catch(() => undefined); } onChange(); })(); @@ -109,7 +133,7 @@ export function createNoticeWiring( CATCH_UP_BATCH_LIMIT, ); if (entries !== null) { - await session.sendDataFrame(entries); + await from.sendDataFrame(entries); } })().catch(() => undefined); } @@ -119,24 +143,45 @@ export function createNoticeWiring( handleFrame, async post(postOptions) { const { dataHave } = await board.postEncryptedNotice(postOptions); - await session.sendDataFrame(dataHave); + await Promise.allSettled( + senders().map(async (sender) => sender.sendDataFrame(dataHave)), + ); onChange(); }, - async readRoom(room) { + async readRoom(room, peers) { const own = await board.readOwnNotices(room); const perPeer = await Promise.all( - [...trackedPeers.values()].map(async (peer) => - board.readPeerNotices(peer, room), - ), + peers.map(async (peer) => board.readPeerNotices(peer, room)), ); return [...own, ...perPeer.flat()]; }, + async sync(sender, peers) { + await announceOwnLog(sender); + await Promise.all( + peers.map(async (peer) => requestCatchUp(sender, peer)), + ); + }, + async pull(peers) { + await Promise.allSettled( + senders().flatMap((sender) => + peers.map(async (peer) => requestCatchUp(sender, peer)), + ), + ); + }, roomKeys, }; } +/** Whether the peer has been given a room's first epoch key, kept so a rekey that failed (the peer offline, or not yet holding its own token) is sent again by the next call instead of being forgotten. */ +export interface RoomKeyDelivery { + isDelivered: (room: string) => Promise; + markDelivered: (room: string) => Promise; +} + export interface BootstrapDmEpoch1Options { - session: MeshSession; + /** Where the rekey is sent: the direct session to the peer, or a hub with `target` naming the peer to reach through it. */ + session: Pick; + target?: DeviceId | undefined; identity: IdentityPort; clock: Clock; revocation: RevocationCheck; @@ -144,6 +189,7 @@ export interface BootstrapDmEpoch1Options { ownRoomMemberToken: CapabilityToken; roomPath: string; roomKeys: RoomKeyStore; + delivery: RoomKeyDelivery; } /** @@ -154,24 +200,25 @@ export interface BootstrapDmEpoch1Options { * (I approved your join, so my token chains to you; your key wraps my epoch). * The higher participant never calls this: it waits to receive room.rekey, per * the deterministic lower-mints rule that mirrors dmRoomPath's own sorted-pair - * convention. A no-op when the store already holds any epoch (idempotent across - * re-sends) or when this side is the higher participant. + * convention. Idempotent once the peer has acknowledged the key; until then each + * call sends the same epoch-1 key again, and rejects when the peer cannot take + * it yet, so the caller can try again later. */ export async function bootstrapDmEpoch1( options: Readonly, ): Promise { const { session, + target, identity, clock, revocation, ownRoomMemberToken, roomPath, roomKeys, + delivery, } = options; - if ((await roomKeys.currentEpoch(roomPath)) !== undefined) { - return; - } + const FIRST_EPOCH = 1; // Only the lower participant mints epoch 1: dmRoomPath embeds the sorted // pair (lower + "+" + higher), so this side mints iff its own hex sorts // strictly below the peer's -- the participant names ARE the path halves. @@ -179,6 +226,13 @@ export async function bootstrapDmEpoch1( if (lower === "" || deviceIdToHex(identity.deviceId) !== lower) { return; } + const epoch = await roomKeys.currentEpoch(roomPath); + if (epoch !== undefined && epoch !== FIRST_EPOCH) { + return; + } + if (await delivery.isDelivered(roomPath)) { + return; + } const verdict = await verifyRoomToken(ownRoomMemberToken, { identity, clock, @@ -194,14 +248,34 @@ export async function bootstrapDmEpoch1( if (deriveSharedSecret === undefined) { throw new Error("this identity cannot derive ECDH shared secrets"); } - const FIRST_EPOCH = 1; - const contentKey = generateContentKey(); + const contentKey = + epoch === undefined + ? generateContentKey() + : await roomKeys.get(roomPath, FIRST_EPOCH); + if (contentKey === undefined) { + throw new Error( + `room ${roomPath} lost its epoch ${String(FIRST_EPOCH)} key`, + ); + } const sharedSecret = await deriveSharedSecret(verdict.rootIssuerKey); const wrappingKey = await deriveWrappingKey(sharedSecret, { room: roomPath, keyEpoch: FIRST_EPOCH, }); const wrapped = await wrapContentKey(wrappingKey, contentKey); - await roomKeys.set(roomPath, FIRST_EPOCH, contentKey); - await sendRoomRekey(session, roomPath, FIRST_EPOCH, wrapped); + if (epoch === undefined) { + await roomKeys.set(roomPath, FIRST_EPOCH, contentKey); + } + const outcome = await sendRoomRekey( + session, + roomPath, + FIRST_EPOCH, + wrapped, + target, + ownRoomMemberToken, + ); + if (outcome.result !== "ok") { + throw new Error(`room.rekey was refused: ${outcome.code}`); + } + await delivery.markDelivered(roomPath); } diff --git a/ts/packages/web-console/src/persistent-room-state.ts b/ts/packages/web-console/src/persistent-room-state.ts index 15192883..b9d55690 100644 --- a/ts/packages/web-console/src/persistent-room-state.ts +++ b/ts/packages/web-console/src/persistent-room-state.ts @@ -13,9 +13,11 @@ import { import type { Clock } from "wire-mesh-core/ports/clock"; import type { IdentityPort } from "wire-mesh-core/ports/identity"; import type { KeyValueStorage } from "wire-mesh-core/ports/storage"; +import type { RoomKeyDelivery } from "./notices.js"; const KEY_PREFIX = "room-key/"; const TOKEN_PREFIX = "room-token/"; +const DELIVERED_PREFIX = "room-key-delivered/"; /** Epoch numbers are zero-padded to this width so the storage keys of one room sort in epoch order. */ const EPOCH_KEY_WIDTH = String(Number.MAX_SAFE_INTEGER).length; @@ -24,6 +26,8 @@ export interface RoomTokenStore { /** The token held for `room` if it is still valid for this device, otherwise undefined (an invalid stored token is removed). */ get: (room: string) => Promise; set: (room: string, token: Readonly) => Promise; + /** Every room a token is stored for, valid or not: the rooms this device has joined. */ + rooms: () => Promise; } function epochKey(room: string, epoch: number): string { @@ -82,5 +86,22 @@ export function createPersistentRoomTokenStore( new Uint8Array(encode(token, cdeEncodeOptions)), ); }, + rooms: async () => + (await storage.keys(TOKEN_PREFIX)).map((key) => + key.slice(TOKEN_PREFIX.length), + ), + }; +} + +/** Which rooms the peer has acknowledged the first epoch key of. */ +export function createPersistentRoomKeyDelivery( + storage: Readonly, +): RoomKeyDelivery { + return { + isDelivered: async (room) => + (await storage.get(DELIVERED_PREFIX + room)) !== undefined, + markDelivered: async (room) => { + await storage.set(DELIVERED_PREFIX + room, new Uint8Array([1])); + }, }; } diff --git a/ts/packages/web-console/test/RoomPanel.unit.test.tsx b/ts/packages/web-console/test/RoomPanel.unit.test.tsx index 5be3e314..277839e3 100644 --- a/ts/packages/web-console/test/RoomPanel.unit.test.tsx +++ b/ts/packages/web-console/test/RoomPanel.unit.test.tsx @@ -102,6 +102,17 @@ describe("RoomPanel", () => { ); }); + it("offers the notice composer over a hub as well as a direct connection, and not while there is no route", () => { + renderPanel({ view: view({ via: "hub" }) }); + expect(screen.getByPlaceholderText("Post a durable notice")).toBeVisible(); + cleanup(); + + renderPanel({ view: view({ via: undefined }) }); + expect( + screen.queryByPlaceholderText("Post a durable notice"), + ).not.toBeInTheDocument(); + }); + it("renders the session's notices through the notices view", () => { renderPanel({ view: view({ diff --git a/ts/packages/web-console/test/conversations.unit.test.ts b/ts/packages/web-console/test/conversations.unit.test.ts index bc9d2a3a..59924003 100644 --- a/ts/packages/web-console/test/conversations.unit.test.ts +++ b/ts/packages/web-console/test/conversations.unit.test.ts @@ -39,7 +39,10 @@ let session: MeshSession; /** A route through a hub of its own: each call is a different hub, so a test can close one and not another. */ function relayRoute(): RelayRoute { return { - hub: { sendManageRequest: async () => Promise.resolve({ result: "ok" }) }, + hub: { + sendManageRequest: async () => Promise.resolve({ result: "ok" }), + sendDataFrame: async () => Promise.resolve(), + }, peer: deviceIdFromHex(PEER), }; } diff --git a/ts/packages/web-console/test/e2e/notices-offline.e2e.spec.ts b/ts/packages/web-console/test/e2e/notices-offline.e2e.spec.ts new file mode 100644 index 00000000..9f4e8f51 --- /dev/null +++ b/ts/packages/web-console/test/e2e/notices-offline.e2e.spec.ts @@ -0,0 +1,107 @@ +// A notice posted while the other device is offline is still there when that device comes back (wire-mesh#274). Two independent consoles connect to a hub that holds logs for absent devices, with no direct connection possible between them, so everything that reaches the second device does so through the hub. The second device closes its page, the first posts, and the second opens the console again over the same storage: its stored token, epoch key and conversation survive the restart, and the hub hands it the log it missed. + +import { + type BrowserContext, + type Page, + chromium, + expect, + test, +} from "@playwright/test"; +import { MAILBOX_RELAY_ADDRESS, VITE_PORT } from "../../playwright.config.js"; + +// No UDP candidates at all, so no direct connection can open between the two browsers. +const NO_DIRECT_CONNECTION_ARGS = [ + "--force-webrtc-ip-handling-policy=disable_non_proxied_udp", +]; +const TEST_TIMEOUT_MS = 90_000; +const STEP_TIMEOUT_MS = 15_000; +const FIRST_NOTICE = "first notice, written while both are online"; +const OFFLINE_NOTICE = "second notice, written while the other is away"; +const MESSAGE_TEXT = "hello"; +const REPLY_TEXT = "hello back"; + +// A click or fill that cannot find its target fails at its own line instead of consuming the whole test timeout. +test.use({ actionTimeout: STEP_TIMEOUT_MS }); + +async function openConsole(context: Readonly): Promise { + const page = await context.newPage(); + page.on("pageerror", (error) => { + throw error; + }); + await page.goto( + new URL("/", `http://localhost:${String(VITE_PORT)}`).toString(), + ); + await page.getByLabel("Node").fill(MAILBOX_RELAY_ADDRESS); + await page.getByRole("button", { name: "Connect" }).click(); + return page; +} + +/** Posts through the composer's Enter key: the "Ready to work offline" toast a fresh console shows can sit over the Post button. */ +async function postNotice(page: Readonly, text: string): Promise { + const composer = page.getByPlaceholder("Post a durable notice"); + await composer.fill(text); + await composer.press("Enter"); +} + +test("a notice written while the peer is offline is read after the peer reloads", async () => { + test.setTimeout(TEST_TIMEOUT_MS); + const browserA = await chromium.launch({ args: NO_DIRECT_CONNECTION_ARGS }); + const browserB = await chromium.launch({ args: NO_DIRECT_CONNECTION_ARGS }); + try { + const contextA = await browserA.newContext(); + const contextB = await browserB.newContext(); + const pageA = await openConsole(contextA); + const pageB = await openConsole(contextB); + + const messageButton = pageA.getByRole("button", { name: "Message" }); + await expect(messageButton).toBeVisible({ timeout: STEP_TIMEOUT_MS }); + await expect(pageB.getByRole("button", { name: "Message" })).toBeVisible({ + timeout: STEP_TIMEOUT_MS, + }); + await messageButton.click(); + + // Consent runs in each direction, and each side then holds a token for the conversation. + await pageA.getByPlaceholder("Message", { exact: true }).fill(MESSAGE_TEXT); + await pageA.getByRole("button", { name: "Send", exact: true }).click(); + await expect(pageB.getByText("wants to message you")).toBeVisible({ + timeout: STEP_TIMEOUT_MS, + }); + await pageB.getByRole("button", { name: "Allow" }).click(); + await expect(pageB.getByText(MESSAGE_TEXT, { exact: true })).toBeVisible({ + timeout: STEP_TIMEOUT_MS, + }); + await pageB.getByPlaceholder("Message", { exact: true }).fill(REPLY_TEXT); + await pageB.getByRole("button", { name: "Send", exact: true }).click(); + await expect(pageA.getByText("wants to message you")).toBeVisible({ + timeout: STEP_TIMEOUT_MS, + }); + await pageA.getByRole("button", { name: "Allow" }).click(); + await expect(pageA.getByText(REPLY_TEXT, { exact: true })).toBeVisible({ + timeout: STEP_TIMEOUT_MS, + }); + + // The first notice is posted with both online: it hands B the conversation's key through the hub. + await postNotice(pageA, FIRST_NOTICE); + await expect(pageA.getByText(FIRST_NOTICE, { exact: true })).toBeVisible({ + timeout: STEP_TIMEOUT_MS, + }); + + // B goes away; A writes a second notice that only the hub can hold for it. + await pageB.close(); + await postNotice(pageA, OFFLINE_NOTICE); + await expect(pageA.getByText(OFFLINE_NOTICE, { exact: true })).toBeVisible({ + timeout: STEP_TIMEOUT_MS, + }); + + // B comes back on the same storage and reconnects: both notices are read, the second one from the hub. + const reopenedB = await openConsole(contextB); + await expect( + reopenedB.getByText(OFFLINE_NOTICE, { exact: true }), + ).toBeVisible({ timeout: STEP_TIMEOUT_MS }); + await expect( + reopenedB.getByText(FIRST_NOTICE, { exact: true }), + ).toBeVisible({ timeout: STEP_TIMEOUT_MS }); + } finally { + await Promise.all([browserA.close(), browserB.close()]); + } +}); diff --git a/ts/packages/web-console/test/notices.unit.test.ts b/ts/packages/web-console/test/notices.unit.test.ts index e844835d..8928f942 100644 --- a/ts/packages/web-console/test/notices.unit.test.ts +++ b/ts/packages/web-console/test/notices.unit.test.ts @@ -27,7 +27,13 @@ import type { } from "wire-mesh-core/generated/protocol"; import type { IdentityPort } from "wire-mesh-core/ports/identity"; import type { Clock } from "wire-mesh-core/ports/clock"; -import { bootstrapDmEpoch1, createNoticeWiring } from "../src/notices.js"; +import { + bootstrapDmEpoch1, + createNoticeWiring, + type DataFrameSender, + type RoomKeyDelivery, + type NoticeWiring, +} from "../src/notices.js"; const ES256 = -7; const HOUR_MS = 3_600_000; @@ -69,6 +75,17 @@ async function generateEs256Identity(): Promise { ); } +function memoryDelivery(delivered: readonly string[] = []): RoomKeyDelivery { + const rooms = new Set(delivered); + return { + isDelivered: async (room) => Promise.resolve(rooms.has(room)), + markDelivered: async (room) => { + rooms.add(room); + return Promise.resolve(); + }, + }; +} + function memoryKeyStore(): RoomKeyStore { const keys = new Map>(); return { @@ -92,7 +109,10 @@ function memoryKeyStore(): RoomKeyStore { function fakeSession(): MeshSession & { sent: Frame[]; commands: ManageCommand[]; + /** Makes every later manage-request come back as an error with this code, or succeed again when undefined. */ + refuseManageRequests: (code: string | undefined) => void; } { + let refusal: string | undefined; const sent: Frame[] = []; const commands: ManageCommand[] = []; const emptyStream = (): AsyncIterable => ({ @@ -109,6 +129,9 @@ function fakeSession(): MeshSession & { return { sent, commands, + refuseManageRequests: (code) => { + refusal = code; + }, events: emptyStream(), incomingManageRequests: emptyStream(), revocationAnnouncements: emptyStream(), @@ -128,7 +151,11 @@ function fakeSession(): MeshSession & { }, sendManageRequest: async (command: ManageCommand) => { commands.push(command); - return Promise.resolve({ result: "ok" } as const); + return Promise.resolve( + refusal === undefined + ? ({ result: "ok" } as const) + : ({ result: "error", code: refusal } as const), + ); }, }; } @@ -142,17 +169,17 @@ async function flush(): Promise { /** Cross-delivery drain: forward each side's outbound frames to the other wiring, flushing async handlers between hops, until neither side queues anything further -- a bounded stand-in for real transports. */ async function drain( ownerSession: MeshSession & { sent: Frame[] }, - ownerWiring: Readonly<{ handleFrame: (f: Frame) => void }>, + ownerWiring: NoticeWiring, memberSession: MeshSession & { sent: Frame[] }, - memberWiring: Readonly<{ handleFrame: (f: Frame) => void }>, + memberWiring: NoticeWiring, ): Promise { const MAX_DRAIN_ROUNDS = 8; for (let round = 0; round < MAX_DRAIN_ROUNDS; round += 1) { const ownerFrames = ownerSession.sent.splice(0); const memberFrames = memberSession.sent.splice(0); if (ownerFrames.length === 0 && memberFrames.length === 0) break; - for (const f of ownerFrames) memberWiring.handleFrame(f); - for (const f of memberFrames) ownerWiring.handleFrame(f); + for (const f of ownerFrames) memberWiring.handleFrame(f, memberSession); + for (const f of memberFrames) ownerWiring.handleFrame(f, ownerSession); await flush(); } } @@ -204,6 +231,7 @@ describe("bootstrapDmEpoch1", () => { ownRoomMemberToken: lowerToken, roomPath, roomKeys: lowerKeys, + delivery: memoryDelivery(), }); // The lower side stored its own epoch-1 key and sent exactly one rekey @@ -267,11 +295,12 @@ describe("bootstrapDmEpoch1", () => { ownRoomMemberToken: higherToken, roomPath, roomKeys: higherKeys, + delivery: memoryDelivery(), }); expect(higherSession.commands).toHaveLength(0); expect(await higherKeys.currentEpoch(roomPath)).toBeUndefined(); - // Idempotent for the lower side once an epoch exists. + // Idempotent for the lower side once the peer has acknowledged the key. const lowerToken = await mintRoomMemberToken(higher, lower, roomPath); const lowerSession = fakeSession(); const lowerKeys = memoryKeyStore(); @@ -284,11 +313,64 @@ describe("bootstrapDmEpoch1", () => { ownRoomMemberToken: lowerToken, roomPath, roomKeys: lowerKeys, + delivery: memoryDelivery([roomPath]), }); expect(lowerSession.commands).toHaveLength(0); }); }); +describe("bootstrapDmEpoch1 redelivery", () => { + async function dmPair(): Promise<{ + lower: IdentityPort; + lowerToken: CapabilityToken; + roomPath: string; + }> { + const a = await generateEs256Identity(); + const b = await generateEs256Identity(); + const aIsLower = deviceIdToHex(a.deviceId) < deviceIdToHex(b.deviceId); + const lower = aIsLower ? a : b; + const higher = aIsLower ? b : a; + const roomPath = `${deviceIdToHex(lower.deviceId)}+${deviceIdToHex(higher.deviceId)}`; + return { + lower, + lowerToken: await mintRoomMemberToken(higher, lower, roomPath), + roomPath, + }; + } + + it("sends the same first epoch key again until the peer has taken it", async () => { + const { lower, lowerToken, roomPath } = await dmPair(); + const session = fakeSession(); + const keys = memoryKeyStore(); + const delivery = memoryDelivery(); + const options = { + session, + identity: lower, + clock: fixedClock(NOW_MS), + revocation: createRevocationView(), + ownRoomMemberToken: lowerToken, + roomPath, + roomKeys: keys, + delivery, + }; + + session.refuseManageRequests("no_token"); + await expect(bootstrapDmEpoch1(options)).rejects.toThrow("no_token"); + const minted = await keys.get(roomPath, 1); + expect(await delivery.isDelivered(roomPath)).toBe(false); + + session.refuseManageRequests(undefined); + await bootstrapDmEpoch1(options); + + expect(await keys.get(roomPath, 1)).toEqual(minted); + expect(session.commands).toHaveLength(2); + expect(await delivery.isDelivered(roomPath)).toBe(true); + + await bootstrapDmEpoch1(options); + expect(session.commands).toHaveLength(2); + }); +}); + describe("createNoticeWiring", () => { it("a posted notice replicates to the peer and decrypts there, end to end over the frame flow", async () => { const owner = await generateEs256Identity(); @@ -336,7 +418,7 @@ describe("createNoticeWiring", () => { const memberSession = fakeSession(); let memberChanges = 0; const ownerWiring = createNoticeWiring({ - session: ownerSession, + senders: () => [ownerSession], storage: createMemoryStorage(), identity: owner, clock: fixedClock(NOW_MS), @@ -345,7 +427,7 @@ describe("createNoticeWiring", () => { onChange: () => undefined, }); const memberWiring = createNoticeWiring({ - session: memberSession, + senders: () => [memberSession], storage: createMemoryStorage(), identity: member, clock: fixedClock(NOW_MS), @@ -370,7 +452,7 @@ describe("createNoticeWiring", () => { // member's received state: its storage holds the owner's entry and its // read returns the decrypted notice. expect(memberChanges).toBeGreaterThan(0); - const memberRead = await memberWiring.readRoom(roomPath); + const memberRead = await memberWiring.readRoom(roomPath, [owner.deviceId]); const replicated = memberRead.find( (n) => n.verified && n.plaintext !== undefined, ); @@ -391,7 +473,7 @@ describe("createNoticeWiring", () => { await keys.set(roomPath, FIRST_EPOCH, generateContentKey()); const session = fakeSession(); const wiring = createNoticeWiring({ - session, + senders: () => [session], storage: createMemoryStorage(), identity: owner, clock: fixedClock(NOW_MS), @@ -407,7 +489,7 @@ describe("createNoticeWiring", () => { plaintext: new TextEncoder().encode("own log only"), }); - const read = await wiring.readRoom(roomPath); + const read = await wiring.readRoom(roomPath, []); expect(read).toHaveLength(1); expect(read[0]?.plaintext).toEqual( new TextEncoder().encode("own log only"), @@ -415,12 +497,12 @@ describe("createNoticeWiring", () => { expect(session.sent.some((f) => f.type === "data-have")).toBe(true); }); - it("an inbound data-have from an unknown peer seeds tracking and triggers a catch-up request", async () => { + it("an inbound data-have is answered with a catch-up request through the sender it arrived on", async () => { const owner = await generateEs256Identity(); const peer: DeviceId = Uint8Array.from({ length: 32 }, (_, i) => i); const session = fakeSession(); const wiring = createNoticeWiring({ - session, + senders: () => [session], storage: createMemoryStorage(), identity: owner, clock: fixedClock(NOW_MS), @@ -434,10 +516,95 @@ describe("createNoticeWiring", () => { peer, "head-seq": 5, }; - wiring.handleFrame(have); + wiring.handleFrame(have, session); await flush(); const request = session.sent.find((f) => f.type === "data-request"); expect(request).toMatchObject({ peer, "from-seq": 0 }); }); + + it("sync announces the own log to a sender and asks it for each peer's log", async () => { + const owner = await generateEs256Identity(); + const peer: DeviceId = Uint8Array.from({ length: 32 }, (_, i) => i); + const roomPath = ownerNamedRoomPath( + deviceIdToHex(owner.deviceId), + "general", + ); + const keys = memoryKeyStore(); + await keys.set(roomPath, FIRST_EPOCH, generateContentKey()); + const hub = fakeSession(); + const wiring = createNoticeWiring({ + senders: () => [], + storage: createMemoryStorage(), + identity: owner, + clock: fixedClock(NOW_MS), + revocation: createRevocationView(), + roomKeys: keys, + onChange: () => undefined, + }); + await wiring.post({ + room: roomPath, + token: await mintRoomMemberToken(owner, owner, roomPath), + contentType: "text/plain", + plaintext: new TextEncoder().encode("written while unconnected"), + }); + + await wiring.sync(hub, [peer]); + + expect(hub.sent).toEqual([ + { type: "data-have", peer: owner.deviceId, "head-seq": 1 }, + { type: "data-request", peer, "from-seq": 0 }, + ]); + }); + + it("sync announces nothing for an empty own log", async () => { + const owner = await generateEs256Identity(); + const hub = fakeSession(); + const wiring = createNoticeWiring({ + senders: () => [], + storage: createMemoryStorage(), + identity: owner, + clock: fixedClock(NOW_MS), + revocation: createRevocationView(), + roomKeys: memoryKeyStore(), + onChange: () => undefined, + }); + + await wiring.sync(hub, []); + + expect(hub.sent).toEqual([]); + }); + + it("a post still succeeds when one sender is not connected, and reaches the others", async () => { + const owner = await generateEs256Identity(); + const roomPath = ownerNamedRoomPath( + deviceIdToHex(owner.deviceId), + "general", + ); + const keys = memoryKeyStore(); + await keys.set(roomPath, FIRST_EPOCH, generateContentKey()); + const connected = fakeSession(); + const disconnected: DataFrameSender = { + sendDataFrame: async () => Promise.reject(new Error("not connected")), + }; + const wiring = createNoticeWiring({ + senders: () => [disconnected, connected], + storage: createMemoryStorage(), + identity: owner, + clock: fixedClock(NOW_MS), + revocation: createRevocationView(), + roomKeys: keys, + onChange: () => undefined, + }); + + await wiring.post({ + room: roomPath, + token: await mintRoomMemberToken(owner, owner, roomPath), + contentType: "text/plain", + plaintext: new TextEncoder().encode("hello"), + }); + + expect(connected.sent).toHaveLength(1); + expect(await wiring.readRoom(roomPath, [])).toHaveLength(1); + }); }); diff --git a/ts/packages/web-console/test/use-room-messaging-relay.unit.test.tsx b/ts/packages/web-console/test/use-room-messaging-relay.unit.test.tsx index e54b1f18..150ac15a 100644 --- a/ts/packages/web-console/test/use-room-messaging-relay.unit.test.tsx +++ b/ts/packages/web-console/test/use-room-messaging-relay.unit.test.tsx @@ -12,7 +12,10 @@ import type { import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; import { dmRoomPath } from "wire-mesh-core/domain/room-path"; import type { IdentityPort } from "wire-mesh-core/ports/identity"; -import { createWebCryptoIdentity } from "../src/adapters/web-crypto-identity.js"; +import { + createPersistedWebCryptoIdentity, + createWebCryptoIdentity, +} from "../src/adapters/web-crypto-identity.js"; import { useRoomMessaging } from "../src/hooks/use-room-messaging.js"; import { testCapabilities } from "./capability-services.js"; import type { TestCapabilities } from "./capability-services.js"; @@ -34,8 +37,14 @@ let roomPath: string; let capabilities: TestCapabilities; beforeAll(async () => { - own = await createWebCryptoIdentity(); - peer = await createWebCryptoIdentity(); + // Only a persisted identity can derive the ECDH secret the first epoch key is wrapped under. + const first = await createPersistedWebCryptoIdentity(createMemoryStorage()); + const second = await createPersistedWebCryptoIdentity(createMemoryStorage()); + // The lower device-id mints a DM's first epoch, so `own` is always that side. + const ownIsLower = + deviceIdToHex(first.deviceId) < deviceIdToHex(second.deviceId); + own = ownIsLower ? first : second; + peer = ownIsLower ? second : first; roomPath = dmRoomPath( deviceIdToHex(own.deviceId), deviceIdToHex(peer.deviceId), @@ -48,6 +57,8 @@ type SendManageRequest = MeshSession["sendManageRequest"]; /** A hub that answers a room.join with a grant from the peer and accepts every room.send. */ async function answeringHub(): Promise<{ sendManageRequest: ReturnType>; + sentDataFrames: Parameters[0][]; + sendDataFrame: MeshSession["sendDataFrame"]; }> { const grant = await mintRoomInviteGrant( peer, @@ -67,21 +78,24 @@ async function answeringHub(): Promise<{ : { result: "ok" }, ), ); - return { sendManageRequest }; + const sentDataFrames: Parameters[0][] = []; + return { + sendManageRequest, + sentDataFrames, + sendDataFrame: async (frame) => { + sentDataFrames.push(frame); + return Promise.resolve(); + }, + }; } function render(): ReturnType< typeof renderHook, unknown> > { const store = createMessageStore(createMemoryStorage()); + const roomStorage = createMemoryStorage(); return renderHook(() => - useRoomMessaging( - own, - clock, - store, - createMemoryStorage(), - capabilities.services, - ), + useRoomMessaging(own, clock, store, roomStorage, capabilities.services), ); } @@ -139,6 +153,7 @@ describe("a conversation over a hub", () => { act(() => { result.current.watchHub({ sendManageRequest: async () => Promise.resolve({ result: "ok" }), + sendDataFrame: async () => Promise.resolve(), incomingManageRequests: incoming.stream, }); }); @@ -177,6 +192,7 @@ describe("a conversation over a hub", () => { act(() => { result.current.watchHub({ sendManageRequest: async () => Promise.resolve({ result: "ok" }), + sendDataFrame: async () => Promise.resolve(), incomingManageRequests: incoming.stream, }); }); @@ -215,6 +231,7 @@ describe("a conversation over a hub", () => { act(() => { result.current.watchHub({ sendManageRequest: async () => Promise.resolve({ result: "ok" }), + sendDataFrame: async () => Promise.resolve(), incomingManageRequests: incoming.stream, }); }); @@ -246,16 +263,27 @@ describe("a conversation over a hub", () => { expect(conversation?.messages).toHaveLength(1); }); - it("does not post a durable notice, which needs a direct connection", async () => { + it("posts a durable notice over the hub: joins, hands the peer the first epoch key and announces the log", async () => { const { result } = render(); const hub = await answeringHub(); act(() => { result.current.openRelay(hub, peer.deviceId); }); - await expect(result.current.postNotice(roomPath, "note")).rejects.toThrow( - "durable notices need a direct connection", + await act(async () => { + await result.current.postNotice(roomPath, "note"); + }); + + const rekey = hub.sendManageRequest.mock.calls.find( + ([command]) => command.params.verb === "room.rekey", + ); + expect(rekey?.[2]).toEqual(peer.deviceId); + expect(hub.sentDataFrames.some((frame) => frame.type === "data-have")).toBe( + true, ); + await waitFor(() => { + expect(result.current.conversations[0]?.notices).toHaveLength(1); + }); }); it("records the grant a join is answered with as held", async () => { @@ -285,8 +313,9 @@ describe("a conversation over a hub", () => { record: async () => Promise.reject(new Error(STORAGE_FAILURE)), }, }; + const roomStorage = createMemoryStorage(); const { result } = renderHook(() => - useRoomMessaging(own, clock, store, createMemoryStorage(), services), + useRoomMessaging(own, clock, store, roomStorage, services), ); const hub = await answeringHub(); act(() => { @@ -307,6 +336,7 @@ describe("a conversation over a hub", () => { act(() => { result.current.watchHub({ sendManageRequest: async () => Promise.resolve({ result: "ok" }), + sendDataFrame: async () => Promise.resolve(), incomingManageRequests: incoming.stream, }); }); diff --git a/ts/packages/web-console/test/use-room-messaging.unit.test.tsx b/ts/packages/web-console/test/use-room-messaging.unit.test.tsx index a8eea96c..e78d4348 100644 --- a/ts/packages/web-console/test/use-room-messaging.unit.test.tsx +++ b/ts/packages/web-console/test/use-room-messaging.unit.test.tsx @@ -35,8 +35,9 @@ async function renderRestoredConversation(): Promise< sentAt: 1, }); const { services } = await testCapabilities(identity, clock); + const roomStorage = createMemoryStorage(); const rendered = renderHook(() => - useRoomMessaging(identity, clock, store, createMemoryStorage(), services), + useRoomMessaging(identity, clock, store, roomStorage, services), ); await waitFor(() => { expect(rendered.result.current.conversations).toHaveLength(1);