fix(messaging): set introPostedAt only after intro delivery succeeds

Group intros were stamped as posted before the outbox drain ran. When
delivery enqueue failed or the intro row exhausted retries, the channel
stayed marked and no retry path existed — unlinked participants never saw
the onboarding message.

Stamp introPostedAt in the drain path after a successful provider send,
matching the messagingMirroredAt pattern from #42.

Co-authored-by: cynarAI <cynarAI@users.noreply.github.com>
This commit is contained in:
Cursor Agent 2026-09-22 07:02:20 +00:00 committed by Ingo Wagner
parent 31a6438145
commit 2ee72af28b
4 changed files with 73 additions and 4 deletions

View File

@ -804,6 +804,20 @@ describe("createMessagingInboundHandler channel routing", () => {
expect(deps.sendUserMessage).not.toHaveBeenCalled(); 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 () => { it("does not post a second intro once one was posted", async () => {
const deps = createDeps(); const deps = createDeps();
deps.prisma.messagingChannel.upsert = vi.fn(async () => ({ deps.prisma.messagingChannel.upsert = vi.fn(async () => ({

View File

@ -560,10 +560,6 @@ async function handleChannelEvent(
], ],
skipDuplicates: true, skipDuplicates: true,
}); });
await deps.prisma.messagingChannel.update({
where: { id: channel.id },
data: { introPostedAt: new Date() },
});
await deps.jobs.enqueue(messagingDeliverJob()).catch((error) => { await deps.jobs.enqueue(messagingDeliverJob()).catch((error) => {
getLogger().error("messaging intro enqueue error", error); getLogger().error("messaging intro enqueue error", error);
}); });

View File

@ -110,6 +110,9 @@ function createDeps(overrides: {
? [] ? []
: [{ status: overrides.connectionStatus ?? "pending" }], : [{ status: overrides.connectionStatus ?? "pending" }],
), ),
messagingChannel: {
updateMany: vi.fn(async () => ({ count: 1 })),
},
messagingOutbound: { messagingOutbound: {
findUnique: vi.fn(async ({ where }: { where?: { id?: string } } = {}) => { findUnique: vi.fn(async ({ where }: { where?: { id?: string } } = {}) => {
if (overrides.existingOutbox) return overrides.existingOutbox; if (overrides.existingOutbox) return overrides.existingOutbox;
@ -339,6 +342,56 @@ describe("deliverMessagingOutbound", () => {
expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "pending", attempts: 1 })); 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 () => { it("drains leftover pending rows without a run id", async () => {
const deps = createDeps({ const deps = createDeps({
run: null, run: null,

View File

@ -359,6 +359,12 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom
where: { id: row.id }, where: { id: row.id },
data: { providerHandle: sent.handle }, data: { providerHandle: sent.handle },
}); });
if (row.kind === "intro") {
await deps.prisma.messagingChannel.updateMany({
where: { threadId: row.threadId, introPostedAt: null },
data: { introPostedAt: new Date() },
});
}
continue; continue;
} }
const identity = row.identityId const identity = row.identityId