feat: durable delivery stages and follow-up snapshots (package 2)

__tests__/campaign-duplicate.test.ts

@@ -2,10 +2,13 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const { mockPrisma } = vi.hoisted(() => ({
   mockPrisma: {
+    $transaction: vi.fn(),
     automation: {
       findFirst: vi.fn(),
       create: vi.fn(),
     },
+    trackedLink: { findMany: vi.fn() },
+    campaignRevision: { create: vi.fn() },
   },
 }));
 
@@ -88,11 +91,13 @@ function createArgs() {
 
 beforeEach(() => {
   vi.clearAllMocks();
+  mockPrisma.$transaction.mockImplementation((callback) => callback(mockPrisma));
+  mockPrisma.trackedLink.findMany.mockResolvedValue([]);
   mockPrisma.automation.findFirst.mockResolvedValue(sourceCampaign);
   mockPrisma.automation.create.mockImplementation(
     async ({ data }: { data: Record<string, unknown> }) => ({
-      id: "automation_copy",
       ...data,
+      id: "automation_copy",
     })
   );
 });

__tests__/campaign-history.test.ts

@@ -0,0 +1,27 @@
+import { describe, expect, it } from "vitest";
+import { safeDeliveryError, safeRevisionSnapshot } from "@/lib/campaigns/history";
+describe("campaign audit projections", () => {
+  it("whitelists snapshots and link fields, never internal access state", () => {
+    expect(safeRevisionSnapshot({ name: "Campaign", version: 7, dmMessage: "Content", accessToken: "secret", workspaceId: "private", trackedLinks: [{ label: "Read", destinationUrl: "https://example.com", position: 0, slug: "private", token: "secret" }] })).toEqual({ name: "Campaign", version: 7, dmMessage: "Content", trackedLinks: [{ label: "Read", destinationUrl: "https://example.com", position: 0 }] });
+    expect(safeRevisionSnapshot(null)).toEqual({});
+    expect(safeRevisionSnapshot([])).toEqual({});
+  });
+  it("redacts common provider credential forms and bounds diagnostics", () => {
+    const error = safeDeliveryError('https://example.com?access_token=abc&key=xyz Bearer bearer123 client_secret="secret123", password=pwd');
+    for (const value of ["abc", "xyz", "bearer123", "secret123", "pwd"]) expect(error).not.toContain(value);
+    expect(safeDeliveryError(null)).toBeNull();
+    expect(safeDeliveryError("x".repeat(2000))).toHaveLength(1000);
+  });
+  it("projects only saved legacy destinations without inventing a version", () => {
+    expect(safeRevisionSnapshot({ lifecycle: "DRAFT", trackedDestinationUrl: "https://example.com/old", secondaryDestinationUrl: "https://example.com/second", secondaryButtonLabel: "Second", accessToken: "discard" })).toEqual({
+      lifecycle: "DRAFT", trackedLinks: [
+        { label: null, destinationUrl: "https://example.com/old", position: 0 },
+        { label: "Second", destinationUrl: "https://example.com/second", position: 1 },
+      ],
+    });
+    expect(safeRevisionSnapshot({ trackedDestinationUrl: null, secondaryDestinationUrl: 123 })).toEqual({});
+  });
+  it("keeps recorded link arrays authoritative, including an explicitly empty array", () => {
+    expect(safeRevisionSnapshot({ version: 1, trackedLinks: [], trackedDestinationUrl: "https://example.com/legacy" })).toEqual({ version: 1, trackedLinks: [] });
+  });
+});

__tests__/campaign-selection.test.ts

@@ -0,0 +1,75 @@
+import { describe, expect, it, vi } from "vitest";
+import { campaignActivationErrors, campaignCanSend, selectAutomation, type CampaignRule } from "@/lib/campaigns/selection";
+import { assertNextReelAvailable, resolveLifecycle } from "@/lib/campaigns/mutations";
+import type { Prisma } from "@/app/generated/prisma/client";
+
+function campaign(overrides: Partial<CampaignRule> = {}): CampaignRule {
+  return { id: "campaign", lifecycle: "ACTIVE", isActive: true, createdAt: new Date("2026-01-01"), keywords: ["link"], matchAnyWord: false, wholeWordMatch: true, matchAnyPost: true, dmTriggerEnabled: true, ...overrides };
+}
+const comment = { kind: "comment" as const, text: "LINK please", mediaIds: ["post"] };
+describe("deterministic campaign selection", () => {
+  it("prefers a specific post over an older global even with higher priority", () => {
+    const global = campaign({ id: "global", priority: 100 });
+    const specific = campaign({ id: "specific", matchAnyPost: false, postId: "post", priority: -100 });
+    expect(selectAutomation([global, specific], comment).winner?.id).toBe("specific");
+  });
+  it("uses priority then created/id; surfaces semantic ties independent of order", () => {
+    const a = campaign({ id: "a", priority: 10 });
+    const b = campaign({ id: "b", priority: 10 });
+    const c = campaign({ id: "c", priority: 9 });
+    expect(selectAutomation([c, b, a], comment).winner?.id).toBe("a");
+    expect(selectAutomation([a, b, c], comment).conflicts.map((c) => c.id)).toEqual(["b"]);
+  });
+  it("never sends drafts, archives, paused or pending next reels", () => {
+    for (const lifecycle of ["DRAFT", "ARCHIVED", "PAUSED"]) expect(campaignCanSend(campaign({ lifecycle }))).toBe(false);
+    expect(selectAutomation([campaign({ pendingNextReel: true })], comment).winner).toBeNull();
+    expect(selectAutomation([campaign({ isActive: false })], comment).winner).toBeNull();
+  });
+  it("does not match a specific campaign to a different post", () => {
+    expect(selectAutomation([campaign({ matchAnyPost: false, postId: "other" })], comment).winner).toBeNull();
+  });
+  it("uses provider-normalized whole word rules for exclusions", () => {
+    expect(selectAutomation([campaign({ excludedKeywords: ["nein"] })], { ...comment, text: "LINK aber nein" }).winner).toBeNull();
+    expect(selectAutomation([campaign({ excludedKeywords: ["no"] })], { ...comment, text: "link november" }).winner).not.toBeNull();
+    expect(selectAutomation([campaign({ excludedKeywords: ["no"], wholeWordMatch: false })], { ...comment, text: "link november" }).winner).toBeNull();
+  });
+  it("has separate DM ranking without post specificity; matches only DM-enabled campaigns", () => {
+    const specific = campaign({ id: "specific", matchAnyPost: false, postId: "post", priority: 0 });
+    const global = campaign({ id: "global", priority: 1 });
+    expect(selectAutomation([specific, global], { kind: "dm", text: "link" }).winner?.id).toBe("global");
+    expect(selectAutomation([campaign({ dmTriggerEnabled: false })], { kind: "dm", text: "link" }).winner).toBeNull();
+  });
+  it("retains matched keywords and ignores empty any-word input", () => {
+    expect(selectAutomation([campaign()], comment).matchedKeyword).toBe("link");
+    expect(selectAutomation([campaign({ matchAnyWord: true })], { ...comment, text: "  " }).winner).toBeNull();
+  });
+  it("supports legacy rows without lifecycle", () => {
+    expect(campaignCanSend(campaign({ lifecycle: undefined }))).toBe(true);
+    expect(resolveLifecycle({}, { lifecycle: "PAUSED", isActive: false })).toBe("PAUSED");
+    expect(resolveLifecycle({ isActive: false }, { lifecycle: "ACTIVE", isActive: true })).toBe("PAUSED");
+    expect(resolveLifecycle({ lifecycle: "ARCHIVED", isActive: true })).toBe("ARCHIVED");
+  });
+});
+describe("activation safety", () => {
+  it("rejects incomplete content and accepts complete final content", () => {
+    expect(campaignActivationErrors({ keywords: [], matchAnyWord: false, dmMessage: "" })).toHaveLength(3);
+    expect(campaignActivationErrors({ postId: "post", keywords: ["link"], matchAnyWord: false, dmMessage: "resource" })).toEqual([]);
+    expect(campaignActivationErrors({ matchAnyPost: true, pendingNextReel: true, keywords: ["link"], matchAnyWord: false, dmMessage: "resource" })).toContain("Choose exactly one post scope.");
+  });
+  it("locks the account before checking another armed campaign", async () => {
+    const execute = vi.fn().mockResolvedValue(1);
+    const findFirst = vi.fn().mockResolvedValue({ name: "Other reel" });
+    const tx = { $executeRaw: execute, automation: { findFirst } } as unknown as Prisma.TransactionClient;
+    await expect(assertNextReelAvailable(tx, "account", true, "campaign")).rejects.toThrow("Other reel");
+    expect(execute.mock.calls[0][1]).toBe("next-reel:account");
+    expect(execute.mock.invocationCallOrder[0]).toBeLessThan(findFirst.mock.invocationCallOrder[0]);
+    expect(findFirst.mock.calls[0][0].where).toMatchObject({ lifecycle: "ACTIVE", isActive: true, pendingNextReel: true, id: { not: "campaign" } });
+  });
+  it("always locks but does not reject saving an unarmed draft", async () => {
+    const execute = vi.fn().mockResolvedValue(1);
+    const findFirst = vi.fn();
+    await assertNextReelAvailable({ $executeRaw: execute, automation: { findFirst } } as unknown as Prisma.TransactionClient, "account", false);
+    expect(execute).toHaveBeenCalledOnce();
+    expect(findFirst).not.toHaveBeenCalled();
+  });
+});

__tests__/campaign-simulation.test.ts

@@ -0,0 +1,31 @@
+import { describe, expect, it, vi } from "vitest";
+import { NextRequest } from "next/server";
+vi.mock("@/lib/auth", () => ({ getCurrentWorkspaceId: async () => "test-workspace" }));
+vi.mock("@/lib/db/client", () => ({ prisma: {} }));
+import { POST } from "@/app/api/automations/validate/route";
+
+describe("simulation body validation", () => {
+  it("rejects malformed JSON without an unhandled server exception", async () => {
+    const response = await POST(new NextRequest("http://localhost/api/automations/validate", {
+      method: "POST", headers: { "Content-Type": "application/json" }, body: "{",
+    }));
+    expect(response.status).toBe(400);
+    expect(await response.json()).toMatchObject({ success: false, error: "Invalid JSON" });
+  });
+  it("bounds streamed bodies independently of content-length", async () => {
+    const response = await POST(new NextRequest("http://localhost/api/automations/validate", {
+      method: "POST", headers: { "Content-Type": "application/json", "Content-Length": "1" }, body: JSON.stringify({ text: "a".repeat(70_000) }),
+    }));
+    expect(response.status).toBe(413);
+  });
+  it("requires a JSON content type", async () => {
+    const response = await POST(new NextRequest("http://localhost/api/automations/validate", { method: "POST", body: "{}" }));
+    expect(response.status).toBe(415);
+  });
+  it("rejects foreign browser origins before accessing campaign data", async () => {
+    const response = await POST(new NextRequest("http://localhost/api/automations/validate", {
+      method: "POST", headers: { "Content-Type": "application/json", Origin: "https://foreign.example.test" }, body: "{}",
+    }));
+    expect(response.status).toBe(403);
+  });
+});

__tests__/comment-delivery.test.ts

@@ -15,6 +15,8 @@ vi.mock("@/lib/db/client", () => ({
   },
 }));
 import { claimCommentDelivery } from "../lib/queue/comment-delivery";
