fix(zalo): stop duplicating inbox-sent messages and handle postback/receipt webhooks (#1012)

* fix(zalo): dedup inbox message echoes and ack delivery receipts

Two webhook-driven bugs around inbox-sent messages:

- The Send API response was mistyped as flat, so message_id was discarded
  and sendMessage/sendFlowStep returned messageIds: []. The worker's
  sourceId backfill no-oped, the oa_send_* echo missed findBySourceId,
  and every inbox-sent message appeared twice in the conversation.
  Re-shape the schema to the real { data, error, message } envelope and
  return the provider ids so the existing dedup path works.
- user_received_message delivery receipts carry no content; routing them
  to incomingMessage made the handler throw "No content found" and the
  job retry forever. Ack and drop them at the webhook router.

* fix(zalo): show tapped flow button label instead of raw postback payload

Zalo (and Telegram) deliver a button tap as a plain text message whose
body is the encoded payload (postback_<flowId>::<buttonId>) with no
button title alongside it, so the inbox displayed the raw payload.
Resolve the tapped button's label from the flow version before saving;
fall back to the raw text when the flow can no longer be resolved so a
stale tap never fails the message save.

---------

Co-authored-by: Deathgiver <anonymous@users.noreply.github.com>

Author: sung17

apps/worker/__tests__/received-message.test.ts

@@ -38,6 +38,7 @@ const {
   mockChatQueueAdd,
   mockSyncScopedIdentity,
   mockIsUniqueViolationError,
+  mockDetectFlowVersion,
 } = vi.hoisted(() => {
   const mockDbSet = vi.fn()
   const updateChain = { set: mockDbSet, where: vi.fn() }
@@ -111,6 +112,7 @@ const {
       }),
     ),
     mockIsUniqueViolationError: vi.fn().mockReturnValue(false),
+    mockDetectFlowVersion: vi.fn(),
   }
 })
 
@@ -289,6 +291,10 @@ vi.mock("../src/lib/logger", () => ({
   logger: { error: vi.fn(), warn: vi.fn(), info: vi.fn(), debug: vi.fn() },
 }))
 
