diff --git a/desktop/src/features/messages/hooks.ts b/desktop/src/features/messages/hooks.ts index 7c10967efe..952416c216 100644 --- a/desktop/src/features/messages/hooks.ts +++ b/desktop/src/features/messages/hooks.ts @@ -4,6 +4,7 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { channelMessagesKey, dedupeMessagesById, + mergeTimelineHistoryMessages, normalizeTimelineMessages, sortMessages, } from "@/features/messages/lib/messageQueryKeys"; @@ -180,10 +181,10 @@ export function useChannelMessagesQuery(channel: Channel | null) { ); const currentMessages = queryClient.getQueryData(queryKey) ?? []; - const mergedHistory = normalizeTimelineMessages([ - ...currentMessages, - ...history, - ]); + const mergedHistory = mergeTimelineHistoryMessages( + currentMessages, + history, + ); return mergedHistory; }, @@ -208,14 +209,7 @@ export function useChannelSubscription(channel: Channel | null) { queryClient.setQueryData( channelMessagesKey(channelId), - (current = []) => { - const mergedHistory = normalizeTimelineMessages([ - ...current, - ...history, - ]); - - return mergedHistory; - }, + (current = []) => mergeTimelineHistoryMessages(current, history), ); }); diff --git a/desktop/src/features/messages/lib/messageQueryKeys.test.mjs b/desktop/src/features/messages/lib/messageQueryKeys.test.mjs new file mode 100644 index 0000000000..6c954f34ed --- /dev/null +++ b/desktop/src/features/messages/lib/messageQueryKeys.test.mjs @@ -0,0 +1,204 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + mergeTimelineHistoryMessages, + normalizeTimelineMessages, +} from "./messageQueryKeys.ts"; +import { mergeTimelineCacheMessages } from "../hooks.ts"; + +const CHANNEL_ID = "timeline-window-test"; +const PUBKEY = "a".repeat(64); + +function event({ id, kind = 9, createdAt, tags, content = "" }) { + return { + id, + pubkey: PUBKEY, + created_at: createdAt, + kind, + tags: tags ?? [["h", CHANNEL_ID]], + content, + sig: "mocksig".repeat(20).slice(0, 128), + }; +} + +function id(prefix, index) { + return `${prefix}${String(index).padStart(64 - prefix.length, "0")}`; +} + +test("normalizeTimelineMessages caps visible content, not unrelated auxiliary events", () => { + const june12Roots = [ + "1f86d0450b3c2c376691e7a8232fbd8a5b8408ecad2f5eb0209e7bfcfdf9af80", + "3b745de3d9ff91c464b0cbf26e3e628a5a8c05dccf3f9781b1fbd99a0f6f5e7b", + "cb404eeae1517bb2ed2e9975dfd2efd12d0a7ef17b87c3a6573b251b9865f7a4", + "337de49c712fcf84f5689a6c11ce36018817197d578468b8132c6dc3d1a13131", + "ac683f35cda2e8c1e9ae609e0b9f5dc23a7d434637b4f0505495c3a2b6f52aae", + ]; + const messages = []; + + for (let index = 0; index < 500; index += 1) { + messages.push(event({ id: id("old", index), createdAt: 1_000 + index })); + } + for (const [index, rootId] of june12Roots.entries()) { + messages.push( + event({ id: rootId, createdAt: 2_000 + index, content: "June 12 root" }), + ); + } + for (let index = 0; index < 1_303; index += 1) { + messages.push( + event({ + id: id("del", index), + kind: 5, + createdAt: 3_000 + index, + tags: [ + ["h", CHANNEL_ID], + ["e", id("zzz", index)], + ], + }), + ); + } + for (let index = 0; index < 231; index += 1) { + messages.push( + event({ + id: id("rea", index), + kind: 7, + createdAt: 5_000 + index, + tags: [ + ["h", CHANNEL_ID], + ["e", id("yyy", index)], + ], + content: "+", + }), + ); + } + for (let index = 0; index < 1_495; index += 1) { + messages.push(event({ id: id("new", index), createdAt: 6_000 + index })); + } + + const normalized = normalizeTimelineMessages(messages); + + assert.equal(normalized.filter((item) => item.kind === 9).length, 2_000); + assert.deepEqual( + june12Roots.map((rootId) => normalized.some((item) => item.id === rootId)), + [true, true, true, true, true], + ); + assert.equal(normalized.filter((item) => item.kind === 5).length, 1_303); + assert.equal(normalized.filter((item) => item.kind === 7).length, 231); +}); + +test("normalizeTimelineMessages still caps old visible content", () => { + const retainedRoot = `${"a".repeat(63)}1`; + const reaction = `${"b".repeat(63)}1`; + const reactionDeletion = `${"c".repeat(63)}1`; + const messages = []; + + for (let index = 0; index < 2_000; index += 1) { + messages.push(event({ id: id("old", index), createdAt: 1_000 + index })); + } + messages.push(event({ id: retainedRoot, createdAt: 4_000 })); + messages.push( + event({ + id: reaction, + kind: 7, + createdAt: 4_001, + tags: [ + ["h", CHANNEL_ID], + ["e", retainedRoot], + ], + content: "+", + }), + ); + messages.push( + event({ + id: reactionDeletion, + kind: 5, + createdAt: 4_002, + tags: [ + ["h", CHANNEL_ID], + ["e", reaction], + ], + }), + ); + + const normalized = normalizeTimelineMessages(messages); + + assert.equal( + normalized.some((item) => item.id === id("old", 0)), + false, + ); + assert.equal( + normalized.some((item) => item.id === retainedRoot), + true, + ); + assert.equal( + normalized.some((item) => item.id === reaction), + true, + ); + assert.equal( + normalized.some((item) => item.id === reactionDeletion), + true, + ); +}); + +test("timeline history and live cache merges retain the same visible content regardless of order", () => { + const seedMessages = []; + const olderPage = []; + const liveMessage = event({ id: id("liv", 0), createdAt: 20_000 }); + + for (let index = 0; index < 700; index += 1) { + seedMessages.push( + event({ id: id("new", index), createdAt: 10_000 + index }), + ); + } + for (let index = 0; index < 1_303; index += 1) { + seedMessages.push( + event({ + id: id("del", index), + kind: 5, + createdAt: 11_000 + index, + tags: [ + ["h", CHANNEL_ID], + ["e", id("zzz", index)], + ], + }), + ); + } + for (let index = 0; index < 231; index += 1) { + seedMessages.push( + event({ + id: id("rea", index), + kind: 7, + createdAt: 13_000 + index, + tags: [ + ["h", CHANNEL_ID], + ["e", id("yyy", index)], + ], + content: "+", + }), + ); + } + for (let index = 0; index < 1_500; index += 1) { + olderPage.push(event({ id: id("old", index), createdAt: 1_000 + index })); + } + + const historyThenLive = mergeTimelineCacheMessages( + mergeTimelineHistoryMessages(seedMessages, olderPage), + liveMessage, + ); + const liveThenHistory = mergeTimelineHistoryMessages( + mergeTimelineCacheMessages(seedMessages, liveMessage), + olderPage, + ); + const historyThenLiveContent = historyThenLive + .filter((item) => item.kind === 9) + .map((item) => item.id); + const liveThenHistoryContent = liveThenHistory + .filter((item) => item.kind === 9) + .map((item) => item.id); + + assert.equal(historyThenLiveContent.length, 2_000); + assert.equal(liveThenHistoryContent.length, 2_000); + assert.deepEqual(liveThenHistoryContent, historyThenLiveContent); + assert.equal(historyThenLiveContent[0], id("old", 201)); + assert.equal(historyThenLiveContent.at(-1), liveMessage.id); +}); diff --git a/desktop/src/features/messages/lib/messageQueryKeys.ts b/desktop/src/features/messages/lib/messageQueryKeys.ts index 316ca3a7a6..0ea36739f6 100644 --- a/desktop/src/features/messages/lib/messageQueryKeys.ts +++ b/desktop/src/features/messages/lib/messageQueryKeys.ts @@ -1,4 +1,16 @@ import type { RelayEvent } from "@/shared/api/types"; +import { + KIND_JOB_ACCEPTED, + KIND_JOB_CANCEL, + KIND_JOB_ERROR, + KIND_JOB_PROGRESS, + KIND_JOB_REQUEST, + KIND_JOB_RESULT, + KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_DIFF, + KIND_STREAM_MESSAGE_V2, + KIND_SYSTEM_MESSAGE, +} from "@/shared/constants/kinds"; const MAX_TIMELINE_MESSAGES = 2_000; @@ -30,17 +42,51 @@ export function sortMessages(messages: RelayEvent[]) { ); } +function isTimelineWindowContentEvent(event: RelayEvent) { + return ( + event.kind === KIND_STREAM_MESSAGE || + event.kind === KIND_STREAM_MESSAGE_V2 || + event.kind === KIND_STREAM_MESSAGE_DIFF || + event.kind === KIND_SYSTEM_MESSAGE || + event.kind === KIND_JOB_REQUEST || + event.kind === KIND_JOB_ACCEPTED || + event.kind === KIND_JOB_PROGRESS || + event.kind === KIND_JOB_RESULT || + event.kind === KIND_JOB_CANCEL || + event.kind === KIND_JOB_ERROR + ); +} + /** - * Sort, dedupe, and cap the timeline at {@link MAX_TIMELINE_MESSAGES} so - * de-virtualized rendering does not grow into an unbounded DOM during - * long-lived channel sessions. + * Sort, dedupe, and cap the timeline at {@link MAX_TIMELINE_MESSAGES} visible + * content events so de-virtualized rendering does not grow into an unbounded + * DOM during long-lived channel sessions. + * + * Auxiliary events (reactions, edits, tombstones) are kept in cache so they can + * still apply to retained or later-loaded content, but they must not consume the + * visible message window and evict older loaded roots. */ export function normalizeTimelineMessages(messages: RelayEvent[]) { const normalized = sortMessages(messages); + const contentEvents = normalized.filter(isTimelineWindowContentEvent); - if (normalized.length <= MAX_TIMELINE_MESSAGES) { + if (contentEvents.length <= MAX_TIMELINE_MESSAGES) { return normalized; } - return normalized.slice(-MAX_TIMELINE_MESSAGES); + const retainedContentIds = new Set( + contentEvents.slice(-MAX_TIMELINE_MESSAGES).map((event) => event.id), + ); + + return normalized.filter( + (event) => + !isTimelineWindowContentEvent(event) || retainedContentIds.has(event.id), + ); +} + +export function mergeTimelineHistoryMessages( + current: RelayEvent[], + history: RelayEvent[], +) { + return normalizeTimelineMessages([...current, ...history]); } diff --git a/desktop/src/features/messages/useFetchOlderMessages.ts b/desktop/src/features/messages/useFetchOlderMessages.ts index 01cd75d03b..0203199a4a 100644 --- a/desktop/src/features/messages/useFetchOlderMessages.ts +++ b/desktop/src/features/messages/useFetchOlderMessages.ts @@ -3,7 +3,7 @@ import { useQueryClient } from "@tanstack/react-query"; import { channelMessagesKey, - sortMessages, + mergeTimelineHistoryMessages, } from "@/features/messages/lib/messageQueryKeys"; import { relayClient } from "@/shared/api/relayClient"; import type { Channel, RelayEvent } from "@/shared/api/types"; @@ -64,7 +64,7 @@ export function useFetchOlderMessages(channel: Channel | null) { if (olderMessages.length > 0) { queryClient.setQueryData(queryKey, (current = []) => - sortMessages([...current, ...olderMessages]), + mergeTimelineHistoryMessages(current, olderMessages), ); const updatedMessages = diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 5ae49d3c55..1d14265a4f 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -21,6 +21,7 @@ import { type RelaySubscription, type RelaySubscriptionFilter, } from "@/shared/api/relayClientShared"; +import { replayLiveSubscriptions } from "@/shared/api/relayReconnectReplay"; import { RelayConnectionStateEmitter } from "@/shared/api/relayConnectionStateEmitter"; import { shouldRefuseConnect, @@ -31,7 +32,6 @@ import { buildThreadReferenceTags } from "@/features/messages/lib/threading"; const RECONNECT_BASE_DELAY_MS = 1_000, RECONNECT_MAX_DELAY_MS = 30_000, - RECONNECT_REPLAY_SKEW_SECS = 5, EVENT_BATCH_MS = 16; /** @@ -165,7 +165,10 @@ export class RelayClient { private async fetchHistory(filter: RelaySubscriptionFilter) { await this.ensureConnected(); + return this.requestHistory(filter); + } + private requestHistory(filter: RelaySubscriptionFilter) { return new Promise((resolve, reject) => { const subId = `history-${crypto.randomUUID()}`; const timeout = window.setTimeout(() => { @@ -860,45 +863,20 @@ export class RelayClient { return false; } - private buildReplayFilter(filter: RelaySubscriptionFilter, since?: number) { - if (since === undefined) { - return filter; - } - - return { - ...filter, - since: filter.since === undefined ? since : Math.max(filter.since, since), - }; - } - private async replayLiveSubscriptions() { - for (const [subId, subscription] of this.subscriptions) { - if (subscription.mode !== "live") { - continue; - } - - const replaySince = - subscription.lastSeenCreatedAt === undefined - ? undefined - : Math.max( - 0, - subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS, - ); - - try { - await this.sendRaw([ - "REQ", - subId, - this.buildReplayFilter(subscription.filter, replaySince), - ]); - } catch (error) { - const reconnectError = - error instanceof Error - ? error - : new Error("Failed to restore relay subscriptions."); - this.resetConnection(reconnectError); - throw reconnectError; - } + try { + await replayLiveSubscriptions({ + subscriptions: this.subscriptions, + sendRaw: (payload) => this.sendRaw(payload), + requestHistory: (filter) => this.requestHistory(filter), + }); + } catch (error) { + const reconnectError = + error instanceof Error + ? error + : new Error("Failed to restore relay subscriptions."); + this.resetConnection(reconnectError); + throw reconnectError; } } diff --git a/desktop/src/shared/api/relayReconnectReplay.test.mjs b/desktop/src/shared/api/relayReconnectReplay.test.mjs new file mode 100644 index 0000000000..4f1c2731e5 --- /dev/null +++ b/desktop/src/shared/api/relayReconnectReplay.test.mjs @@ -0,0 +1,175 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + buildReconnectReplayFilter, + replayLiveSubscriptions, +} from "./relayReconnectReplay.ts"; +import { RelayClient } from "./relayClientSession.ts"; + +function replayFilter(filter, since, until) { + return buildReconnectReplayFilter(filter, since, until); +} + +function event(id, createdAt) { + return { + id, + pubkey: "pubkey", + created_at: createdAt, + kind: 9, + tags: [], + content: "", + sig: "sig", + }; +} + +function eventRange(prefix, start, count) { + return Array.from({ length: count }, (_, index) => + event(`${prefix}-${index}`, start + index), + ); +} + +test("reconnect replay preserves small steady-state limits when adding since", () => { + const filter = { + kinds: [9, 40002], + "#h": ["channel-1"], + limit: 50, + }; + + assert.deepEqual(replayFilter(filter, 123), { + kinds: [9, 40002], + "#h": ["channel-1"], + limit: 50, + since: 123, + }); +}); + +test("reconnect replay caps large steady-state limits", () => { + const filter = { + kinds: [9], + "#h": ["channel-1"], + limit: 1000, + }; + + assert.deepEqual(replayFilter(filter, 123), { + kinds: [9], + "#h": ["channel-1"], + limit: 500, + since: 123, + }); +}); + +test("reconnect replay keeps the stricter existing since window", () => { + const filter = { + kinds: [9], + "#h": ["channel-1"], + limit: 50, + since: 200, + }; + + assert.deepEqual(replayFilter(filter, 123), { + kinds: [9], + "#h": ["channel-1"], + limit: 50, + since: 200, + }); +}); + +test("reconnect replay applies the stricter until window", () => { + const filter = { + kinds: [9], + "#h": ["channel-1"], + limit: 50, + until: 300, + }; + + assert.deepEqual(replayFilter(filter, 123, 400), { + kinds: [9], + "#h": ["channel-1"], + limit: 50, + since: 123, + until: 300, + }); +}); + +test("initial subscription replay preserves the original filter", () => { + const filter = { + kinds: [9], + "#h": ["channel-1"], + limit: 50, + }; + + assert.equal(replayFilter(filter, undefined), filter); +}); + +test("channel reconnect replay pages the missed window until a short page", async () => { + const delivered = []; + const historyFilters = []; + const sentPayloads = []; + const pages = [ + eventRange("newest", 1501, 500), + eventRange("middle", 1002, 500), + eventRange("oldest", 995, 8), + ]; + const client = new RelayClient(); + const filter = client.buildChannelFilter("channel-1", 50); + const subscriptions = new Map([ + [ + "live-1", + { + mode: "live", + filter, + onEvent: (event) => delivered.push(event), + lastSeenCreatedAt: 1000, + }, + ], + ]); + + await replayLiveSubscriptions({ + subscriptions, + now: 2000, + sendRaw: async (payload) => { + sentPayloads.push(payload); + }, + requestHistory: async (filter) => { + historyFilters.push(filter); + return pages.shift() ?? []; + }, + }); + + assert.deepEqual(sentPayloads, [ + [ + "REQ", + "live-1", + { + kinds: filter.kinds, + "#h": ["channel-1"], + limit: 50, + }, + ], + ]); + assert.deepEqual(historyFilters, [ + { + kinds: filter.kinds, + "#h": ["channel-1"], + limit: 500, + since: 995, + until: 2000, + }, + { + kinds: filter.kinds, + "#h": ["channel-1"], + limit: 500, + since: 995, + until: 1501, + }, + { + kinds: filter.kinds, + "#h": ["channel-1"], + limit: 500, + since: 995, + until: 1002, + }, + ]); + assert.equal(delivered.length, 1008); +}); diff --git a/desktop/src/shared/api/relayReconnectReplay.ts b/desktop/src/shared/api/relayReconnectReplay.ts new file mode 100644 index 0000000000..d41f2c3ba8 --- /dev/null +++ b/desktop/src/shared/api/relayReconnectReplay.ts @@ -0,0 +1,124 @@ +import { CHANNEL_EVENT_KINDS } from "@/shared/constants/kinds"; +import type { + RelaySubscription, + RelaySubscriptionFilter, +} from "@/shared/api/relayClientShared"; +import type { RelayEvent } from "@/shared/api/types"; + +const RECONNECT_REPLAY_SKEW_SECS = 5; +export const RECONNECT_REPLAY_PAGE_LIMIT = 500; + +export function buildReconnectReplayFilter( + filter: RelaySubscriptionFilter, + since?: number, + until?: number, + limit = Math.min(filter.limit, RECONNECT_REPLAY_PAGE_LIMIT), +) { + if (since === undefined) return filter; + + const replayFilter: RelaySubscriptionFilter = { + ...filter, + limit, + since: filter.since === undefined ? since : Math.max(filter.since, since), + }; + + if (until !== undefined) { + replayFilter.until = + filter.until === undefined ? until : Math.min(filter.until, until); + } + + return replayFilter; +} + +export function shouldPageReconnectReplay(filter: RelaySubscriptionFilter) { + return ( + filter.limit > 0 && + Array.isArray(filter["#h"]) && + filter["#h"].length === 1 && + CHANNEL_EVENT_KINDS.every((kind) => filter.kinds.includes(kind)) + ); +} + +export async function replayReconnectHistoryPages({ + subscription, + since, + until, + isActive, + requestHistory, +}: { + subscription: Extract; + since: number; + until: number; + isActive: () => boolean; + requestHistory: (filter: RelaySubscriptionFilter) => Promise; +}) { + let pageUntil = until; + + while (pageUntil >= since) { + if (!isActive()) return; + + const events = await requestHistory( + buildReconnectReplayFilter( + subscription.filter, + since, + pageUntil, + RECONNECT_REPLAY_PAGE_LIMIT, + ), + ); + + if (!isActive()) return; + + for (const event of events) subscription.onEvent(event); + if (events.length < RECONNECT_REPLAY_PAGE_LIMIT) return; + + const oldestCreatedAt = events[0]?.created_at; + if (oldestCreatedAt === undefined || oldestCreatedAt <= since) return; + + pageUntil = + oldestCreatedAt < pageUntil ? oldestCreatedAt : oldestCreatedAt - 1; + } +} + +export async function replayLiveSubscriptions({ + subscriptions, + sendRaw, + requestHistory, + now = Math.floor(Date.now() / 1_000), +}: { + subscriptions: Map; + sendRaw: (payload: unknown[]) => Promise; + requestHistory: (filter: RelaySubscriptionFilter) => Promise; + now?: number; +}) { + for (const [subId, subscription] of subscriptions) { + if (subscription.mode !== "live") continue; + + const replaySince = + subscription.lastSeenCreatedAt === undefined + ? undefined + : Math.max( + 0, + subscription.lastSeenCreatedAt - RECONNECT_REPLAY_SKEW_SECS, + ); + const shouldPageReplay = + replaySince !== undefined && + shouldPageReconnectReplay(subscription.filter); + await sendRaw([ + "REQ", + subId, + shouldPageReplay + ? subscription.filter + : buildReconnectReplayFilter(subscription.filter, replaySince), + ]); + + if (shouldPageReplay) { + await replayReconnectHistoryPages({ + subscription, + since: replaySince, + until: now, + isActive: () => subscriptions.get(subId) === subscription, + requestHistory, + }); + } + } +} diff --git a/desktop/src/testing/e2eBridge.ts b/desktop/src/testing/e2eBridge.ts index 743e40ded4..a97172df72 100644 --- a/desktop/src/testing/e2eBridge.ts +++ b/desktop/src/testing/e2eBridge.ts @@ -466,6 +466,9 @@ type MockFilter = { "#h"?: string[]; authors?: string[]; kinds?: number[]; + limit?: number; + since?: number; + until?: number; }; type MockSocket = { @@ -606,6 +609,7 @@ declare global { }>; __BUZZ_E2E_SET_RELAY_CONNECTION_STATE__?: (state: ConnectionState) => void; __BUZZ_E2E_SET_STALL_WEBSOCKET_SENDS__?: (stall: boolean) => void; + __BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__?: () => number; __BUZZ_E2E_SET_MESH__?: (mesh: { admitted?: boolean; models?: Array<{ id: string; name: string | null }>; @@ -2263,8 +2267,29 @@ function getMockMessageStore(channelId: string): RelayEvent[] { return seeded; } -function emitMockHistory(socket: MockSocket, subId: string, channelId: string) { - const events = getMockMessageStore(channelId); +function emitMockHistory( + socket: MockSocket, + subId: string, + channelId: string, + filter: MockFilter, +) { + const events = getMockMessageStore(channelId) + .filter((event) => { + if (filter.kinds && !filter.kinds.includes(event.kind)) { + return false; + } + if (filter.since !== undefined && event.created_at < filter.since) { + return false; + } + if (filter.until !== undefined && event.created_at > filter.until) { + return false; + } + return true; + }) + .sort((left, right) => right.created_at - left.created_at) + .slice(0, filter.limit ?? 50) + .sort((left, right) => left.created_at - right.created_at); + for (const event of events) { sendWsText(socket.handler, ["EVENT", subId, event]); } @@ -5847,7 +5872,7 @@ function sendToMockSocket(args: { return; } - emitMockHistory(socket, subId, channelId); + emitMockHistory(socket, subId, channelId, filter); return; } @@ -6100,6 +6125,11 @@ export function maybeInstallE2eTauriMocks() { config.mock.stallWebsocketSends = stall; if (!stall) mockWebsocketSendMutexWedged = false; }; + window.__BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__ = () => { + const socketIds = [...mockSockets.keys()]; + for (const socketId of socketIds) disconnectMockSocket(socketId); + return socketIds.length; + }; // Tests flip `admitted` to exercise the denial path: mesh_ensure_client_node // rejects when not admitted, which proves relay membership is the gate and // that the create flow surfaces denial copy without spawning the agent. diff --git a/desktop/tests/e2e/relay-reconnect.spec.ts b/desktop/tests/e2e/relay-reconnect.spec.ts index 63ae34588a..890940f3ae 100644 --- a/desktop/tests/e2e/relay-reconnect.spec.ts +++ b/desktop/tests/e2e/relay-reconnect.spec.ts @@ -19,6 +19,46 @@ async function setMockWebsocketSendsStalled( }, stall); } +async function disconnectMockWebsockets(page: import("@playwright/test").Page) { + const disconnected = await page.evaluate(() => { + const disconnect = ( + window as Window & { + __BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__?: () => number; + } + ).__BUZZ_E2E_DISCONNECT_MOCK_WEBSOCKETS__; + if (!disconnect) { + throw new Error("E2E mock websocket disconnect seam is not installed."); + } + return disconnect(); + }); + + expect(disconnected).toBeGreaterThan(0); +} + +async function emitMockMessages( + page: import("@playwright/test").Page, + messages: Array<{ content: string; createdAt: number }>, +) { + await page.evaluate((items) => { + const emit = ( + window as Window & { + __BUZZ_E2E_EMIT_MOCK_MESSAGE__?: (input: { + channelName: string; + content: string; + createdAt: number; + }) => unknown; + } + ).__BUZZ_E2E_EMIT_MOCK_MESSAGE__; + if (!emit) { + throw new Error("E2E mock message emitter is not installed."); + } + + for (const item of items) { + emit({ channelName: "general", ...item }); + } + }, messages); +} + async function driveConnectionDegraded( page: import("@playwright/test").Page, state: "reconnecting" | "stalled" | "disconnected", @@ -110,3 +150,36 @@ test("profile popover does not show relay reconnect controls", async ({ }); await expect(page.getByTestId("profile-popover-reconnect")).toHaveCount(0); }); + +test("reconnect backfills more missed channel messages than the live subscription limit", async ({ + page, +}) => { + await page.goto("/"); + await page.getByTestId("channel-general").click(); + await expect(page.getByTestId("chat-title")).toHaveText("general"); + + const baseCreatedAt = Math.floor(Date.now() / 1_000) - 300; + const seenBeforeDisconnect = "reconnect e2e seen before disconnect"; + await emitMockMessages(page, [ + { content: seenBeforeDisconnect, createdAt: baseCreatedAt }, + ]); + await expect(page.getByTestId("message-timeline")).toContainText( + seenBeforeDisconnect, + ); + + await disconnectMockWebsockets(page); + + const missedMessages = Array.from({ length: 260 }, (_, index) => ({ + content: `reconnect e2e missed ${String(index + 1).padStart(3, "0")}`, + createdAt: baseCreatedAt + index + 1, + })); + await emitMockMessages(page, missedMessages); + + await expect(page.getByTestId("message-timeline")).toContainText( + "reconnect e2e missed 001", + { timeout: 15_000 }, + ); + await expect(page.getByTestId("message-timeline")).toContainText( + "reconnect e2e missed 260", + ); +});