+import { selectTriggerWinner } from "../lib/queue/trigger-selection";
+import { scheduleFollowUp } from "../lib/queue/follow-up-delivery";
 
 const databaseUrl = process.env.TEST_DATABASE_URL;
 const schema = `delivery_claims_${randomBytes(4).toString("hex")}`;
@@ -127,6 +129,84 @@ describe.skipIf(!databaseUrl)("durable comment delivery on Postgres", () => {
         .attempts,
     ).toBe(3);
   });
+  it("atomically creates one stage event with the persistent send claim", async () => {
+    await seed("stage-claim");
+    const delivery = { workspaceId: "workspace", instagramAccountId: "account", version: 3,
+      stage: "REVEAL", recipientId: "user_ig", message: "Frozen message" };
+    const results = await Promise.all(Array.from({ length: 8 }, () =>
+      claimCommentDelivery("automation", "stage-claim", "dm", delivery)));
+    expect(results.filter(Boolean)).toHaveLength(1);
+    const events = await state.db.deliveryEvent.findMany({ where: { stage: "REVEAL" } });
+    expect(events).toHaveLength(1);
+    expect(events[0]).toMatchObject({ status: "CLAIMED", attempts: 1, campaignVersion: 3, message: "Frozen message" });
+    expect(await claimCommentDelivery("automation", "stage-claim", "dm", delivery)).toBe(false);
+  });
+  it("a paused campaign cannot claim a new durable stage", async () => {
+    await seed("paused-stage");
+    await state.db.automation.update({ where: { id: "automation" }, data: { isActive: false, lifecycle: "PAUSED" } });
+    try {
+      expect(await claimCommentDelivery("automation", "paused-stage", "dm", {
+        workspaceId: "workspace", instagramAccountId: "account", version: 1,
+        stage: "OPENING_DM", recipientId: "user_ig", message: "Never send" })).toBe(false);
+      expect(await state.db.deliveryEvent.count({ where: { stage: "OPENING_DM" } })).toBe(0);
+    } finally {
+      await state.db.automation.update({ where: { id: "automation" }, data: { isActive: true, lifecycle: "ACTIVE" } });
+    }
+  });
+  it("freezes a trigger winner and content across concurrent jobs and edits without falling through", async () => {
+    await state.db.automation.update({ where: { id: "automation" },
+      data: { matchAnyPost: true, dmMessage: "Original delivery", priority: 10 } });
+    await state.db.automation.create({ data: { id: "lower-priority", workspaceId: "workspace",
+      instagramAccountId: "account", name: "Lower priority", keywords: ["AI"],
+      dmMessage: "Never fall through", matchAnyPost: true, priority: 0 } });
+    const candidates = () => state.db.automation.findMany({
+      where: { id: { in: ["automation", "lower-priority"] } }, include: { instagramAccount: true, trackedLinks: true } });
+    const input = { kind: "comment" as const, text: "AI", inputId: "frozen-trigger", instagramId: "test_ig" };
+    try {
+      const campaigns = await candidates();
+      const winners = await Promise.all(Array.from({ length: 8 }, () => selectTriggerWinner(campaigns, input)));
+      expect(winners.every((winner) => winner?.id === "automation")).toBe(true);
+      const reservation = await state.db.deliveryEvent.findFirstOrThrow({ where: { stage: "TRIGGER" } });
+      expect(await state.db.deliveryEvent.count({ where: { stage: "TRIGGER" } })).toBe(1);
+      expect(JSON.stringify(reservation.payload)).not.toContain("local-test-only");
+      await state.db.automation.update({ where: { id: "automation" }, data: { dmMessage: "Edited delivery", version: 2 } });
+      expect(await selectTriggerWinner(await candidates(), input)).toMatchObject({ id: "automation", version: 1, dmMessage: "Original delivery" });
+      await state.db.automation.update({ where: { id: "automation" }, data: { isActive: false, lifecycle: "PAUSED" } });
+      expect(await selectTriggerWinner(await candidates(), input)).toBeNull();
+    } finally {
+      await state.db.automation.update({ where: { id: "automation" }, data: { isActive: true, lifecycle: "ACTIVE" } });
+    }
+  });
+  it("persists one immutable follow-up across parallel result transactions", async () => {
+    const campaign = await state.db.automation.update({ where: { id: "automation" },
+      data: { followUpEnabled: true, followUpMessage: "Thanks {username}", followUpDelayMinutes: 5 },
+      include: { instagramAccount: true } });
+    const interactionAt = new Date();
+    await Promise.all(Array.from({ length: 8 }, () => state.db.$transaction((tx) =>
+      scheduleFollowUp(campaign, { userId: "user_ig", commenterName: "Example", sourceId: "parallel-tap", interactionAt }, tx))));
+    expect(await state.db.deliveryEvent.count({ where: { stage: "FOLLOW_UP" } })).toBe(1);
+    const event = await state.db.deliveryEvent.findFirstOrThrow({ where: { stage: "FOLLOW_UP" } });
+    expect(event).toMatchObject({ status: "PENDING", message: "Thanks Example", recipientId: "user_ig" });
+    await state.db.$transaction((tx) => scheduleFollowUp({ ...campaign, followUpMessage: "Edited text" },
+      { userId: "user_ig", commenterName: "Example", sourceId: "parallel-tap", interactionAt }, tx));
+    expect((await state.db.deliveryEvent.findUniqueOrThrow({ where: { id: event.id } })).message).toBe("Thanks Example");
+  });
+  it("parallel inbound-DM log creation still permits only one pre-send claim", async () => {
+    const results = await Promise.all(Array.from({ length: 16 }, async () => {
+      await state.db.dmLog.upsert({
+        where: { automationId_commentId: { automationId: "automation", commentId: "dm:parallel-message" } },
+        create: { workspaceId: "workspace", instagramAccountId: "account", automationId: "automation",
+          commenterId: "user_ig", commentId: "dm:parallel-message", commentText: "AI", status: "PENDING" },
+        update: { commentId: "dm:parallel-message" },
+      });
+      return claimCommentDelivery("automation", "dm:parallel-message", "dm", {
+        workspaceId: "workspace", instagramAccountId: "account", version: 2,
+        stage: "REVEAL", recipientId: "user_ig", message: "Frozen message" });
+    }));
+    expect(results.filter(Boolean)).toHaveLength(1);
+    expect(await state.db.dmLog.count({ where: { commentId: "dm:parallel-message" } })).toBe(1);
+    expect((await state.db.dmLog.findFirstOrThrow({ where: { commentId: "dm:parallel-message" } })).attempts).toBe(1);
+  });
   it("claims public replies independently and never repeats a confirmed send", async () => {
     const row = await seed("public");
     expect(await claimCommentDelivery("automation", "public", "public")).toBe(