+vi.mock("../src/lib/db", () => ({
+  detectFlowVersion: mockDetectFlowVersion,
+}))
+
 vi.mock("../src/services/integrations", () => ({
   allIntegrations: {
     messenger: {
@@ -801,6 +807,110 @@ describe("receiveMessage — message repository branch", () => {
     })
   })
 
+  test("replaces a raw postback payload echo with the flow button label", async () => {
+    const postbackAction = encodeButtonPayload({ flowId: "42", buttonId: "77" })
+    mockDetectFlowVersion.mockResolvedValue({
+      flowVersion: {
+        id: "fv-1",
+        nodes: [
+          {
+            id: "node-1",
+            data: {
+              details: {
+                steps: [
+                  {
+                    buttons: [
+                      {
+                        id: "77",
+                        label: "Xem sản phẩm",
+                        buttonType: "nextStep",
+                        beforeStep: null,
+                        steps: [],
+                      },
+                    ],
+                  },
+                ],
+              },
+            },
+          },
+        ],
+      },
+      useLatestFlowVersion: true,
+    })
+    mockRunChannelHandler.mockResolvedValue({
+      message: {
+        ...baseIncomingMessage,
+        text: `postback_${postbackAction}`,
+        attachments: [],
+      },
+      contact: { sourceId: "psid-123", firstName: "Test" },
+      postbackAction,
+      quickReplyAction: null,
+      ref: null,
+    })
+
+    await receiveMessage({ ...baseProps, integrationType: "zalo" })
+
+    expect(mockDetectFlowVersion).toHaveBeenCalledWith({
+      flowId: "42",
+      flowVersionId: undefined,
+      workspaceId: "ws-1",
+    })
+    expect(mockCreateOrUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({ text: "Xem sản phẩm" }),
+    )
+    expect(mockUpdateTracking).toHaveBeenCalledWith(
+      expect.objectContaining({
+        data: expect.objectContaining({ lastBtnTitle: "Xem sản phẩm" }),
+      }),
+    )
+    expect(mockAutomatedResponseEnqueueFlowAction).toHaveBeenCalledWith({
+      kind: "postback",
+      data: expect.objectContaining({ action: postbackAction }),
+    })
+  })
+
+  test("keeps the raw postback text when the flow can no longer be resolved", async () => {
+    const postbackAction = encodeButtonPayload({ flowId: "42", buttonId: "77" })
+    mockDetectFlowVersion.mockRejectedValue(new Error("FlowVersion not found"))
+    mockRunChannelHandler.mockResolvedValue({
+      message: {
+        ...baseIncomingMessage,
+        text: `postback_${postbackAction}`,
+        attachments: [],
+      },
+      contact: { sourceId: "psid-123", firstName: "Test" },
+      postbackAction,
+      quickReplyAction: null,
+      ref: null,
+    })
+
+    await receiveMessage({ ...baseProps, integrationType: "zalo" })
+
+    expect(mockCreateOrUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({ text: `postback_${postbackAction}` }),
+    )
+  })
+
+  test("does not rewrite text when the channel already supplies a button title", async () => {
+    const postbackAction = encodeButtonPayload({ flowId: "42", buttonId: "77" })
+    mockRunChannelHandler.mockResolvedValue({
+      message: { ...baseIncomingMessage, attachments: [] },
+      contact: { sourceId: "psid-123", firstName: "Test" },
+      postbackAction,
+      quickReplyAction: null,
+      buttonTitle: "Nút Messenger",
+      ref: null,
+    })
+
+    await receiveMessage(baseProps)
+
+    expect(mockDetectFlowVersion).not.toHaveBeenCalled()
+    expect(mockCreateOrUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({ text: "hello" }),
+    )
+  })
+
   test("sends Vietnamese feedback instead of going silent when an appointment cancel token is invalid", async () => {
     mockParseAppointmentCancelPostback.mockReturnValue("cancel-token")
     mockVerifyAppointmentCancelPostback.mockRejectedValue(

apps/worker/src/integration/handlers/flow-action.ts

@@ -1,4 +1,12 @@
-import { decodeButtonPayload } from "@chatbotx.io/flow-config"
+import {
+  decodeButtonPayload,
+  type FlowNode,
+  resolveFlowActionTarget,
+  ZALO_POSTBACK_TEXT_PREFIX,
+} from "@chatbotx.io/flow-config"
+import type { IncomingMessage } from "@chatbotx.io/sdk"
+import { normalizeError } from "universal-error-normalizer"
+import { detectFlowVersion } from "../../lib/db"
 import { logger } from "../../lib/logger"
 
 export type FlowActionKind = "postback" | "quickReply"
@@ -36,3 +44,58 @@ export const sanitizeFlowAction = (
 
   return null
 }
+
+/**
+ * Zalo and Telegram deliver a button tap as a plain text message whose body is
+ * the encoded postback payload, with no button title alongside it — unlike
+ * Messenger/WhatsApp, which send the title. The echo formats differ: Zalo
+ * prefixes the payload (`postback_123::456`), Telegram sends it bare
+ * (`123::456`). Resolve the tapped button's label from the flow so the inbox
+ * shows what the contact actually pressed instead of the raw payload.
+ *
+ * Returns null (leaving the raw text untouched) whenever the channel already
+ * provided a title, the text is not the bare payload echo, or the flow/button
+ * can no longer be resolved — a stale tap must never fail the message save.
+ */
+export const resolvePostbackButtonLabel = async (props: {
+  postbackAction: string | null
+  buttonTitle: string | null | undefined
+  message: IncomingMessage | null
+  workspaceId: string
+}): Promise<string | null> => {
+  const { postbackAction, buttonTitle, message, workspaceId } = props
+  if (!postbackAction || buttonTitle || !message?.text) {
+    return null
+  }
+
+  const isRawPayloadEcho =
+    message.text === postbackAction ||
+    message.text === `${ZALO_POSTBACK_TEXT_PREFIX}${postbackAction}`
+  if (!isRawPayloadEcho) {
+    return null
+  }
+
+  const parsedAction = decodeButtonPayload(postbackAction)
+  if (!parsedAction?.buttonId) {
+    return null
+  }
+
+  try {
+    const { flowVersion } = await detectFlowVersion({
+      flowId: parsedAction.flowId,
+      flowVersionId: parsedAction.flowVersionId,
+      workspaceId,
+    })
+    const target = resolveFlowActionTarget(
+      flowVersion.nodes as unknown as FlowNode[],
+      parsedAction.buttonId,
+    )
+    return target?.details.label ?? null
+  } catch (error) {
+    logger.warn(
+      { error: normalizeError(error), postbackAction, workspaceId },
+      "Could not resolve button label for postback message",
+    )
+    return null
+  }
+}

apps/worker/src/integration/handlers/received-message.ts

@@ -85,7 +85,7 @@ import {
   integrationService,
   isInstagramViaFacebook,
 } from "../../services/integrations"
-import { sanitizeFlowAction } from "./flow-action"
+import { resolvePostbackButtonLabel, sanitizeFlowAction } from "./flow-action"
 
 type ContactInboxTracking = ContactInboxTrackingData
 
@@ -249,25 +249,40 @@ export const receiveMessage = async (
     integrationIdentifier,
   })
 
-  const detected = await detectContactAndConversation({
-    incomingContact,
-    inbox,
-    integrationRow,
-    source:
-      metaReferralToContactSource(referralSource) ??
-      contactSources.enum.inboundMessage,
-  })
+  // Label resolution only reads the raw text (direction correction never
+  // changes it) and the workspace, so it can overlap the contact lookup.
+  const [detected, postbackButtonLabel] = await Promise.all([
+    detectContactAndConversation({
+      incomingContact,
+      inbox,
+      integrationRow,
+      source:
+        metaReferralToContactSource(referralSource) ??
+        contactSources.enum.inboundMessage,
+    }),
+    resolvePostbackButtonLabel({
+      postbackAction,
+      buttonTitle: parsedMessage.buttonTitle,
+      message: rawIncomingMessage,
+      workspaceId: inbox.workspaceId,
+    }),
+  ])
   if (!detected) {
     throw new SdkException("Unable to resolve contact and conversation")
   }
   const { contactInbox, conversation, contact, isNewContact } = detected
-  const incomingMessage = correctStoryReplyDirectionForNewContact(
+  const directedIncomingMessage = correctStoryReplyDirectionForNewContact(
     rawIncomingMessage,
     isNewContact,
   )
+  const incomingMessage =
+    postbackButtonLabel && directedIncomingMessage
+      ? { ...directedIncomingMessage, text: postbackButtonLabel }
+      : directedIncomingMessage
   const systemFieldUpdates = getReceivedMessageSystemFieldUpdates({
-    ...parsedMessage,
+    buttonTitle: parsedMessage.buttonTitle || postbackButtonLabel,
     message: incomingMessage,
+    referral: parsedMessage.referral,
   })
 
   // Overwrite Contact.phoneNumber/email from message text — every inbound

integrations/zalo/__tests__/outgoing-message-ids.test.ts

@@ -0,0 +1,97 @@
+import { beforeEach, describe, expect, test, vi } from "vitest"
+
+const { mockSendMessageToZaloOA } = vi.hoisted(() => ({
+  mockSendMessageToZaloOA: vi.fn(),
+}))
+
+vi.mock("../src/api/message", () => ({
+  sendMessageToZaloOA: mockSendMessageToZaloOA,
+  uploadAttachment: vi.fn(),
+}))
+
+vi.mock("../src/lib/logger", () => ({
+  logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
+}))
+
+const { sendMessage, sendFlowStep } = await import(
+  "../src/handlers/message/outgoing-message"
+)
+
+const ctx = {
+  auth: { tokens: { accessToken: "tok" } },
+} as never
+
+const contact = {
+  id: "contact-1",
+  sourceId: "zalo-uid-1",
+} as never
+
+describe("zalo outgoing handlers return provider message ids", () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockSendMessageToZaloOA.mockResolvedValue({
+      error: 0,
+      message: "Success",
+      data: { message_id: "m_provider-1" },
+    })
+  })
+
+  test("sendMessage returns the Send API message_id so the worker can persist sourceId", async () => {
+    const result = await sendMessage({
+      ctx,
+      data: {
+        contact,
+        message: {
+          id: "msg-1",
+          contentType: "text",
+          messageType: "outgoing",
+          text: "hello",
+        },
+      },
+    } as never)
+
+    expect(mockSendMessageToZaloOA).toHaveBeenCalledTimes(1)
+    expect(result).toEqual({ messageIds: ["m_provider-1"] })
+  })
+
+  test("sendFlowStep (sendText) returns the Send API message_id", async () => {
+    const result = await sendFlowStep({
+      ctx,
+      data: {
+        contact,
+        flowId: "flow-1",
+        step: {
+          id: "step-1",
+          stepType: "sendText",
+          text: "automated reply",
+          buttons: [],
+        },
+      },
+    } as never)
+
+    expect(mockSendMessageToZaloOA).toHaveBeenCalledTimes(1)
+    expect(result).toEqual({ messageIds: ["m_provider-1"] })
+  })
+
+  test("sendMessage tolerates a response without data.message_id", async () => {
+    mockSendMessageToZaloOA.mockResolvedValue({
+      error: 0,
+      message: "Success",
+    })
+
+    const result = await sendMessage({
+      ctx,
+      data: {
+        contact,
+        message: {
+          id: "msg-1",
+          contentType: "text",
+          messageType: "outgoing",
+          text: "hello",
+        },
+      },
+    } as never)
+
+    expect(result).toEqual({ messageIds: [] })
+  })
+})