diff --git a/extensions/slack/src/monitor/message-handler.test.ts b/extensions/slack/src/monitor/message-handler.test.ts index ba15e9cfb208..96d1bb3e504f 100644 --- a/extensions/slack/src/monitor/message-handler.test.ts +++ b/extensions/slack/src/monitor/message-handler.test.ts @@ -390,6 +390,95 @@ describe("createSlackMessageHandler", () => { expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); }); + it.each([ + ["message", "app_mention"], + ["app_mention", "message"], + ] as const)( + "deduplicates message/app_mention twins in one flush (%s before %s)", + async (firstSource, secondSource) => { + const { handler } = createHandlerWithTracker(); + const twinTs = firstSource === "message" ? "1709000000.001777" : "1709000000.001778"; + const message = { + type: "message" as const, + channel: "C111", + user: "U111", + ts: twinTs, + text: "<@UBOT> hello", + }; + const handleTwin = (source: "message" | "app_mention") => + handler(message as never, { + source, + awaitDispatch: true, + ...(source === "app_mention" ? { wasMentioned: true } : {}), + }); + + const first = handleTwin(firstSource); + const second = handleTwin(secondSource); + await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(2)); + + const entries = enqueueMock.mock.calls.map((call) => call[0]) as Array< + Record + >; + await onFlushCallbacks[0]?.(entries); + + await expect(Promise.all([first, second])).resolves.toEqual([undefined, undefined]); + expect(prepareSlackMessageMock).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + message: expect.objectContaining({ text: message.text, ts: twinTs }), + opts: expect.objectContaining({ source: "app_mention", wasMentioned: true }), + }), + ); + expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); + const prepared = dispatchPreparedSlackMessageMock.mock.calls[0]?.[0] as { + ctxPayload: { MessageSids?: string[] }; + }; + expect(prepared.ctxPayload.MessageSids).toBeUndefined(); + }, + ); + + it("preserves distinct messages and identities in the same debounced flush", async () => { + const { handler } = createHandlerWithTracker(); + const messages = [ + { ts: "1709000000.001779", text: "first message" }, + { ts: "1709000000.001780", text: "second message" }, + ] as const; + const handled = messages.map((message) => + handler( + { + type: "message", + channel: "D111", + user: "U111", + ...message, + } as never, + { source: "message", awaitDispatch: true }, + ), + ); + await vi.waitFor(() => expect(enqueueMock).toHaveBeenCalledTimes(2)); + + const entries = enqueueMock.mock.calls.map((call) => call[0]) as Array>; + await onFlushCallbacks[0]?.(entries); + + await expect(Promise.all(handled)).resolves.toEqual([undefined, undefined]); + expect(prepareSlackMessageMock).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + message: expect.objectContaining({ text: "first message\nsecond message" }), + }), + ); + expect(dispatchPreparedSlackMessageMock).toHaveBeenCalledTimes(1); + const prepared = dispatchPreparedSlackMessageMock.mock.calls[0]?.[0] as { + ctxPayload: { + MessageSids?: string[]; + MessageSidFirst?: string; + MessageSidLast?: string; + }; + }; + expect(prepared.ctxPayload).toMatchObject({ + MessageSids: [messages[0].ts, messages[1].ts], + MessageSidFirst: messages[0].ts, + MessageSidLast: messages[1].ts, + }); + }); + it("propagates debounced dispatch failures to relay delivery", async () => { dispatchPreparedSlackMessageMock.mockRejectedValueOnce(new Error("dispatch failed")); const { handler } = createHandlerWithTracker(); diff --git a/extensions/slack/src/monitor/message-handler.ts b/extensions/slack/src/monitor/message-handler.ts index ccfeadcd1e68..ed6c2f6b83c0 100644 --- a/extensions/slack/src/monitor/message-handler.ts +++ b/extensions/slack/src/monitor/message-handler.ts @@ -169,11 +169,12 @@ export function createSlackMessageHandler(params: { await (async () => { // Logical-identity claims: Slack sends message + app_mention twins with // distinct event_ids for one post, so the durable queue cannot dedupe - // them. Same-flush twins share one claim; a later twin claims duplicate - // and is dropped before it can produce a second visible reply. + // them. Same-flush twins share one claim and one logical message while + // retaining the latest event's routing and any earlier mention. const claims: SlackMessageDispatchReplayClaim[] = []; - const claimedKeys = new Set(); + const claimedKeys = new Map(); const surviving: typeof entries = []; + let latestSurviving: (typeof entries)[number] | undefined; for (const entry of entries) { const replayKey = buildSlackMessageDispatchReplayKey({ accountId: ctx.accountId, @@ -181,8 +182,26 @@ export function createSlackMessageHandler(params: { ts: entry.message.ts, teamId: entry.opts.eventScope?.teamId, }); - if (!replayKey || claimedKeys.has(replayKey)) { + if (!replayKey) { surviving.push(entry); + latestSurviving = entry; + continue; + } + const existingIndex = claimedKeys.get(replayKey); + if (existingIndex !== undefined) { + const existing = surviving[existingIndex]; + const merged = { + ...entry, + opts: { + ...entry.opts, + ...(existing?.opts.source === "app_mention" + ? { source: "app_mention" as const } + : {}), + ...(existing?.opts.wasMentioned ? { wasMentioned: true } : {}), + }, + }; + surviving[existingIndex] = merged; + latestSurviving = merged; continue; } const claim = await claimSlackMessageDispatchReplay({ @@ -191,8 +210,9 @@ export function createSlackMessageHandler(params: { }); if (claim.kind === "claimed") { claims.push(claim.handle); - claimedKeys.add(replayKey); + claimedKeys.set(replayKey, surviving.length); surviving.push(entry); + latestSurviving = entry; } } const releaseClaims = (error?: unknown) => { @@ -205,7 +225,7 @@ export function createSlackMessageHandler(params: { await handle.commit(); } }; - const last = surviving.at(-1); + const last = latestSurviving; if (!last) { releaseClaims(); return;