From 0e7008fbf6877ef63a5301a1bf2e705becc88270 Mon Sep 17 00:00:00 2001 From: Shakker Date: Fri, 10 Jul 2026 12:27:41 +0100 Subject: [PATCH] fix: subscribe ClickClack from startup tail cursor --- extensions/clickclack/src/gateway.test.ts | 77 ++++++++++++++++++- extensions/clickclack/src/gateway.ts | 77 +++++++++++++++---- extensions/clickclack/src/http-client.test.ts | 29 +++++++ extensions/clickclack/src/http-client.ts | 40 +++++++--- 4 files changed, 196 insertions(+), 27 deletions(-) diff --git a/extensions/clickclack/src/gateway.test.ts b/extensions/clickclack/src/gateway.test.ts index b2fbb886c847..f2ef0ef8e040 100644 --- a/extensions/clickclack/src/gateway.test.ts +++ b/extensions/clickclack/src/gateway.test.ts @@ -20,6 +20,7 @@ const mocks = vi.hoisted(() => ({ client: { me: vi.fn(), events: vi.fn(), + eventPage: vi.fn(), websocket: vi.fn(), channelMessages: vi.fn(), directMessages: vi.fn(), @@ -79,6 +80,19 @@ function createGatewayContext( }; } +function createBacklogEvent(index: number, type = "channel.updated") { + return { + id: `evt-${index}`, + cursor: `cursor-${index}`, + type, + workspace_id: "workspace-1", + channel_id: "chan-1", + seq: index, + created_at: "2026-01-01T00:00:00.000Z", + payload: type === "message.created" ? { message_id: "msg-1", author_id: "human-1" } : undefined, + }; +} + describe("ClickClack gateway", () => { beforeEach(() => { vi.clearAllMocks(); @@ -90,7 +104,7 @@ describe("ClickClack gateway", () => { avatar_url: "", created_at: "2026-01-01T00:00:00.000Z", }); - mocks.client.events.mockResolvedValue([]); + mocks.client.eventPage.mockResolvedValue({ events: [] }); mocks.resolveClickClackInboundAccess.mockResolvedValue({ shouldDispatch: true, commandAuthorized: true, @@ -118,6 +132,65 @@ describe("ClickClack gateway", () => { ]); }); + it("opens realtime from the server startup tail without dispatching the returned page", async () => { + mocks.client.eventPage.mockResolvedValueOnce({ + events: [createBacklogEvent(1)], + tailCursor: "cursor-501", + }); + const socket = new FakeSocket(); + mocks.client.websocket.mockReturnValue(socket); + const abort = new AbortController(); + const ctx = createGatewayContext(abort.signal); + const run = startClickClackGatewayAccount(ctx); + + await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(1)); + + expect(mocks.client.eventPage).toHaveBeenCalledWith("workspace-1", { includeTail: true }); + expect(mocks.client.websocket).toHaveBeenCalledWith("workspace-1", "cursor-501"); + expect(mocks.handleClickClackInbound).not.toHaveBeenCalled(); + + abort.abort(); + await run; + }); + + it("drains and processes every reconnect page before reopening realtime", async () => { + const firstSocket = new FakeSocket(); + const secondSocket = new FakeSocket(); + const firstReconnectPage = Array.from({ length: 500 }, (_, index) => + createBacklogEvent(index + 1), + ); + mocks.client.eventPage + .mockResolvedValueOnce({ events: [], tailCursor: "" }) + .mockResolvedValueOnce({ events: firstReconnectPage }) + .mockResolvedValueOnce({ events: [createBacklogEvent(501, "message.created")] }); + mocks.client.websocket.mockReturnValueOnce(firstSocket).mockReturnValueOnce(secondSocket); + const abort = new AbortController(); + const ctx = createGatewayContext(abort.signal); + const run = startClickClackGatewayAccount(ctx); + + await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(1)); + firstSocket.emit("close"); + await vi.waitFor(() => expect(mocks.client.websocket).toHaveBeenCalledTimes(2)); + + expect(mocks.client.eventPage).toHaveBeenNthCalledWith(2, "workspace-1", { + afterCursor: "", + limit: 500, + }); + expect(mocks.client.eventPage).toHaveBeenNthCalledWith(3, "workspace-1", { + afterCursor: "cursor-500", + limit: 500, + }); + expect(mocks.client.eventPage).toHaveBeenNthCalledWith(4, "workspace-1", { + afterCursor: "cursor-501", + limit: 500, + }); + expect(mocks.handleClickClackInbound).toHaveBeenCalledTimes(1); + expect(mocks.client.websocket).toHaveBeenLastCalledWith("workspace-1", "cursor-501"); + + abort.abort(); + await run; + }); + it("skips malformed websocket frames without stopping the monitor", async () => { const socket = new FakeSocket(); mocks.client.websocket.mockReturnValue(socket); @@ -324,7 +397,7 @@ describe("ClickClack gateway", () => { }); it("clears running status when backlog polling fails", async () => { - mocks.client.events.mockRejectedValue(new Error("clickclack unavailable")); + mocks.client.eventPage.mockRejectedValue(new Error("clickclack unavailable")); const abort = new AbortController(); const ctx = createGatewayContext(abort.signal); diff --git a/extensions/clickclack/src/gateway.ts b/extensions/clickclack/src/gateway.ts index 766e737a1099..ed5d7ecebafa 100644 --- a/extensions/clickclack/src/gateway.ts +++ b/extensions/clickclack/src/gateway.ts @@ -17,6 +17,8 @@ import type { ResolvedClickClackAccount, } from "./types.js"; +const CLICKCLACK_EVENT_PAGE_LIMIT = 500; + function payloadString(event: ClickClackEvent, key: string): string { const value = event.payload?.[key]; return typeof value === "string" ? value : ""; @@ -132,6 +134,37 @@ async function processEvent(params: { }); } +async function drainEventBacklog(params: { + client: ReturnType; + workspaceId: string; + afterCursor: string; + abortSignal: AbortSignal; + onEvent: (event: ClickClackEvent) => Promise; +}): Promise { + let afterCursor = params.afterCursor; + while (!params.abortSignal.aborted) { + const page = await params.client.eventPage(params.workspaceId, { + afterCursor, + limit: CLICKCLACK_EVENT_PAGE_LIMIT, + }); + const events = page.events; + for (const event of events) { + if (params.abortSignal.aborted) { + return afterCursor; + } + if (!event.cursor || event.cursor === afterCursor) { + throw new Error("ClickClack event backlog returned a non-advancing cursor"); + } + await params.onEvent(event); + afterCursor = event.cursor; + } + if (events.length === 0) { + return afterCursor; + } + } + return afterCursor; +} + export async function startClickClackGatewayAccount( ctx: ChannelGatewayContext, ) { @@ -164,25 +197,39 @@ export async function startClickClackGatewayAccount( let initialized = false; try { while (!ctx.abortSignal.aborted) { - const backlog = await client.events(workspaceId, afterCursor); if (!initialized) { - // First pass establishes the cursor without replaying historical backlog - // into fresh gateway sessions. - for (const event of backlog) { - afterCursor = event.cursor || afterCursor; + const page = await client.eventPage(workspaceId, { includeTail: true }); + // Newer servers capture this cursor before listing the page, so events + // created during startup remain eligible for websocket delivery. + if (page.tailCursor !== undefined) { + afterCursor = page.tailCursor; + } else { + // Older servers omit tail_cursor; preserve the shipped one-page + // startup behavior instead of extending the history-skip window. + for (const event of page.events) { + afterCursor = event.cursor || afterCursor; + } } initialized = true; } else { - for (const event of backlog) { - afterCursor = event.cursor || afterCursor; - await processEvent({ - account, - config: ctx.cfg, - client, - event, - botUserId: account.botUserId, - }); - } + afterCursor = await drainEventBacklog({ + client, + workspaceId, + afterCursor, + abortSignal: ctx.abortSignal, + onEvent: async (event) => { + await processEvent({ + account, + config: ctx.cfg, + client, + event, + botUserId: account.botUserId, + }); + }, + }); + } + if (ctx.abortSignal.aborted) { + break; } const socket = client.websocket(workspaceId, afterCursor); await new Promise((resolve, reject) => { diff --git a/extensions/clickclack/src/http-client.test.ts b/extensions/clickclack/src/http-client.test.ts index 7aeecd123902..6a14569889b2 100644 --- a/extensions/clickclack/src/http-client.test.ts +++ b/extensions/clickclack/src/http-client.test.ts @@ -125,6 +125,35 @@ function streamedErrorResponse(body: string, limit: number) { } describe("ClickClack HTTP client", () => { + it("adds paged tail queries without changing the legacy events result", async () => { + const fetchMock = vi.fn(async () => Response.json({ events: [], tail_cursor: "cursor-900" })); + const client = createClickClackClient({ + baseUrl: "https://clickclack.example", + token: "test-token", + fetch: fetchMock, + }); + + const page = await client.eventPage("workspace-1", { + afterCursor: "cursor-500", + limit: 500, + includeTail: true, + }); + const legacyEvents = await client.events("workspace-1", "cursor-900"); + + expect(fetchMock).toHaveBeenNthCalledWith( + 1, + "https://clickclack.example/api/realtime/events?workspace_id=workspace-1&after_cursor=cursor-500&limit=500&include_tail=true", + expect.any(Object), + ); + expect(fetchMock).toHaveBeenNthCalledWith( + 2, + "https://clickclack.example/api/realtime/events?workspace_id=workspace-1&after_cursor=cursor-900", + expect.any(Object), + ); + expect(page).toEqual({ events: [], tailCursor: "cursor-900" }); + expect(legacyEvents).toEqual([]); + }); + it("sends only safe bounded request correlation", async () => { const fetchMock = vi.fn(async (_input: string | URL | Request, _init?: RequestInit) => Response.json({ user: { id: "usr_1" } }), diff --git a/extensions/clickclack/src/http-client.ts b/extensions/clickclack/src/http-client.ts index e77a2cbf6884..b762857e5790 100644 --- a/extensions/clickclack/src/http-client.ts +++ b/extensions/clickclack/src/http-client.ts @@ -100,6 +100,33 @@ export function createClickClackClient(options: ClientOptions) { }); } + async function fetchEventPage( + workspaceId: string, + pageOptions: { + afterCursor?: string; + limit?: number; + includeTail?: boolean; + } = {}, + ): Promise<{ events: ClickClackEvent[]; tailCursor?: string }> { + const query = new URLSearchParams({ workspace_id: workspaceId }); + if (pageOptions.afterCursor) { + query.set("after_cursor", pageOptions.afterCursor); + } + if (pageOptions.limit !== undefined) { + query.set("limit", String(pageOptions.limit)); + } + if (pageOptions.includeTail) { + query.set("include_tail", "true"); + } + const data = await request<{ events: ClickClackEvent[]; tail_cursor?: unknown }>( + `/api/realtime/events?${query.toString()}`, + ); + return { + events: data.events, + ...(typeof data.tail_cursor === "string" ? { tailCursor: data.tail_cursor } : {}), + }; + } + return { me: async (): Promise => { const data = await request<{ user: ClickClackUser }>("/api/me"); @@ -235,16 +262,9 @@ export function createClickClackClient(options: ClientOptions) { ); return data.message; }, - events: async (workspaceId: string, afterCursor?: string): Promise => { - const query = new URLSearchParams({ workspace_id: workspaceId }); - if (afterCursor) { - query.set("after_cursor", afterCursor); - } - const data = await request<{ events: ClickClackEvent[] }>( - `/api/realtime/events?${query.toString()}`, - ); - return data.events; - }, + events: async (workspaceId: string, afterCursor?: string): Promise => + (await fetchEventPage(workspaceId, { afterCursor })).events, + eventPage: fetchEventPage, websocket: (workspaceId: string, afterCursor?: string): WebSocket => { const url = new URL(`${baseUrl}/api/realtime/ws`); url.protocol = url.protocol === "https:" ? "wss:" : "ws:";