diff --git a/apps/api/src/messaging-inbound.test.ts b/apps/api/src/messaging-inbound.test.ts index 5f6991d..a331198 100644 --- a/apps/api/src/messaging-inbound.test.ts +++ b/apps/api/src/messaging-inbound.test.ts @@ -804,6 +804,20 @@ describe("createMessagingInboundHandler channel routing", () => { expect(deps.sendUserMessage).not.toHaveBeenCalled(); }); + it("does not mark intro posted before delivery enqueue succeeds", async () => { + const deps = createDeps(); + deps.jobs.enqueue = vi.fn(async () => { + throw new Error("queue unavailable"); + }); + const handle = createMessagingInboundHandler(deps); + await handle(groupEvent); + + expect(deps.outboundRows).toEqual( + expect.arrayContaining([expect.objectContaining({ kind: "intro", idempotencyKey: "intro:ch-1" })]), + ); + expect(deps.prisma.messagingChannel.update).not.toHaveBeenCalled(); + }); + it("does not post a second intro once one was posted", async () => { const deps = createDeps(); deps.prisma.messagingChannel.upsert = vi.fn(async () => ({ diff --git a/apps/api/src/messaging-inbound.ts b/apps/api/src/messaging-inbound.ts index 0a00b7a..673b026 100644 --- a/apps/api/src/messaging-inbound.ts +++ b/apps/api/src/messaging-inbound.ts @@ -560,10 +560,6 @@ async function handleChannelEvent( ], skipDuplicates: true, }); - await deps.prisma.messagingChannel.update({ - where: { id: channel.id }, - data: { introPostedAt: new Date() }, - }); await deps.jobs.enqueue(messagingDeliverJob()).catch((error) => { getLogger().error("messaging intro enqueue error", error); }); diff --git a/packages/adapters/src/messaging-delivery.test.ts b/packages/adapters/src/messaging-delivery.test.ts index 2d06a41..143b698 100644 --- a/packages/adapters/src/messaging-delivery.test.ts +++ b/packages/adapters/src/messaging-delivery.test.ts @@ -110,6 +110,9 @@ function createDeps(overrides: { ? [] : [{ status: overrides.connectionStatus ?? "pending" }], ), + messagingChannel: { + updateMany: vi.fn(async () => ({ count: 1 })), + }, messagingOutbound: { findUnique: vi.fn(async ({ where }: { where?: { id?: string } } = {}) => { if (overrides.existingOutbox) return overrides.existingOutbox; @@ -339,6 +342,56 @@ describe("deliverMessagingOutbound", () => { expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "pending", attempts: 1 })); }); + it("marks introPostedAt only after a pending intro row is delivered", async () => { + const deps = createDeps({ + run: null, + outboundRows: [ + { + id: "out-intro", + idempotencyKey: "intro:ch-1", + kind: "intro", + threadId: "sendblue:grp-1", + body: "Hi. This line hosts Rakazo personal agents.", + status: "pending", + providerHandle: null, + }, + ], + }); + await deliverMessagingOutbound(deps, {}, context); + + expect(deps.sendToThread).toHaveBeenCalledWith( + { threadId: "sendblue:grp-1", body: "Hi. This line hosts Rakazo personal agents." }, + context, + ); + expect(deps.prisma.messagingChannel.updateMany).toHaveBeenCalledWith({ + where: { threadId: "sendblue:grp-1", introPostedAt: null }, + data: { introPostedAt: expect.any(Date) }, + }); + }); + + it("does not mark introPostedAt when intro delivery fails", async () => { + const deps = createDeps({ + run: null, + sendError: new Error("provider down"), + outboundRows: [ + { + id: "out-intro", + idempotencyKey: "intro:ch-1", + kind: "intro", + threadId: "sendblue:grp-1", + body: "Hi. This line hosts Rakazo personal agents.", + status: "pending", + attempts: 0, + providerHandle: null, + }, + ], + }); + await deliverMessagingOutbound(deps, {}, context); + + expect(deps.prisma.messagingChannel.updateMany).not.toHaveBeenCalled(); + expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "pending", attempts: 1 })); + }); + it("drains leftover pending rows without a run id", async () => { const deps = createDeps({ run: null, diff --git a/packages/adapters/src/messaging-delivery.ts b/packages/adapters/src/messaging-delivery.ts index b372f61..8d8522e 100644 --- a/packages/adapters/src/messaging-delivery.ts +++ b/packages/adapters/src/messaging-delivery.ts @@ -359,6 +359,12 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom where: { id: row.id }, data: { providerHandle: sent.handle }, }); + if (row.kind === "intro") { + await deps.prisma.messagingChannel.updateMany({ + where: { threadId: row.threadId, introPostedAt: null }, + data: { introPostedAt: new Date() }, + }); + } continue; } const identity = row.identityId