fix(messaging): monotonic outbound status webhooks
Drain claims outbox rows as sent before the provider acks. Late ERROR webhooks were regressing delivered rows to failed; late DELIVERED could upgrade exhausted failures. Gate webhook updates by allowed source status. Co-authored-by: cynarAI <cynarAI@users.noreply.github.com>
This commit is contained in:
parent
a38cfb437f
commit
bf65df8603
@ -151,14 +151,24 @@ function createDeps(overrides: {
|
|||||||
where,
|
where,
|
||||||
data,
|
data,
|
||||||
}: {
|
}: {
|
||||||
where: { id?: string; providerHandle?: string; status?: string };
|
where: {
|
||||||
|
id?: string;
|
||||||
|
providerHandle?: string;
|
||||||
|
status?: string | { in: string[] };
|
||||||
|
};
|
||||||
data: Record<string, unknown>;
|
data: Record<string, unknown>;
|
||||||
}) => {
|
}) => {
|
||||||
let count = 0;
|
let count = 0;
|
||||||
for (const row of rows) {
|
for (const row of rows) {
|
||||||
if (where.id && row.id !== where.id) continue;
|
if (where.id && row.id !== where.id) continue;
|
||||||
if (where.providerHandle && row.providerHandle !== where.providerHandle) 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);
|
Object.assign(row, data);
|
||||||
count += 1;
|
count += 1;
|
||||||
}
|
}
|
||||||
@ -540,7 +550,7 @@ describe("deliverMessagingOutbound", () => {
|
|||||||
});
|
});
|
||||||
|
|
||||||
describe("applyMessagingOutboundStatus", () => {
|
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({});
|
const deps = createDeps({});
|
||||||
await applyMessagingOutboundStatus(deps.prisma, {
|
await applyMessagingOutboundStatus(deps.prisma, {
|
||||||
type: "status",
|
type: "status",
|
||||||
@ -549,7 +559,7 @@ describe("applyMessagingOutboundStatus", () => {
|
|||||||
status: "ERROR",
|
status: "ERROR",
|
||||||
});
|
});
|
||||||
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
|
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
|
||||||
where: { providerHandle: "h-1" },
|
where: { providerHandle: "h-1", status: { in: ["pending"] } },
|
||||||
data: { status: "failed" },
|
data: { status: "failed" },
|
||||||
});
|
});
|
||||||
|
|
||||||
@ -560,7 +570,7 @@ describe("applyMessagingOutboundStatus", () => {
|
|||||||
status: "DELIVERED",
|
status: "DELIVERED",
|
||||||
});
|
});
|
||||||
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
|
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
|
||||||
where: { providerHandle: "h-2" },
|
where: { providerHandle: "h-2", status: { in: ["pending", "sent"] } },
|
||||||
data: { status: "sent" },
|
data: { status: "sent" },
|
||||||
});
|
});
|
||||||
|
|
||||||
@ -572,6 +582,62 @@ describe("applyMessagingOutboundStatus", () => {
|
|||||||
});
|
});
|
||||||
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledTimes(2);
|
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(
|
function createChannelDeps(
|
||||||
|
|||||||
@ -468,20 +468,32 @@ function messagingOutboundRetryDelayMs(attempts: number): number {
|
|||||||
return Math.min(2 ** attempts * 1000, 60_000);
|
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. */
|
/** Outbound status webhooks update outbox rows by provider handle. */
|
||||||
export async function applyMessagingOutboundStatus(
|
export async function applyMessagingOutboundStatus(
|
||||||
prisma: PrismaClient,
|
prisma: PrismaClient,
|
||||||
event: MessagingOutboundStatus,
|
event: MessagingOutboundStatus,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
const status =
|
const status = outboundStatusFromWebhook(event);
|
||||||
event.status === "ERROR" || event.status === "DECLINED"
|
|
||||||
? "failed"
|
|
||||||
: event.status === "SENT" || event.status === "DELIVERED"
|
|
||||||
? "sent"
|
|
||||||
: null;
|
|
||||||
if (!status) return;
|
if (!status) return;
|
||||||
await prisma.messagingOutbound.updateMany({
|
await prisma.messagingOutbound.updateMany({
|
||||||
where: { providerHandle: event.handle },
|
where: {
|
||||||
|
providerHandle: event.handle,
|
||||||
|
status: { in: allowedSourcesForWebhookTarget(status) },
|
||||||
|
},
|
||||||
data: { status },
|
data: { status },
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user