fix: stop repeated DMs after ambiguous Meta delivery errors

__tests__/comment-delivery.test.ts

@@ -0,0 +1,155 @@
+import { randomBytes } from "node:crypto";
+import { readdirSync, readFileSync } from "node:fs";
+import path from "node:path";
+import { Client } from "pg";
+import { PrismaPg } from "@prisma/adapter-pg";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { PrismaClient } from "../app/generated/prisma/client";
+
+const state = vi.hoisted(() => ({
+  db: undefined as unknown as import("../app/generated/prisma/client").PrismaClient,
+}));
+vi.mock("@/lib/db/client", () => ({
+  get prisma() {
+    return state.db;
+  },
+}));
+import { claimCommentDelivery } from "../lib/queue/comment-delivery";
+
+const databaseUrl = process.env.TEST_DATABASE_URL;
+const schema = `delivery_claims_${randomBytes(4).toString("hex")}`;
+let sql: Client;
+
+describe.skipIf(!databaseUrl)("durable comment delivery on Postgres", () => {
+  beforeAll(async () => {
+    sql = new Client({ connectionString: databaseUrl });
+    await sql.connect();
+    await sql.query(`CREATE SCHEMA "${schema}"`);
+    await sql.query(`SET search_path TO "${schema}"`);
+    const root = path.join(__dirname, "..", "prisma", "migrations");
+    for (const entry of readdirSync(root, { withFileTypes: true })
+      .filter((e) => e.isDirectory())
+      .sort((a, b) => a.name.localeCompare(b.name))) {
+      await sql.query(
+        readFileSync(path.join(root, entry.name, "migration.sql"), "utf8"),
+      );
+    }
+    state.db = new PrismaClient({
+      adapter: new PrismaPg({ connectionString: databaseUrl }, { schema }),
+    });
+    await state.db.user.create({
+      data: { id: "user", email: "delivery@example.test" },
+    });
+    await state.db.workspace.create({
+      data: { id: "workspace", name: "Delivery", ownerId: "user" },
+    });
+    await state.db.instagramAccount.create({
+      data: {
+        id: "account",
+        workspaceId: "workspace",
+        instagramId: "test_ig",
+        username: "delivery",
+        accessToken: "local-test-only",
+      },
+    });
+    await state.db.automation.create({
+      data: {
+        id: "automation",
+        workspaceId: "workspace",
+        instagramAccountId: "account",
+        name: "Test",
+        keywords: ["AI"],
+        dmMessage: "Test",
+      },
+    });
+  }, 60_000);
+  afterAll(async () => {
+    await state.db?.$disconnect();
+    if (sql) {
+      await sql.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`);
+      await sql.end();
+    }
+  });
+  async function seed(commentId: string) {
+    return state.db.dmLog.create({
+      data: {
+        workspaceId: "workspace",
+        instagramAccountId: "account",
+        automationId: "automation",
+        commenterId: "user_ig",
+        commentId,
+        commentText: "AI",
+      },
+    });
+  }
+  it("allows exactly one concurrent send and survives a lost result write", async () => {
+    await seed("concurrent");
+    const results = await Promise.all(
+      Array.from({ length: 8 }, () =>
+        claimCommentDelivery("automation", "concurrent", "dm"),
+      ),
+    );
+    expect(results.filter(Boolean)).toHaveLength(1);
+    // A restarted worker or a fresh polling job still observes the persisted claim.
+    expect(await claimCommentDelivery("automation", "concurrent", "dm")).toBe(
+      false,
+    );
+    const row = await state.db.dmLog.findUniqueOrThrow({
+      where: {
+        automationId_commentId: {
+          automationId: "automation",
+          commentId: "concurrent",
+        },
+      },
+    });
+    expect(row).toMatchObject({ attempts: 1, dmDeliveryUnconfirmed: true });
+    await seed("another-comment");
+    expect(
+      await claimCommentDelivery("automation", "another-comment", "dm"),
+    ).toBe(true);
+  });
+  it("caps actual attempts across independent jobs after confirmed rejections", async () => {
+    const row = await seed("bounded");
+    for (let i = 0; i < 3; i++) {
+      expect(await claimCommentDelivery("automation", "bounded", "dm")).toBe(
+        true,
+      );
+      await state.db.dmLog.update({
+        where: { id: row.id },
+        data: { dmDeliveryUnconfirmed: false, status: "FAILED" },
+      });
+    }
+    expect(await claimCommentDelivery("automation", "bounded", "dm")).toBe(
+      false,
+    );
+    expect(
+      (await state.db.dmLog.findUniqueOrThrow({ where: { id: row.id } }))
+        .attempts,
+    ).toBe(3);
+  });
+  it("claims public replies independently and never repeats a confirmed send", async () => {
+    const row = await seed("public");
+    expect(await claimCommentDelivery("automation", "public", "public")).toBe(
+      true,
+    );
+    expect(await claimCommentDelivery("automation", "public", "public")).toBe(
+      false,
+    );
+    expect(await claimCommentDelivery("automation", "public", "dm")).toBe(true);
+    await state.db.dmLog.update({
+      where: { id: row.id },
+      data: {
+        publicReplyDeliveryUnconfirmed: false,
+        publicReplySentAt: new Date(),
+        dmDeliveryUnconfirmed: false,
+        status: "SENT",
+      },
+    });
+    expect(await claimCommentDelivery("automation", "public", "public")).toBe(
+      false,
+    );
+    expect(await claimCommentDelivery("automation", "public", "dm")).toBe(
+      false,
+    );
+  });
+});

__tests__/comment-reconciler.test.ts

@@ -7,13 +7,34 @@
 
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
-const { mockPrisma } = vi.hoisted(() => ({
-  mockPrisma: { $queryRaw: vi.fn() },
+const { mockPrisma, queueAdd, readComments } = vi.hoisted(() => ({
+  mockPrisma: {
+    $queryRaw: vi.fn(),
+    automation: { findMany: vi.fn() },
+    dmLog: { findMany: vi.fn() },
+    operationalEvent: { create: vi.fn() },
+  },
+  queueAdd: vi.fn(),
+  readComments: vi.fn(),
+}));
+vi.mock("@/lib/queue/client", () => ({
+  getDMQueue: () => ({ add: queueAdd }),
+}));
+vi.mock("@/lib/instagram/provider", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@/lib/instagram/provider")>()),
+  createInstagramContext: async () => ({
+    provider: "META",
+    accessToken: "local-test",
+  }),
+  getRecentMediaComments: readComments,
 }));
 
 vi.mock("@/lib/db/client", () => ({ prisma: mockPrisma }));
 
-import { adMediaFor } from "../lib/polling/comment-reconciler";
+import {
+  adMediaFor,
+  reconcileComments,
+} from "../lib/polling/comment-reconciler";
 
 const POST = "18023946917554990";
 const AD = "17899788633163100";
@@ -29,12 +50,18 @@ describe("adMediaFor", () => {
   });
 
   it("never returns the post itself, so it is not swept twice", async () => {
-    mockPrisma.$queryRaw.mockResolvedValue([{ mediaId: AD }, { mediaId: POST }]);
+    mockPrisma.$queryRaw.mockResolvedValue([
+      { mediaId: AD },
+      { mediaId: POST },
+    ]);
     await expect(adMediaFor(POST)).resolves.toEqual([AD]);
   });
 
   it("drops rows without a media id", async () => {
-    mockPrisma.$queryRaw.mockResolvedValue([{ mediaId: null }, { mediaId: AD }]);
+    mockPrisma.$queryRaw.mockResolvedValue([
+      { mediaId: null },
+      { mediaId: AD },
+    ]);
     await expect(adMediaFor(POST)).resolves.toEqual([AD]);
   });
 
@@ -48,3 +75,64 @@ describe("adMediaFor", () => {
     await expect(adMediaFor(POST)).resolves.toEqual([]);
   });
 });
+
+describe("comment polling does not recreate unsafe sends", () => {
+  beforeEach(() => {
+    queueAdd.mockReset();
+    mockPrisma.operationalEvent.create.mockResolvedValue({});
+    mockPrisma.$queryRaw.mockResolvedValue([]);
+    mockPrisma.automation.findMany.mockResolvedValue([
+      {
+        id: "campaign",
+        name: "Campaign",
+        workspaceId: "workspace",
+        postId: POST,
+        matchAnyWord: false,
+        keywords: ["AI"],
+        publicReplyEnabled: true,
+        instagramAccount: {
+          id: "connection",
+          instagramId: "owner",
+          provider: "META",
+          accessToken: "test",
+        },
+      },
+    ]);
+    readComments.mockResolvedValue([
+      {
+        id: "old",
+        text: "AI",
+        from: { id: "reader" },
+        timestamp: new Date().toISOString(),
+      },
+      {
+        id: "new",
+        text: "AI",
+        from: { id: "another-reader" },
+        timestamp: new Date().toISOString(),
+      },
+    ]);
+  });
+  it.each([
+    { status: "FAILED", attempts: 1, dmDeliveryUnconfirmed: true },
+    { status: "FAILED", attempts: 3, dmDeliveryUnconfirmed: false },
+    {
+      status: "FAILED",
+      attempts: 1,
+      errorMessage: "MetaApiError 1: An unknown error has occurred.",
+    },
+    { status: "PENDING", attempts: 1, dmDeliveryUnconfirmed: true },
+  ])(
+    "leaves an unsafe old comment alone while continuing new comments: %j",
+    async (state) => {
+      mockPrisma.dmLog.findMany.mockResolvedValue([
+        { commentId: "old", publicReplySentAt: new Date(), ...state },
+      ]);
+      await reconcileComments();
+      await reconcileComments();
+      expect(queueAdd).toHaveBeenCalledTimes(2);
+      for (const [, data] of queueAdd.mock.calls)
+        expect(data.commentId).toBe("new");
+    },
+  );
+});