diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index 2d06a41..ec15a39 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -151,14 +151,24 @@ function createDeps(overrides: { where, data, }: { - where: { id?: string; providerHandle?: string; status?: string }; + where: { + id?: string; + providerHandle?: string; + status?: string | { in: string[] }; + }; data: Record; }) => { let count = 0; for (const row of rows) { if (where.id && row.id !== where.id) continue; if (where.providerHandle && row.providerHandle !== where.providerHandle) continue; - if (where.status && row.status !== where.status) continue; + if (where.status) { + const allowed = + typeof where.status === "string" + ? [where.status] + : where.status.in; + if (!allowed.includes(String(row.status))) continue; + } Object.assign(row, data); count += 1; } @@ -540,7 +550,7 @@ describe("deliverMessagingOutbound", () => { }); describe("applyMessagingOutboundStatus", () => { - it("maps terminal statuses onto outbox rows by handle", async () => { + it("maps terminal statuses onto pending outbox rows by handle", async () => { const deps = createDeps({}); await applyMessagingOutboundStatus(deps.prisma, { type: "status", @@ -549,7 +559,7 @@ describe("applyMessagingOutboundStatus", () => { status: "ERROR", }); expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ - where: { providerHandle: "h-1" }, + where: { providerHandle: "h-1", status: { in: ["pending"] } }, data: { status: "failed" }, }); @@ -560,7 +570,7 @@ describe("applyMessagingOutboundStatus", () => { status: "DELIVERED", }); expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ - where: { providerHandle: "h-2" }, + where: { providerHandle: "h-2", status: { in: ["pending", "sent"] } }, data: { status: "sent" }, }); @@ -572,6 +582,62 @@ describe("applyMessagingOutboundStatus", () => { }); expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledTimes(2); }); + + it("does not regress a sent row when a late ERROR webhook arrives", async () => { + const deps = createDeps({ + outboundRows: [ + { + id: "out-delivered", + idempotencyKey: "dm:1", + kind: "dm", + identityId: "mi-1", + body: "already delivered", + status: "sent", + providerHandle: "h-late-error", + }, + ], + }); + await applyMessagingOutboundStatus(deps.prisma, { + type: "status", + provider: "sendblue", + handle: "h-late-error", + status: "ERROR", + }); + + expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" })); + expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ + where: { providerHandle: "h-late-error", status: { in: ["pending"] } }, + data: { status: "failed" }, + }); + }); + + it("does not upgrade a failed row when a late DELIVERED webhook arrives", async () => { + const deps = createDeps({ + outboundRows: [ + { + id: "out-failed", + idempotencyKey: "dm:2", + kind: "dm", + identityId: "mi-1", + body: "gave up", + status: "failed", + providerHandle: "h-late-sent", + }, + ], + }); + await applyMessagingOutboundStatus(deps.prisma, { + type: "status", + provider: "sendblue", + handle: "h-late-sent", + status: "DELIVERED", + }); + + expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "failed" })); + expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ + where: { providerHandle: "h-late-sent", status: { in: ["pending", "sent"] } }, + data: { status: "sent" }, + }); + }); }); function createChannelDeps( diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index 3429d5e..2ed3a48 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -468,20 +468,32 @@ function messagingOutboundRetryDelayMs(attempts: number): number { return Math.min(2 ** attempts * 1000, 60_000); } +function outboundStatusFromWebhook(event: MessagingOutboundStatus): "failed" | "sent" | null { + if (event.status === "ERROR" || event.status === "DECLINED") return "failed"; + if (event.status === "SENT" || event.status === "DELIVERED") return "sent"; + return null; +} + +/** Stored statuses that may transition to the webhook target. */ +function allowedSourcesForWebhookTarget(target: "failed" | "sent"): Array<"pending" | "sent" | "failed"> { + // Drain claims rows `sent` before the provider acks; a late ERROR must not + // regress a delivered row. Terminal `failed` must not be upgraded by SENT. + if (target === "failed") return ["pending"]; + return ["pending", "sent"]; +} + /** Outbound status webhooks update outbox rows by provider handle. */ export async function applyMessagingOutboundStatus( prisma: PrismaClient, event: MessagingOutboundStatus, ): Promise { - const status = - event.status === "ERROR" || event.status === "DECLINED" - ? "failed" - : event.status === "SENT" || event.status === "DELIVERED" - ? "sent" - : null; + const status = outboundStatusFromWebhook(event); if (!status) return; await prisma.messagingOutbound.updateMany({ - where: { providerHandle: event.handle }, + where: { + providerHandle: event.handle, + status: { in: allowedSourcesForWebhookTarget(status) }, + }, data: { status }, }); }