diff --git a/__tests__/comment-delivery.test.ts b/__tests__/comment-delivery.test.ts new file mode 100644 index 000000000..0bb95b939 --- /dev/null +++ b/__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, + ); + }); +}); diff --git a/__tests__/comment-reconciler.test.ts b/__tests__/comment-reconciler.test.ts index 9f41e99f1..c1f5ba301 100644 --- a/__tests__/comment-reconciler.test.ts +++ b/__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()), + 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"); + }, + ); +}); diff --git a/__tests__/dm-worker.test.ts b/__tests__/dm-worker.test.ts index 7fc996741..d99a526cf 100644 --- a/__tests__/dm-worker.test.ts +++ b/__tests__/dm-worker.test.ts @@ -29,6 +29,7 @@ const { findFirst: vi.fn(), upsert: vi.fn(), update: vi.fn(), + updateMany: vi.fn(), create: vi.fn(), }, instagramAccount: { @@ -136,6 +137,7 @@ vi.mock("bullmq", () => { }; }); +import { MetaApiError, RateLimitError } from "@/lib/meta/client"; import { createDMWorker } from "../lib/queue/dm-worker"; const usagePeriodStart = new Date("2026-05-01T00:00:00.000Z"); @@ -234,7 +236,8 @@ beforeEach(() => { args.where?.status === "SENT" ? null : { commenterName: "commenter_user" } ); mockPrisma.dmLog.upsert.mockResolvedValue({}); - mockPrisma.dmLog.update.mockResolvedValue({}); + mockPrisma.dmLog.update.mockReset().mockResolvedValue({}); + mockPrisma.dmLog.updateMany.mockReset().mockResolvedValue({ count: 1 }); mockPrisma.instagramAccount.findUnique.mockResolvedValue({ workspaceId: "workspace_123", }); @@ -485,7 +488,7 @@ describe("DM Worker — Full Pipeline", () => { }); it("should log FAILED, release usage, and re-throw when private reply sending fails", async () => { - const error = new Error("API Error"); + const error = new RateLimitError("API Error"); mockSendPrivateReply.mockRejectedValue(error); const processor = getProcessor(); @@ -815,7 +818,7 @@ describe("DM Worker — Full Pipeline", () => { trackedLinks: [], }); mockSendDirectMessage.mockRejectedValue( - new Error("This message is sent outside of allowed window.") + new MetaApiError(10, undefined, undefined, "This message is sent outside of allowed window.") ); const processor = getProcessor(); @@ -901,7 +904,7 @@ describe("DM Worker — one private reply per comment", () => { }, ]); mockSendPrivateReplyWithLinkButton.mockRejectedValue( - new Error("The comment is invalid for a private reply") + new MetaApiError(100, undefined, undefined, "The comment is invalid for a private reply") ); const processor = getProcessor(); @@ -920,7 +923,7 @@ describe("DM Worker — one private reply per comment", () => { expect.objectContaining({ data: expect.objectContaining({ status: "FAILED", - errorMessage: "The comment is invalid for a private reply", + errorMessage: expect.stringContaining("The comment is invalid for a private reply"), }), }) ); @@ -940,7 +943,7 @@ describe("DM Worker — one private reply per comment", () => { }, ]); mockSendPrivateReplyWithLinkButton.mockRejectedValue( - new Error("Unsupported message template") + new MetaApiError(100, undefined, undefined, "Unsupported message template") ); const processor = getProcessor(); @@ -1401,3 +1404,99 @@ describe("durable Zernio postback delivery", () => { } }); }); + +describe("ambiguous Meta sends and durable comment claims", () => { + const withLinks = { ...mockAutomation, trackedLinks: [{ slug: "resource", label: null, destinationUrl: "https://example.com" }] }; + + it("never falls back or retries Meta code 1, even though Meta may have sent the message", async () => { + mockPrisma.automation.findMany.mockResolvedValue([withLinks]); + mockSendPrivateReplyWithLinkButton.mockRejectedValue(new MetaApiError(1, undefined, undefined, "An unknown error has occurred.")); + await expect(getProcessor()(createMockJob())).rejects.toMatchObject({ name: "UnrecoverableError" }); + expect(mockSendPrivateReplyWithLinkButton).toHaveBeenCalledTimes(1); + expect(mockSendPrivateReply).not.toHaveBeenCalled(); + expect(mockPrisma.dmLog.update).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ status: "FAILED", dmDeliveryUnconfirmed: true }) })); + expect(mockReleaseDMSlot).not.toHaveBeenCalled(); + expect(mockReleaseWorkspaceDMReservation).not.toHaveBeenCalled(); + }); + + it("retains the fallback's uncertain outcome instead of replacing it with the first rejection", async () => { + mockPrisma.automation.findMany.mockResolvedValue([withLinks]); + mockSendPrivateReplyWithLinkButton.mockRejectedValue(new MetaApiError(100, undefined, undefined, "Unsupported message template")); + mockSendPrivateReply.mockRejectedValue(new Error("Connection reset after send")); + await expect(getProcessor()(createMockJob())).rejects.toMatchObject({ name: "UnrecoverableError", message: expect.stringContaining("Connection reset after send") }); + expect(mockSendPrivateReply).toHaveBeenCalledTimes(1); + expect(mockPrisma.dmLog.update).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ dmDeliveryUnconfirmed: true }) })); + }); + + it("blocks the historical code 1 failures before any further network send", async () => { + mockPrisma.dmLog.findUnique.mockResolvedValue({ status: "FAILED", attempts: 1, errorMessage: "MetaApiError 1: An unknown error has occurred.", dmDeliveryUnconfirmed: false }); + await getProcessor()(createMockJob()); + expect(mockSendPrivateReply).not.toHaveBeenCalled(); + expect(mockPrisma.dmLog.update).toHaveBeenCalledWith(expect.objectContaining({ data: { dmDeliveryUnconfirmed: true } })); + }); + + it("cannot reset the lifetime attempt limit by enqueueing a new job", async () => { + mockPrisma.dmLog.findUnique.mockResolvedValue({ status: "FAILED", attempts: 3, dmDeliveryUnconfirmed: false }); + await getProcessor()({ ...createMockJob(), id: "new-poll-job", attemptsMade: 0 }); + expect(mockSendPrivateReply).not.toHaveBeenCalled(); + expect(mockPrisma.dmLog.updateMany).not.toHaveBeenCalled(); + }); + + it("allows only the claim winner to send when webhook and polling overlap", async () => { + let claimed = false; + mockPrisma.dmLog.updateMany.mockImplementation(async () => { + if (claimed) return { count: 0 }; + claimed = true; + return { count: 1 }; + }); + const process = getProcessor(); + await Promise.all([process(createMockJob()), process({ ...createMockJob(), id: "poll-job" })]); + expect(mockSendPrivateReply).toHaveBeenCalledTimes(1); + }); + + it("keeps the pre-send claim when both success and failure log writes fail", async () => { + mockPrisma.dmLog.update.mockRejectedValue(new Error("Database unavailable")); + const process = getProcessor(); + await expect(process(createMockJob())).rejects.toThrow("Database unavailable"); + expect(mockPrisma.dmLog.updateMany).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ dmDeliveryUnconfirmed: true }) })); + mockPrisma.dmLog.findUnique.mockResolvedValue({ status: "PENDING", attempts: 1, dmDeliveryUnconfirmed: true }); + await process({ ...createMockJob(), id: "retry-after-db-recovers" }); + expect(mockSendPrivateReply).toHaveBeenCalledTimes(1); + expect(mockReleaseWorkspaceDMReservation).not.toHaveBeenCalled(); + }); + + it("does not send at all when storing the claim fails", async () => { + mockPrisma.dmLog.updateMany.mockRejectedValue(new Error("Database unavailable")); + await expect(getProcessor()(createMockJob())).rejects.toThrow("Database unavailable"); + expect(mockSendPrivateReply).not.toHaveBeenCalled(); + }); +}); + +it("deduplicates a redelivered Meta button tap after queue retention expires", async () => { + const claims = new Set(); + mockPrisma.postbackDelivery.create.mockImplementation(async ({ data }: { data: { id: string } }) => { + if (claims.has(data.id)) throw { code: "P2002" }; + claims.add(data.id); + return data; + }); + mockPrisma.automation.findFirst.mockResolvedValue(mockAutomation); + const data = { instagramAccountId: "ig_456", userId: "commenter_999", payload: "reveal:auto_789", mid: "same-meta-tap" }; + const process = getProcessor(); + await process(createMockPostbackJob(data)); + await process({ ...createMockPostbackJob(data), id: "redelivered-after-eviction" }); + expect(mockSendDirectMessage).toHaveBeenCalledTimes(1); +}); + +it("retains the public reply claim if sending succeeded but its log write failed", async () => { + const { sendCommentReply } = await import("@/lib/meta/client"); + vi.mocked(sendCommentReply).mockResolvedValue({ id: "public-reply" }); + mockPrisma.automation.findMany.mockResolvedValue([{ ...mockAutomation, publicReplyEnabled: true, publicReplyMessages: ["Sent!"] }]); + mockPrisma.dmLog.update.mockRejectedValueOnce(new Error("Lost database connection after public send")); + const process = getProcessor(); + await process(createMockJob()); + expect(mockPrisma.dmLog.update).toHaveBeenCalledWith(expect.objectContaining({ data: expect.objectContaining({ publicReplyDeliveryUnconfirmed: true }) })); + mockPrisma.dmLog.findUnique.mockResolvedValue({ status: "SENT", publicReplyDeliveryUnconfirmed: true }); + await process({ ...createMockJob(), id: "next-poll" }); + expect(sendCommentReply).toHaveBeenCalledTimes(1); + expect(mockSendPrivateReply).toHaveBeenCalledTimes(1); +}); diff --git a/docs/delivery-retries.md b/docs/delivery-retries.md new file mode 100644 index 000000000..60213d874 --- /dev/null +++ b/docs/delivery-retries.md @@ -0,0 +1,27 @@ +# Comment delivery retries + +A failed API response is not proof that Instagram did not deliver a message. +In particular, Meta error 1 has been observed alongside actual inbox delivery. +Do not retry that outcome as a plain-text fallback or a new polling job. + +The comment worker claims each DM/public-reply leg with an atomic DmLog update +before calling Instagram. The existing delivery-unconfirmed flags serve as +in-flight claims. Success replaces a claim with the corresponding sent state. +Only explicit provider rejections release a claim. Network/unknown API errors, +process crashes and failed result writes leave the claim in place. This favors +avoiding duplicate messages; inspect the inbox before manually retrying an +uncertain delivery, including a crash between the claim and the network call. + +DM send attempts are incremented in the database, with a limit of three across +all webhook, retry and polling jobs for the campaign/comment pair. New job IDs +cannot reset that limit. Polling skips finished, uncertain or exhausted DM legs +while still allowing an unfinished public reply leg to be handled independently. +Historical Meta code 1/2/5xx failures are treated as uncertain on both paths. + +A text fallback requires an explicit Meta code 100 template/button rejection. +If that fallback fails, its own outcome is preserved instead of being replaced +by the earlier template rejection. Meta button-tap jobs also use the existing +durable postback claim so redelivering the same tap cannot send it again. + +No campaign is paused by this policy. New comments continue through the normal +flow. No schema migration is required. diff --git a/lib/instagram/delivery-errors.ts b/lib/instagram/delivery-errors.ts new file mode 100644 index 000000000..ebc69889f --- /dev/null +++ b/lib/instagram/delivery-errors.ts @@ -0,0 +1,57 @@ +import { + MetaApiError, + RateLimitError, + TokenExpiredError, +} from "@/lib/meta/client"; +import { + ZernioApiError, + ZernioDeliveryUnconfirmedError, +} from "@/lib/zernio/client"; + +export class DeliveryUnconfirmedError extends Error { + constructor(error: unknown) { + const detail = + error instanceof Error ? error.message : "Unknown send outcome"; + super( + `Message delivery is unconfirmed; automatic retries stopped. Inspect the Instagram inbox before retrying. ${detail}`, + ); + this.name = "DeliveryUnconfirmedError"; + } +} + +export function isDeliveryUnconfirmed( + error: unknown, +): error is DeliveryUnconfirmedError | ZernioDeliveryUnconfirmedError { + return ( + error instanceof DeliveryUnconfirmedError || + error instanceof ZernioDeliveryUnconfirmedError + ); +} + +// Only explicit rejections prove that no message was delivered. Meta code 1, +// network failures, invalid responses and local persistence failures do not. +export function isConfirmedSendRejection(error: unknown): boolean { + if (error instanceof RateLimitError || error instanceof TokenExpiredError) + return true; + if (error instanceof ZernioApiError) + return error.code >= 400 && error.code < 500; + return ( + error instanceof MetaApiError && [10, 100, 200, 551].includes(error.code) + ); +} + +export function classifySendError(error: unknown): unknown { + return isDeliveryUnconfirmed(error) || isConfirmedSendRejection(error) + ? error + : new DeliveryUnconfirmedError(error); +} + +// Old workers logged these uncertain Meta responses as retryable failures. +export function hasLegacyUnconfirmedDelivery( + error: string | null | undefined, +): boolean { + return Boolean( + error && + /(?:MetaApiError (?:1|2|5\d\d):|\[code=(?:1|2|5\d\d)\b)/.test(error), + ); +} diff --git a/lib/polling/comment-reconciler.ts b/lib/polling/comment-reconciler.ts index ecc9cc16d..ffcc6ea62 100644 --- a/lib/polling/comment-reconciler.ts +++ b/lib/polling/comment-reconciler.ts @@ -25,6 +25,8 @@ * filter on the account to widen results. */ +import { MAX_COMMENT_SEND_ATTEMPTS } from "@/lib/queue/comment-delivery"; +import { hasLegacyUnconfirmedDelivery } from "@/lib/instagram/delivery-errors"; import { prisma } from "@/lib/db/client"; import { getDMQueue } from "@/lib/queue/client"; import { @@ -235,18 +237,25 @@ async function sweepCampaign({ // enough — the reply still has to land); otherwise a SENT DM is enough. This // is what lets a comment whose DM sent but whose public reply failed come // back and retry the reply. - const handled = await prisma.dmLog.findMany({ + const logs = await prisma.dmLog.findMany({ where: { automationId: automation.id, commentId: { in: needsAction.map((c) => c.id) }, - AND: [ - { OR: [{ status: "SENT" }, { dmDeliveryUnconfirmed: true }] }, - ...(automation.publicReplyEnabled ? [{ OR: [{ publicReplySentAt: { not: null } }, { publicReplyDeliveryUnconfirmed: true }] }] : []), - ], }, - select: { commentId: true }, + select: { + commentId: true, status: true, attempts: true, errorMessage: true, + dmDeliveryUnconfirmed: true, publicReplySentAt: true, + publicReplyDeliveryUnconfirmed: true, + }, }); - const handledSet = new Set(handled.map((h) => h.commentId)); + const handledSet = new Set(logs.filter((log) => { + const dmStopped = log.status === "SENT" || log.status === "SKIPPED_PLAN_LIMIT" || + log.dmDeliveryUnconfirmed || log.attempts >= MAX_COMMENT_SEND_ATTEMPTS || + (log.status === "FAILED" && hasLegacyUnconfirmedDelivery(log.errorMessage)); + const replyStopped = !automation.publicReplyEnabled || + log.publicReplySentAt || log.publicReplyDeliveryUnconfirmed; + return dmStopped && replyStopped; + }).map((log) => log.commentId)); // Oldest first, so whoever commented earliest gets answered first, capped. const fresh = needsAction @@ -258,8 +267,8 @@ async function sweepCampaign({ // No deterministic jobId here: a retained completed/failed job from an // earlier sweep would otherwise be treated as a duplicate and silently // drop this add, so the comment would never be retried. Dedup is handled - // above (owner-reply + DmLog guards) and the worker is idempotent - // (publicReplySentAt / SENT), so re-processing a comment is safe. + // above and by the worker's atomic, durable per-leg claims. Send attempts + // are counted in DmLog across jobs, so a sweep cannot reset the budget. await queue.add("process-comment", { instagramAccountId: account.instagramId, accountConnectionId: account.id, diff --git a/lib/queue/comment-delivery.ts b/lib/queue/comment-delivery.ts new file mode 100644 index 000000000..74be7e550 --- /dev/null +++ b/lib/queue/comment-delivery.ts @@ -0,0 +1,35 @@ +import { prisma } from "@/lib/db/client"; + +export const MAX_COMMENT_SEND_ATTEMPTS = 3; + +// A durable, atomic claim precedes the external side effect. If the process +// crashes or the final log write fails, an uncertain send must not be repeated. +export async function claimCommentDelivery( + automationId: string, + commentId: string, + leg: "dm" | "public", +): Promise { + const result = await prisma.dmLog.updateMany({ + where: { + automationId, + commentId, + ...(leg === "dm" + ? { + status: { not: "SENT" as const }, + dmDeliveryUnconfirmed: false, + attempts: { lt: MAX_COMMENT_SEND_ATTEMPTS }, + } + : { publicReplySentAt: null, publicReplyDeliveryUnconfirmed: false }), + }, + data: + leg === "dm" + ? { + dmDeliveryUnconfirmed: true, + attempts: { increment: 1 }, + status: "PENDING", + errorMessage: null, + } + : { publicReplyDeliveryUnconfirmed: true }, + }); + return result.count === 1; +} diff --git a/lib/queue/dm-worker.ts b/lib/queue/dm-worker.ts index ec5f12d75..3f94e553e 100644 --- a/lib/queue/dm-worker.ts +++ b/lib/queue/dm-worker.ts @@ -1,3 +1,10 @@ +import { + classifySendError, + hasLegacyUnconfirmedDelivery, + isConfirmedSendRejection, + isDeliveryUnconfirmed, +} from "@/lib/instagram/delivery-errors"; +import { claimCommentDelivery, MAX_COMMENT_SEND_ATTEMPTS } from "./comment-delivery"; import { createHash } from "node:crypto"; import { UnrecoverableError, Worker, type Job } from "bullmq"; import { @@ -45,10 +52,7 @@ import { } from "@/lib/tracking/message"; import { TRACKED_LINK_ORDER } from "@/lib/tracking/link-order"; -import { - ZernioApiError, - ZernioDeliveryUnconfirmedError, -} from "@/lib/zernio/client"; +import { ZernioApiError } from "@/lib/zernio/client"; const BACKOFF_DELAYS = [5 * 60 * 1000, 15 * 60 * 1000, 45 * 60 * 1000]; @@ -81,8 +85,10 @@ function isTemplateRejection(error: unknown): boolean { ) { return false; } - const message = error instanceof Error ? error.message : ""; - return !NON_TEMPLATE_REJECTIONS.some((pattern) => pattern.test(message)); + // Falling back is another send: allow it only for a proven template error. + return error instanceof MetaApiError && error.code === 100 && + /template|button/i.test(error.message) && + !NON_TEMPLATE_REJECTIONS.some((pattern) => pattern.test(error.message)); } type WorkerTrackedLink = { @@ -206,8 +212,8 @@ async function sendRevealDirectMessage({ bodyText ), }); - } catch { - throw buttonError; + } catch (fallbackError) { + throw classifySendError(fallbackError); } } } @@ -284,9 +290,18 @@ async function processComment(job: Job): Promise { }, }); + if (existingLog?.status === "FAILED" && hasLegacyUnconfirmedDelivery(existingLog.errorMessage)) { + await prisma.dmLog.update({ + where: { automationId_commentId: { automationId: automation.id, commentId } }, + data: { dmDeliveryUnconfirmed: true }, + }); + existingLog.dmDeliveryUnconfirmed = true; + } + const alreadyDmd = existingLog?.status === "SENT"; const alreadyPublicReplied = Boolean(existingLog?.publicReplySentAt); - const needsDm = !alreadyDmd && !existingLog?.dmDeliveryUnconfirmed; + const needsDm = !alreadyDmd && !existingLog?.dmDeliveryUnconfirmed && + (existingLog?.attempts ?? 0) < MAX_COMMENT_SEND_ATTEMPTS; // Skip only when there is genuinely nothing left to do. A comment whose DM // already sent but whose public reply never posted (e.g. it hit a rate @@ -361,37 +376,18 @@ async function processComment(job: Job): Promise { continue; } - // Ensure a log row exists before the public reply leg (which updates it). - // Only (re)set PENDING when the DM will actually be attempted, so a prior - // SENT is never clobbered while we come back just to retry the public reply. - if (!existingLog) { - await prisma.dmLog.create({ - data: { - workspaceId: automation.workspaceId, - automationId: automation.id, - instagramAccountId: automation.instagramAccountId, - commenterId, - commenterName, - commentText, - commentId, - matchedKeyword: matchResult.matchedKeyword, - status: "PENDING", - attempts: job.attemptsMade + 1, - }, - }); - } else if (needsDm) { - await prisma.dmLog.update({ - where: { - automationId_commentId: { automationId: automation.id, commentId }, - }, - data: { - status: "PENDING", - attempts: job.attemptsMade + 1, - matchedKeyword: matchResult.matchedKeyword, - errorMessage: null, - }, - }); - } + await prisma.dmLog.upsert({ + where: { automationId_commentId: { automationId: automation.id, commentId } }, + create: { + workspaceId: automation.workspaceId, + automationId: automation.id, + instagramAccountId: automation.instagramAccountId, + commenterId, commenterName, commentText, commentId, + matchedKeyword: matchResult.matchedKeyword, + status: "PENDING", + }, + update: {}, + }); // Public reply leg — decoupled from the DM and posted first so a DM failure // (e.g. a non-follower whose messaging is restricted) never suppresses it. @@ -406,7 +402,8 @@ async function processComment(job: Job): Promise { automation.publicReplyEnabled && replyPool.length > 0 && !existingLog?.publicReplySentAt && - !existingLog?.publicReplyDeliveryUnconfirmed + !existingLog?.publicReplyDeliveryUnconfirmed && + await claimCommentDelivery(automation.id, commentId, "public") ) { try { const chosen = replyPool[Math.floor(Math.random() * replyPool.length)]; @@ -425,7 +422,7 @@ async function processComment(job: Job): Promise { where: { automationId_commentId: { automationId: automation.id, commentId }, }, - data: { publicReplySentAt: new Date(), publicReplyError: null }, + data: { publicReplySentAt: new Date(), publicReplyError: null, publicReplyDeliveryUnconfirmed: false }, }); } catch (error) { console.error( @@ -440,7 +437,7 @@ async function processComment(job: Job): Promise { commentId, }, }, - data: { publicReplyError: formatError(error), publicReplyDeliveryUnconfirmed: error instanceof ZernioDeliveryUnconfirmedError }, + data: { publicReplyError: formatError(classifySendError(error)), publicReplyDeliveryUnconfirmed: !isConfirmedSendRejection(error) }, }) .catch(() => {}); } @@ -514,9 +511,7 @@ async function processComment(job: Job): Promise { }, data: { status: "FAILED", - attempts: job.attemptsMade + 1, errorMessage: formatError(error), - dmDeliveryUnconfirmed: error instanceof ZernioDeliveryUnconfirmedError, }, }); throw error; @@ -600,6 +595,20 @@ async function processComment(job: Job): Promise { : alreadyFollows !== true; } + let claimed; + try { + claimed = await claimCommentDelivery(automation.id, commentId, "dm"); + } catch (error) { + if (rateLimit?.reserved) await releaseDMSlot(instagramAccountId); + await releaseWorkspaceDMReservation(automation.workspaceId, usage.periodStart); + throw error; + } + if (!claimed) { + if (rateLimit?.reserved) await releaseDMSlot(instagramAccountId); + await releaseWorkspaceDMReservation(automation.workspaceId, usage.periodStart); + continue; + } + let delivered = false; try { if (useOpeningDm) { const openingText = renderMessageWithTracking({ @@ -679,11 +688,8 @@ async function processComment(job: Job): Promise { message: fallbackMessage, postId: mediaId, }); - } catch { - // The first attempt consumed the comment's single private reply, so - // this one reports "invalid for a private reply" no matter what the - // underlying problem was. Surface the original rejection instead. - throw buttonError; + } catch (fallbackError) { + throw classifySendError(fallbackError); } } } else { @@ -701,6 +707,7 @@ async function processComment(job: Job): Promise { }); } + delivered = true; await prisma.dmLog.update({ where: { automationId_commentId: { @@ -711,36 +718,27 @@ async function processComment(job: Job): Promise { data: { status: "SENT", dmSentAt: new Date(), + dmDeliveryUnconfirmed: false, errorMessage: null, }, }); } catch (error) { - // The rate slot was reserved before the send; this send did not deliver a - // DM, so hand the slot back instead of burning it (and burning more on - // each BullMQ retry) until the hourly TTL expires. - if (rateLimit?.reserved) { - await releaseDMSlot(instagramAccountId); + const sendError = classifySendError(error); + // Retain reservations if the provider may have delivered the message. + if (isConfirmedSendRejection(sendError)) { + if (rateLimit?.reserved) await releaseDMSlot(instagramAccountId); + await releaseWorkspaceDMReservation(automation.workspaceId, usage.periodStart); } - await releaseWorkspaceDMReservation( - automation.workspaceId, - usage.periodStart - ); - await prisma.dmLog.update({ - where: { - automationId_commentId: { - automationId: automation.id, - commentId, - }, - }, + where: { automationId_commentId: { automationId: automation.id, commentId } }, data: { - status: "FAILED", - attempts: job.attemptsMade + 1, - errorMessage: formatError(error), - dmDeliveryUnconfirmed: error instanceof ZernioDeliveryUnconfirmedError, + status: delivered ? "SENT" : "FAILED", + ...(delivered ? { dmSentAt: new Date() } : {}), + errorMessage: formatError(sendError), + dmDeliveryUnconfirmed: isDeliveryUnconfirmed(sendError), }, }); - throw error; + throw sendError; } } } @@ -749,13 +747,9 @@ async function sendPostbackOnce({ operationId, send, }: { - operationId: string | null; + operationId: string; send: () => Promise; }): Promise { - if (!operationId) { - await send(); - return true; - } try { await prisma.postbackDelivery.create({ data: { id: operationId } }); } catch (error) { @@ -774,17 +768,11 @@ async function sendPostbackOnce({ } catch (error) { // A durable claim survives queue eviction, concurrent redelivery, and a // process crash during delivery. Only confirmed rejections permit retry. - if ( - (error instanceof ZernioApiError && error.code < 500) || - error instanceof RateLimitError || - error instanceof TokenExpiredError - ) { + if (isConfirmedSendRejection(error)) { await prisma.postbackDelivery.delete({ where: { id: operationId } }); throw error; } - throw error instanceof ZernioDeliveryUnconfirmedError - ? error - : new ZernioDeliveryUnconfirmedError(); + throw classifySendError(error); } } @@ -859,19 +847,14 @@ async function processPostback(job: Job): Promise { return; } - const operationId = - accessToken.provider === "ZERNIO" - ? createHash("sha256") - .update( - JSON.stringify([ - automation.instagramAccountId, - automation.id, - userId, - job.data.mid ?? job.id ?? payload, - ]), - ) - .digest("hex") - : null; + const operationId = createHash("sha256") + .update(JSON.stringify([ + automation.instagramAccountId, + automation.id, + userId, + job.data.mid ?? job.id ?? payload, + ])) + .digest("hex"); // Follow-gate: before revealing the link, verify the user follows. On a // `followcheck:` tap a non-follower gets the prompt again (no quota spent); @@ -1002,8 +985,9 @@ async function processPostback(job: Job): Promise { }, update: { status: "SENT", dmSentAt: new Date(), errorMessage: null }, }); - } catch (error) { - await releaseWorkspaceDMReservation( + } catch (originalError) { + const error = classifySendError(originalError); + if (isConfirmedSendRejection(error)) await releaseWorkspaceDMReservation( automation.workspaceId, usage.periodStart, ); @@ -1015,7 +999,7 @@ async function processPostback(job: Job): Promise { // failure the user can act on — so don't log it as FAILED and don't retry // it against a window that cannot reopen on its own. It still delivers in // the case that does work: the user replied by typing instead of tapping. - if (fallback && !(error instanceof ZernioDeliveryUnconfirmedError)) { + if (fallback && !isDeliveryUnconfirmed(error)) { console.log( "[DM Worker] Read fallback not delivered (messaging window closed):", formatError(error), @@ -1040,12 +1024,12 @@ async function processPostback(job: Job): Promise { commentId: dedupeId, status: "FAILED", errorMessage: formatError(error), - dmDeliveryUnconfirmed: error instanceof ZernioDeliveryUnconfirmedError, + dmDeliveryUnconfirmed: isDeliveryUnconfirmed(error), }, update: { status: "FAILED", errorMessage: formatError(error), - dmDeliveryUnconfirmed: error instanceof ZernioDeliveryUnconfirmedError, + dmDeliveryUnconfirmed: isDeliveryUnconfirmed(error), }, }); throw error; @@ -1354,13 +1338,13 @@ async function processMessage(job: Job): Promise { status: "FAILED", attempts: job.attemptsMade + 1, errorMessage: formatError(error), - dmDeliveryUnconfirmed: error instanceof ZernioDeliveryUnconfirmedError, + dmDeliveryUnconfirmed: isDeliveryUnconfirmed(error), }, update: { status: "FAILED", attempts: job.attemptsMade + 1, errorMessage: formatError(error), - dmDeliveryUnconfirmed: error instanceof ZernioDeliveryUnconfirmedError, + dmDeliveryUnconfirmed: isDeliveryUnconfirmed(error), }, }); throw error; @@ -1385,7 +1369,7 @@ async function processJob(job: Job): Promise { try { await dispatchJob(job); } catch (error) { - if (error instanceof ZernioDeliveryUnconfirmedError) + if (isDeliveryUnconfirmed(error)) throw new UnrecoverableError(error.message); throw error; }