fix(slack): deduplicate message and mention events in one flush (#115302)

Co-authored-by: Peter Steinberger <steipete@gmail.com>

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<string, unknown>
+      >;
+      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<Record<string, unknown>>;
+    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();

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<string>();
+          const claimedKeys = new Map<string, number>();
           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;