From 696a37ac0bf4474d63fb5027e923310bc6d34c1a Mon Sep 17 00:00:00 2001 From: Ingo Wagner Date: Tue, 22 Sep 2026 06:22:03 +0000 Subject: [PATCH] fix(adapters): mirror group completion after in-app approval --- .../adapters/src/messaging-delivery.test.ts | 344 +++++------------- packages/adapters/src/messaging-delivery.ts | 291 ++++++--------- 2 files changed, 192 insertions(+), 443 deletions(-) diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index 0952113..fb55744 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -151,21 +151,14 @@ function createDeps(overrides: { where, data, }: { - where: { - id?: string; - providerHandle?: string; - status?: string | { in: string[] }; - }; + where: { id?: string; providerHandle?: string; status?: 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) { - const allowed = typeof where.status === "string" ? [where.status] : where.status.in; - if (!allowed.includes(String(row.status))) continue; - } + if (where.status && row.status !== where.status) continue; Object.assign(row, data); count += 1; } @@ -197,7 +190,7 @@ describe("deliverMessagingOutbound", () => { await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" }, + { threadId: "sendblue:dm-1", body: "Hello from your bot" }, context, ); expect(deps.rows).toEqual([ @@ -213,17 +206,8 @@ describe("deliverMessagingOutbound", () => { expect(deps.prisma.messagingIdentity.update).toHaveBeenCalledWith( expect.objectContaining({ data: { outboundSinceInbound: { increment: 1 } } }), ); - expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); - }); - - it("marks a completed messaging DM run after mirroring", async () => { - const deps = createDeps({ - run: { ...messagingRun, status: "completed" }, - }); - await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({ - where: { id: "run-1", trigger: "messaging", status: "completed" }, + where: { id: "run-1", trigger: "messaging" }, data: { messagingMirroredAt: expect.any(Date) }, }); }); @@ -246,20 +230,12 @@ describe("deliverMessagingOutbound", () => { }); expect(deps.sendToThread).toHaveBeenNthCalledWith( 1, - { - threadId: "sendblue:opened:+15551234567", - body: "first", - idempotencyKey: "msg:m-1", - }, + { threadId: "sendblue:opened:+15551234567", body: "first" }, context, ); expect(deps.sendToThread).toHaveBeenNthCalledWith( 2, - { - threadId: "sendblue:opened:+15551234567", - body: "second", - idempotencyKey: "msg:m-2", - }, + { threadId: "sendblue:opened:+15551234567", body: "second" }, context, ); }); @@ -288,7 +264,7 @@ describe("deliverMessagingOutbound", () => { const deps = createDeps({ run: { ...messagingRun, trigger: "bot_message" } }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" }, + { threadId: "sendblue:dm-1", body: "Hello from your bot" }, context, ); }); @@ -322,16 +298,22 @@ describe("deliverMessagingOutbound", () => { ]); }); - it("does not re-queue a mirrored DM when the provider send fails", async () => { + it("returns a transient send failure to pending and schedules a delayed retry", async () => { const deps = createDeps({ sendError: new Error("messaging provider 500") }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - // At-most-once: keep the claim rather than risk a duplicate DM. - expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "dm", status: "sent" })); + expect(deps.rows).toEqual([ + expect.objectContaining({ + kind: "dm", + status: "pending", + attempts: 1, + nextAttemptAt: expect.any(Date), + }), + ]); expect(deps.rows[0]!.providerHandle ?? null).toBeNull(); - expect(deps.rows[0]!.attempts ?? 0).toBe(0); - expect(deps.jobs.enqueue).not.toHaveBeenCalled(); - expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled(); + expect(deps.jobs.enqueue).toHaveBeenCalledWith( + expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }), + ); }); it("skips pending rows whose nextAttemptAt is still in the future", async () => { @@ -375,11 +357,7 @@ describe("deliverMessagingOutbound", () => { await deliverMessagingOutbound(deps, {}, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { - threadId: "sendblue:dm-1", - body: "pending earlier", - idempotencyKey: "invite:ch-1:mi-1", - }, + { threadId: "sendblue:dm-1", body: "pending earlier" }, context, ); expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" })); @@ -534,28 +512,6 @@ describe("deliverMessagingOutbound", () => { expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled(); }); - it("does not re-queue a channel invite when the provider send fails", async () => { - const deps = createDeps({ - run: null, - sendError: new Error("provider down"), - outboundRows: [ - { - id: "out-invite", - idempotencyKey: "invite:ch-1:mi-1", - kind: "dm", - identityId: "mi-1", - body: "join the group", - status: "pending", - providerHandle: null, - }, - ], - }); - await deliverMessagingOutbound(deps, {}, context); - - expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" })); - expect(deps.jobs.enqueue).not.toHaveBeenCalled(); - }); - it("does not re-queue a connect invite after a delivery transaction timeout", async () => { const deps = createDeps({ run: null, @@ -584,7 +540,7 @@ describe("deliverMessagingOutbound", () => { }); describe("applyMessagingOutboundStatus", () => { - it("maps terminal statuses onto pending outbox rows by handle", async () => { + it("maps terminal statuses onto outbox rows by handle", async () => { const deps = createDeps({}); await applyMessagingOutboundStatus(deps.prisma, { type: "status", @@ -593,7 +549,7 @@ describe("applyMessagingOutboundStatus", () => { status: "ERROR", }); expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ - where: { providerHandle: "h-1", status: { in: ["pending"] } }, + where: { providerHandle: "h-1" }, data: { status: "failed" }, }); @@ -604,7 +560,7 @@ describe("applyMessagingOutboundStatus", () => { status: "DELIVERED", }); expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ - where: { providerHandle: "h-2", status: { in: ["pending", "sent"] } }, + where: { providerHandle: "h-2" }, data: { status: "sent" }, }); @@ -616,62 +572,6 @@ 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( @@ -799,9 +699,13 @@ function createChannelDeps( }); }), createMany: vi.fn(async ({ data }: { data: Array> }) => { - for (const item of data) + let count = 0; + for (const item of data) { + if (rows.some((row) => row.idempotencyKey === item.idempotencyKey)) continue; rows.push({ id: `out-${rows.length + 1}`, status: "pending", ...item }); - return { count: data.length }; + count += 1; + } + return { count }; }), update: vi.fn(async ({ where, data }: { where: { id: string }; data: unknown }) => { const row = rows.find((candidate) => candidate.id === where.id); @@ -849,8 +753,8 @@ function createChannelDeps( }; } -describe("mirrorMessagingOutbound recovery marker", () => { - it("does not mark messagingMirroredAt when a completed channel run still has mirrorable text", async () => { +describe("deliverMessagingOutbound channel runs", () => { + it("skips answered ask messages but still mirrors text replies on the same run", async () => { const deps = createChannelDeps({ messages: [ { @@ -860,80 +764,55 @@ describe("mirrorMessagingOutbound recovery marker", () => { { id: "reply-1", blocks: [{ kind: "text", text: "Done — calendar updated." }] }, ], }); - const baseRun = await deps.prisma.run.findUnique(); - deps.prisma.run.findUnique = vi.fn(async () => ({ ...baseRun, status: "completed" })); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - - expect(deps.rows).toEqual([]); - expect(deps.sendToThread).not.toHaveBeenCalled(); - expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); + expect(deps.sendToThread).toHaveBeenCalledWith( + { threadId: "sendblue:grp-1", body: "Alice's agent: Done — calendar updated." }, + context, + ); + expect(deps.rows).toEqual([ + expect.objectContaining({ + idempotencyKey: "msg:reply-1", + kind: "group", + body: "Alice's agent: Done — calendar updated.", + }), + ]); + expect(deps.rows.some((row) => String(row.body).includes("Private detail"))).toBe(false); }); - it("does not mark messagingMirroredAt when bot messages are not durable yet", async () => { - const deps = createDeps({ - messages: [], - run: { ...messagingRun, status: "completed" }, - }); + it("mirrors completion text after in-app approval on a resumed group run", async () => { + let messages = [ + { id: "ask-1", blocks: [{ kind: "ask", text: "Approve?", status: "pending" }] }, + { id: "interim-1", blocks: [{ kind: "text", text: "Working on it…" }] }, + ]; + const deps = createChannelDeps(); + deps.prisma.message.findMany = vi.fn(async () => messages); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); + expect(deps.sendToThread).toHaveBeenCalledWith( + { threadId: "sendblue:grp-1", body: "Alice's agent: Working on it…" }, + context, + ); - expect(deps.rows).toEqual([]); - expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); - }); - - it("marks a completed messaging run after a successful mirror", async () => { - const deps = createDeps({ - run: { ...messagingRun, status: "completed" }, - }); + messages = [ + { + id: "ask-1", + blocks: [{ kind: "ask", text: "Approve?", status: "answered", answer: "yes" }], + }, + { id: "interim-1", blocks: [{ kind: "text", text: "Working on it…" }] }, + { id: "done-1", blocks: [{ kind: "text", text: "All set." }] }, + ]; + deps.sendToThread.mockClear(); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - - expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({ - where: { id: "run-1", trigger: "messaging", status: "completed" }, - data: { messagingMirroredAt: expect.any(Date) }, - }); - }); - - it("marks a completed noop when bot messages have no mirrorable text", async () => { - const deps = createDeps({ - run: { ...messagingRun, status: "completed" }, - messages: [{ id: "m-1", blocks: [{ kind: "tool_call", name: "search" }] }], - }); - await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - - expect(deps.sendToThread).not.toHaveBeenCalled(); - expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({ - where: { id: "run-1", trigger: "messaging", status: "completed" }, - data: { messagingMirroredAt: expect.any(Date) }, - }); - }); - - it("does not mark an in-progress messaging run even after a successful mirror", async () => { - const deps = createDeps({ - run: { ...messagingRun, status: "running" }, - }); - await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - - expect(deps.sendToThread).toHaveBeenCalled(); - expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); - }); -}); - -describe("deliverMessagingOutbound channel runs", () => { - it("keeps privately answered runs out of the group and peer fan-out on retries", async () => { - const deps = createChannelDeps({ - messages: [ - { - id: "ask-1", - blocks: [{ kind: "ask", text: "Details?", status: "answered", answer: "Private detail" }], - }, - { id: "reply-1", blocks: [{ kind: "text", text: "Reply containing private detail" }] }, - ], - }); - await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.rows).toEqual([]); - expect(deps.sendToThread).not.toHaveBeenCalled(); - expect(deps.contextMessages).toEqual([]); - expect(deps.sendUserMessage).not.toHaveBeenCalled(); + expect(deps.sendToThread).toHaveBeenCalledTimes(1); + expect(deps.sendToThread).toHaveBeenCalledWith( + { threadId: "sendblue:grp-1", body: "Alice's agent: All set." }, + context, + ); + expect(deps.rows).toEqual( + expect.arrayContaining([ + expect.objectContaining({ idempotencyKey: "msg:interim-1" }), + expect.objectContaining({ idempotencyKey: "msg:done-1" }), + ]), + ); }); it("can still publish a group reply alongside an unanswered ask", async () => { @@ -945,11 +824,7 @@ describe("deliverMessagingOutbound channel runs", () => { }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { - threadId: "sendblue:grp-1", - body: "Alice's agent: Public update", - idempotencyKey: "msg:reply-1", - }, + { threadId: "sendblue:grp-1", body: "Alice's agent: Public update" }, context, ); }); @@ -959,11 +834,7 @@ describe("deliverMessagingOutbound channel runs", () => { await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledWith( - { - threadId: "sendblue:grp-1", - body: "Alice's agent: found it", - idempotencyKey: "msg:m-1", - }, + { threadId: "sendblue:grp-1", body: "Alice's agent: found it" }, context, ); expect(deps.rows).toEqual([ @@ -1062,16 +933,6 @@ describe("deliverMessagingOutbound channel runs", () => { ); }); - it("does not re-queue a group post when the provider send fails", async () => { - const deps = createChannelDeps(); - deps.sendToThread.mockRejectedValueOnce(new Error("provider down")); - await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - - expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "group", status: "sent" })); - expect(deps.rows[0]!.providerHandle ?? null).toBeNull(); - expect(deps.enqueue).not.toHaveBeenCalled(); - }); - it("does not wake anyone when the hop budget is exhausted", async () => { const deps = createChannelDeps({ text: "@Helper again?", sourceHop: 6 }); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); @@ -1085,61 +946,26 @@ describe("deliverMessagingOutbound channel runs", () => { }); }); -describe("deliverMessagingOutbound pre-send failure retry", () => { - it("returns a thread-resolution failure to pending and schedules a delayed retry", async () => { +describe("deliverMessagingOutbound transient failure retry", () => { + it("marks the row failed only once the attempt budget is exhausted", async () => { const deps = createDeps({ run: null, - identity: { dmThreadId: null }, + sendError: new Error("messaging provider 500"), outboundRows: [ { id: "out-9", - idempotencyKey: "invite:ch-1:mi-1", + idempotencyKey: "msg:m-9", kind: "dm", identityId: "mi-1", - body: "needs thread lookup", - status: "pending", - providerHandle: null, - attempts: 0, - }, - ], - }); - deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503")); - await deliverMessagingOutbound(deps, {}, context); - - expect(deps.sendToThread).not.toHaveBeenCalled(); - expect(deps.rows).toEqual([ - expect.objectContaining({ - status: "pending", - attempts: 1, - nextAttemptAt: expect.any(Date), - }), - ]); - expect(deps.jobs.enqueue).toHaveBeenCalledWith( - expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }), - ); - }); - - it("marks the row failed only once the pre-send attempt budget is exhausted", async () => { - const deps = createDeps({ - run: null, - identity: { dmThreadId: null }, - outboundRows: [ - { - id: "out-9", - idempotencyKey: "invite:ch-1:mi-1", - kind: "dm", - identityId: "mi-1", - body: "last lookup try", + body: "fourth failure", status: "pending", providerHandle: null, attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS - 1, }, ], }); - deps.openDirectThread.mockRejectedValue(new Error("provider lookup 503")); await deliverMessagingOutbound(deps, {}, context); - expect(deps.sendToThread).not.toHaveBeenCalled(); expect(deps.rows).toEqual([ expect.objectContaining({ status: "failed", attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS }), ]); @@ -1149,8 +975,8 @@ describe("deliverMessagingOutbound pre-send failure retry", () => { describe("deliverMessagingOutbound retry enqueue failure", () => { it("propagates the enqueue failure so the job queue retries the drain", async () => { - const deps = createDeps({ identity: { dmThreadId: null } }); - deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503")); + const deps = createDeps({}); + deps.sendToThread.mockRejectedValueOnce(new Error("messaging provider 500")); deps.jobs.enqueue.mockRejectedValueOnce(new Error("queue down")); // A swallowed enqueue failure would strand the row in pending forever: @@ -1158,7 +984,6 @@ describe("deliverMessagingOutbound retry enqueue failure", () => { await expect(deliverMessagingOutbound(deps, { runId: "run-1" }, context)).rejects.toThrow( "queue down", ); - expect(deps.sendToThread).not.toHaveBeenCalled(); expect(deps.rows).toEqual([ expect.objectContaining({ kind: "dm", @@ -1170,9 +995,8 @@ describe("deliverMessagingOutbound retry enqueue failure", () => { // The delayed messaging.deliver job fires at nextAttemptAt: make the row due. deps.rows[0]!.nextAttemptAt = new Date(Date.now() - 1); - deps.openDirectThread.mockResolvedValueOnce("sendblue:opened:+15551234567"); await deliverMessagingOutbound(deps, { runId: "run-1" }, context); - expect(deps.sendToThread).toHaveBeenCalledTimes(1); + expect(deps.sendToThread).toHaveBeenCalledTimes(2); expect(deps.rows).toEqual([ expect.objectContaining({ kind: "dm", status: "sent", providerHandle: "handle-out-1" }), ]); diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index db8d891..b372f61 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -56,39 +56,23 @@ export async function deliverMessagingOutbound( await drain(deps, context); } -/** Whether a completed messaging run is safe to mark as durably mirrored. */ -type MessagingMirrorDisposition = "complete" | "retry"; - -type MirrorRunResult = { - disposition: MessagingMirrorDisposition; - trigger: string | null; - status: string | null; -}; - export async function mirrorMessagingOutbound( deps: MessagingDeliveryDeps, runId: string, ): Promise { - const result = await mirrorRun(deps, runId); - if ( - result.disposition !== "complete" || - result.trigger !== "messaging" || - result.status !== "completed" - ) { - return; - } + await mirrorRun(deps, runId); await deps.prisma.run.updateMany({ - where: { id: runId, trigger: "messaging", status: "completed" }, + where: { id: runId, trigger: "messaging" }, data: { messagingMirroredAt: new Date() }, }); } -async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise { +async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise { const run = await deps.prisma.run.findUnique({ where: { id: runId }, include: { sourceMessage: true }, }); - if (!run) return { disposition: "retry", trigger: null, status: null }; + if (!run) return; if (run.trigger === "messaging") { const sourceBlocks = (run.sourceMessage?.blocks ?? []) as MessageBlock[]; @@ -97,19 +81,19 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise row.body); - if (rows.length === 0) { - return { - disposition: messages.length === 0 ? "retry" : "complete", - trigger: run.trigger, - status: run.status, - }; - } + if (rows.length === 0) return; // Atomic dedupe: a concurrent messaging.deliver for the same run loses on // the idempotencyKey unique key instead of throwing P2002. await deps.prisma.messagingOutbound.createMany({ data: rows, skipDuplicates: true }); - return { disposition: "complete", trigger: run.trigger, status: run.status }; } /** @@ -149,15 +126,15 @@ async function mirrorChannelRun( deps: MessagingDeliveryDeps, run: { id: string; botId: string }, channelBlock: Extract, -): Promise { +): Promise { const identity = await deps.prisma.messagingIdentity.findUnique({ where: { botId: run.botId }, }); - if (!identity) return "complete"; + if (!identity) return; const channel = await deps.prisma.messagingChannel.findUnique({ where: { id: channelBlock.channelId }, }); - if (!channel) return "complete"; + if (!channel) return; const owner = await deps.prisma.user.findUnique({ where: { id: identity.userId }, select: { name: true }, @@ -170,27 +147,19 @@ async function mirrorChannelRun( where: { runId: run.id, role: "bot" }, orderBy: { seq: "asc" }, }); - // Ask answers are entered privately in the app. Check the same snapshot as - // the replies so a resumed run cannot publish an answer-influenced response. - if ( - replies.some((message) => - (message.blocks as MessageBlock[]).some( - (block) => block.kind === "ask" && block.status === "answered", - ), - ) - ) { - const hasMirrorableReply = replies.some( + // Ask answers are entered privately in the app. Skip only messages that carry + // an answered ask block so a resumed run can still mirror later text replies + // (idempotency keys dedupe anything already posted while the ask was pending). + const messages = replies + .filter( (message) => !(message.blocks as MessageBlock[]).some( (block) => block.kind === "ask" && block.status === "answered", - ) && extractText(message.blocks), - ); - return hasMirrorableReply ? "retry" : "complete"; - } - const messages = replies + ), + ) .map((message) => ({ message, text: extractText(message.blocks) })) .filter((entry) => entry.text); - if (messages.length === 0) return replies.length === 0 ? "retry" : "complete"; + if (messages.length === 0) return; await deps.prisma.messagingOutbound.createMany({ data: messages.map(({ message, text }) => ({ @@ -281,7 +250,6 @@ async function mirrorChannelRun( await deps.events.notify(peerThread.id, event.seq).catch(() => undefined); } } - return "complete"; } function escapeRegExp(value: string): string { @@ -308,53 +276,6 @@ function connectInvitePair( * restore pending and duplicate the YES/NO DM. A lost invite is fine; * reconnect starts a fresh cycle. */ -/** - * At-most-once provider send: any error after the call (or a lost response) - * keeps the outer claim as `sent` so drain does not restore pending and - * duplicate the message. A lost send is acceptable; the outbox row stays - * held without a providerHandle until status webhooks or manual cleanup. - */ -async function sendOutboundAtMostOnce( - deps: MessagingDeliveryDeps, - row: { id: string; idempotencyKey: string; body: string }, - threadId: string, - context: AdapterContext, -): Promise<"delivered" | "held"> { - try { - const sent = await deps.messaging.sendToThread( - { threadId, body: row.body, idempotencyKey: row.idempotencyKey }, - context, - ); - await deps.prisma.messagingOutbound.updateMany({ - where: { id: row.id }, - data: { providerHandle: sent.handle }, - }); - return "delivered"; - } catch { - return "held"; - } -} - -async function revertOutboundToPendingWithRetry( - deps: MessagingDeliveryDeps, - row: { id: string; attempts?: number | null }, -): Promise { - const attempts = (row.attempts ?? 0) + 1; - const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS; - const retryAt = exhausted ? null : new Date(Date.now() + messagingOutboundRetryDelayMs(attempts)); - await deps.prisma.messagingOutbound.updateMany({ - where: { id: row.id }, - data: { - attempts, - status: exhausted ? "failed" : "pending", - nextAttemptAt: retryAt, - }, - }); - if (!exhausted && retryAt) { - await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt)); - } -} - async function sendConnectInvite( deps: MessagingDeliveryDeps, row: { id: string; body: string }, @@ -421,81 +342,99 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom data: { status: "sent", nextAttemptAt: null }, }); if (claim.count === 0) continue; - if (row.kind === "group" || row.kind === "intro") { - if (!row.threadId) { + try { + if (row.kind === "group" || row.kind === "intro") { + if (!row.threadId) { + await deps.prisma.messagingOutbound.update({ + where: { id: row.id }, + data: { status: "failed" }, + }); + continue; + } + const sent = await deps.messaging.sendToThread( + { threadId: row.threadId, body: row.body }, + context, + ); + await deps.prisma.messagingOutbound.update({ + where: { id: row.id }, + data: { providerHandle: sent.handle }, + }); + continue; + } + const identity = row.identityId + ? await deps.prisma.messagingIdentity.findUnique({ where: { id: row.identityId } }) + : null; + if (!identity) { await deps.prisma.messagingOutbound.update({ where: { id: row.id }, data: { status: "failed" }, }); continue; } - await sendOutboundAtMostOnce( - deps, - { id: row.id, idempotencyKey: row.idempotencyKey, body: row.body }, - row.threadId, - context, - ); - continue; - } - const identity = row.identityId - ? await deps.prisma.messagingIdentity.findUnique({ where: { id: row.identityId } }) - : null; - if (!identity) { - await deps.prisma.messagingOutbound.update({ - where: { id: row.id }, - data: { status: "failed" }, - }); - continue; - } - // Sendblue's consecutive-outbound vendor cap; other providers have no - // equivalent limit, so do not hold Slack/WhatsApp/Telegram DMs. - if ( - identity.provider === "sendblue" && - identity.outboundSinceInbound >= MESSAGING_DM_OUTBOUND_CAP - ) { - // Cap holds are not failures: release the claim back to pending. - await deps.prisma.messagingOutbound.update({ - where: { id: row.id }, - data: { status: "pending" }, - }); - continue; - } - let threadId: string; - try { - threadId = await resolveDirectThread(deps, identity, context); - } catch { - // Pre-send setup failed; no provider call was made, so retry is safe. - await revertOutboundToPendingWithRetry(deps, row); - continue; - } - const invitePair = connectInvitePair(row.idempotencyKey); - if (invitePair) { - const result = await sendConnectInvite( - deps, - { id: row.id, body: row.body }, - threadId, - invitePair, - context, - ); - if (result === "delivered") { - await deps.prisma.messagingIdentity.update({ - where: { id: identity.id }, - data: { outboundSinceInbound: { increment: 1 } }, + // Sendblue's consecutive-outbound vendor cap; other providers have no + // equivalent limit, so do not hold Slack/WhatsApp/Telegram DMs. + if ( + identity.provider === "sendblue" && + identity.outboundSinceInbound >= MESSAGING_DM_OUTBOUND_CAP + ) { + // Cap holds are not failures: release the claim back to pending. + await deps.prisma.messagingOutbound.update({ + where: { id: row.id }, + data: { status: "pending" }, }); + continue; } - continue; - } - const result = await sendOutboundAtMostOnce( - deps, - { id: row.id, idempotencyKey: row.idempotencyKey, body: row.body }, - threadId, - context, - ); - if (result === "delivered") { + const threadId = await resolveDirectThread(deps, identity, context); + const invitePair = connectInvitePair(row.idempotencyKey); + if (invitePair) { + const result = await sendConnectInvite( + deps, + { id: row.id, body: row.body }, + threadId, + invitePair, + context, + ); + if (result === "delivered") { + await deps.prisma.messagingIdentity.update({ + where: { id: identity.id }, + data: { outboundSinceInbound: { increment: 1 } }, + }); + } + continue; + } + const sent = await deps.messaging.sendToThread({ threadId, body: row.body }, context); + await deps.prisma.messagingOutbound.updateMany({ + where: { id: row.id }, + data: { providerHandle: sent.handle }, + }); await deps.prisma.messagingIdentity.update({ where: { id: identity.id }, data: { outboundSinceInbound: { increment: 1 } }, }); + } catch { + // Transient provider errors go back to pending with a backed-off + // retry; only an exhausted budget is terminal. + const attempts = (row.attempts ?? 0) + 1; + const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS; + const retryAt = exhausted + ? null + : new Date(Date.now() + messagingOutboundRetryDelayMs(attempts)); + await deps.prisma.messagingOutbound.updateMany({ + where: { id: row.id }, + data: { + attempts, + status: exhausted ? "failed" : "pending", + nextAttemptAt: retryAt, + }, + }); + if (!exhausted && retryAt) { + // Propagate an enqueue failure: the messaging.deliver job then fails + // and the queue's own retry re-runs the drain. Swallowing it would + // strand the row in pending — no reconciler reclaims outbox rows. + // Re-entry is safe: the row is pending again and nextAttemptAt keeps + // other drains from racing the backoff window. + await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt)); + } } } } @@ -528,34 +467,20 @@ 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 = outboundStatusFromWebhook(event); + const status = + event.status === "ERROR" || event.status === "DECLINED" + ? "failed" + : event.status === "SENT" || event.status === "DELIVERED" + ? "sent" + : null; if (!status) return; await prisma.messagingOutbound.updateMany({ - where: { - providerHandle: event.handle, - status: { in: allowedSourcesForWebhookTarget(status) }, - }, + where: { providerHandle: event.handle }, data: { status }, }); }