feat(apify): digest rate cap (1 per artist per 24h) + email_send_log audit trail

lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts

@@ -3,13 +3,18 @@ import { maybeSendScrapeDigest } from "@/lib/apify/digest/maybeSendScrapeDigest"
 import { selectApifyScraperRuns } from "@/lib/supabase/apify_scraper_runs/selectApifyScraperRuns";
 import { getScrapeDigestRecipients } from "@/lib/apify/digest/getScrapeDigestRecipients";
 import { sendScrapeDigestEmail } from "@/lib/apify/digest/sendScrapeDigestEmail";
+import { selectScrapeDigestLogs } from "@/lib/supabase/email_send_log/selectScrapeDigestLogs";
+import { logEmailAttempt } from "@/lib/emails/logEmailAttempt";
 
 vi.mock("@/lib/supabase/apify_scraper_runs/selectApifyScraperRuns", () => ({
   selectApifyScraperRuns: vi.fn(),
 }));
 vi.mock("@/lib/apify/digest/getScrapeDigestRecipients", () => ({
   getScrapeDigestRecipients: vi.fn(),
 }));
+vi.mock("@/lib/apify/digest/sendScrapeDigestEmail", () => ({
+  sendScrapeDigestEmail: vi.fn(),
+}));
 vi.mock("@/lib/apify/digest/getRunDigestSection", () => ({
   // URL-only translation of a run row — enrichment is unit-tested separately
   getRunDigestSection: vi.fn(async (run: { platform: string; new_post_urls: string[] | null }) =>
@@ -22,9 +27,10 @@ vi.mock("@/lib/apify/digest/getRunDigestSection", () => ({
       : null,
   ),
 }));
-vi.mock("@/lib/apify/digest/sendScrapeDigestEmail", () => ({
-  sendScrapeDigestEmail: vi.fn(),
+vi.mock("@/lib/supabase/email_send_log/selectScrapeDigestLogs", () => ({
+  selectScrapeDigestLogs: vi.fn(async () => []),
 }));
+vi.mock("@/lib/emails/logEmailAttempt", () => ({ logEmailAttempt: vi.fn(async () => {}) }));
 
 const run = (over: Record<string, unknown>) => ({
   run_id: "r1",
@@ -39,7 +45,11 @@ const run = (over: Record<string, unknown>) => ({
 
 beforeEach(() => {
   vi.clearAllMocks();
-  vi.mocked(getScrapeDigestRecipients).mockResolvedValue(["owner@example.com"]);
+  vi.mocked(selectScrapeDigestLogs).mockResolvedValue([]);
+  vi.mocked(getScrapeDigestRecipients).mockResolvedValue({
+    emails: ["owner@example.com"],
+    artistIds: ["artist-1"],
+  } as never);
   vi.mocked(sendScrapeDigestEmail).mockResolvedValue({ id: "email-1" } as never);
 });
 
@@ -53,7 +63,7 @@ describe("maybeSendScrapeDigest", () => {
     expect(sendScrapeDigestEmail).not.toHaveBeenCalled();
   });
 
-  it("sends ONE digest with per-platform new posts when the batch completes", async () => {
+  it("sends ONE digest with per-platform new posts when the batch completes, and records the audit trail", async () => {
     vi.mocked(selectApifyScraperRuns).mockResolvedValue([
       run({ new_post_urls: ["https://instagram.com/p/1"] }),
       run({ run_id: "r2", platform: "tiktok", new_post_urls: ["https://tiktok.com/v/2"] }),
@@ -71,16 +81,49 @@ describe("maybeSendScrapeDigest", () => {
       { platform: "tiktok", posts: [{ url: "https://tiktok.com/v/2" }], artistName: null },
     ]); // x omitted — nothing new
     expect(arg.emails).toEqual(["owner@example.com"]);
-    expect(arg.artistName).toBe("National Geographic"); // digest addressed by artist, not "Your artist"
+    expect(arg.artistName).toBe("National Geographic");
+    // audit trail: the send is recorded per watched artist entity
+    expect(logEmailAttempt).toHaveBeenCalledWith(
+      expect.objectContaining({
+        rawBody: "scrape-digest:b1",
+        accountId: "artist-1",
+        status: "sent",
+        resendId: "email-1",
+      }),
+    );
+  });
+
+  it("never sends twice for the same batch — exact idempotency, not a time window", async () => {
+    vi.mocked(selectApifyScraperRuns).mockResolvedValue([
+      run({ new_post_urls: ["https://instagram.com/p/1"] }),
+    ] as never);
+    vi.mocked(selectScrapeDigestLogs).mockImplementation(async ({ batchId }: never) =>
+      batchId ? ([{ account_id: "artist-1", created_at: "2020-01-01T00:00:00Z" }] as never) : [],
+    );
+    expect(await maybeSendScrapeDigest("b1")).toBeNull();
+    expect(sendScrapeDigestEmail).not.toHaveBeenCalled();
+    expect(selectScrapeDigestLogs).toHaveBeenCalledWith({ batchId: "b1" });
+  });
+
+  it("suppresses a NEW batch when any watching artist got a digest within 24h (rate cap)", async () => {
+    vi.mocked(selectApifyScraperRuns).mockResolvedValue([
+      run({ new_post_urls: ["https://instagram.com/p/1"] }),
+    ] as never);
+    vi.mocked(selectScrapeDigestLogs).mockImplementation(async ({ artistIds }: never) =>
+      artistIds ? ([{ account_id: "artist-1", created_at: "2026-07-07T01:00:00Z" }] as never) : [],
+    );
+    expect(await maybeSendScrapeDigest("b1")).toBeNull();
+    expect(sendScrapeDigestEmail).not.toHaveBeenCalled();
   });
 
-  it("sends nothing when the batch completes with zero new posts anywhere", async () => {
+  it("sends nothing and logs nothing when the batch completes with zero new posts anywhere", async () => {
     vi.mocked(selectApifyScraperRuns).mockResolvedValue([
       run({}),
       run({ run_id: "r2", platform: "tiktok" }),
     ] as never);
     expect(await maybeSendScrapeDigest("b1")).toBeNull();
     expect(sendScrapeDigestEmail).not.toHaveBeenCalled();
+    expect(logEmailAttempt).not.toHaveBeenCalled();
   });
 
   it("no-ops for a null batch id (legacy runs)", async () => {

lib/apify/digest/getScrapeDigestRecipients.ts

@@ -7,9 +7,11 @@ import selectAccountEmails from "@/lib/supabase/account_emails/selectAccountEmai
  * of every artist watching any of them (same chain the per-platform alert
  * used). Recipients span tenants — senders must BCC (chat#1855).
  */
-export async function getScrapeDigestRecipients(socialIds: string[]): Promise<string[]> {
+export async function getScrapeDigestRecipients(
+  socialIds: string[],
+): Promise<{ emails: string[]; artistIds: string[] }> {
   const uniqueSocialIds = Array.from(new Set(socialIds.filter(Boolean)));
-  if (!uniqueSocialIds.length) return [];
+  if (!uniqueSocialIds.length) return { emails: [], artistIds: [] };
 
   const accountSocials = (
     await Promise.all(
@@ -24,5 +26,8 @@ export async function getScrapeDigestRecipients(socialIds: string[]): Promise<st
     new Set(accountArtistIds.map(a => a.account_id).filter((id): id is string => Boolean(id))),
   );
   const accountEmails = await selectAccountEmails({ accountIds: uniqueAccountIds });
-  return Array.from(new Set(accountEmails.map(e => e.email).filter(Boolean)));
+  return {
+    emails: Array.from(new Set(accountEmails.map(e => e.email).filter(Boolean))),
+    artistIds: Array.from(new Set(accountSocials.map(a => a.account_id).filter(Boolean))),
+  };
 }

lib/apify/digest/maybeSendScrapeDigest.ts

@@ -3,13 +3,18 @@ import { getRunDigestSection } from "@/lib/apify/digest/getRunDigestSection";
 import type { RunDigestSection } from "@/lib/apify/digest/getRunDigestSection";
 import { getScrapeDigestRecipients } from "@/lib/apify/digest/getScrapeDigestRecipients";
 import { sendScrapeDigestEmail } from "@/lib/apify/digest/sendScrapeDigestEmail";
+import { selectScrapeDigestLogs } from "@/lib/supabase/email_send_log/selectScrapeDigestLogs";
+import { logEmailAttempt } from "@/lib/emails/logEmailAttempt";
 
 /**
  * Batch-completion check for the one-digest-per-scrape design (chat#1855):
  * called after each platform run's results are processed. Sends the single
- * consolidated digest when (a) every sibling run in the batch has completed
- * and (b) at least one platform found genuinely new posts. Platforms with
- * nothing new are omitted; a batch with nothing new sends nothing.
+ * consolidated digest when (a) every sibling run in the batch has completed,
+ * (b) at least one platform found genuinely new posts, (c) this batch has
+ * never sent before (exact idempotency via its `scrape-digest:<batchId>`
+ * audit row — a webhook retry must not duplicate the send), and (d) no
+ * digest went to any watching artist in the last 24h (product frequency
+ * cap). Every send is recorded in email_send_log per watched artist.
  */
 export async function maybeSendScrapeDigest(batchId: string | null | undefined) {
   if (!batchId) return null;
@@ -22,10 +27,35 @@ export async function maybeSendScrapeDigest(batchId: string | null | undefined)
   );
   if (!sections.length) return null;
 
-  const emails = await getScrapeDigestRecipients(
+  const { emails, artistIds } = await getScrapeDigestRecipients(
     runs.map(r => r.social_id).filter((id): id is string => Boolean(id)),
   );
-  // A batch is one artist's scrape — any platform's profile name addresses it.
+
+  // Exact non-duplication: one digest per batch, ever.
+  const already = await selectScrapeDigestLogs({ batchId });
+  if (already.length > 0) return null;
+
+  // Product frequency cap: at most one digest per watched artist per 24h.
+  const since = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString();
+  const recent = await selectScrapeDigestLogs({ artistIds, since });
+  if (recent.length > 0) return null;
+
   const artistName = sections.map(s => s.artistName).find(Boolean) ?? null;
-  return await sendScrapeDigestEmail({ emails, sections, artistName });
+  const sent = await sendScrapeDigestEmail({ emails, sections, artistName });
+  if (!sent) return null;
+
+  // Audit trail: record the send per watched artist entity — what the batch
+  // dedup and the 24h cap read (the legacy solo path never logged at all).
+  const resendId = (sent as { id?: string }).id;
+  await Promise.all(
+    artistIds.map(artistId =>
+      logEmailAttempt({
+        rawBody: `scrape-digest:${batchId}`,
+        status: "sent",
+        accountId: artistId,
+        resendId,
+      }),
+    ),
+  );
+  return sent;
 }

lib/supabase/apify_scraper_runs/__tests__/updateApifyScraperRun.test.ts

@@ -0,0 +1,36 @@
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { updateApifyScraperRun } from "@/lib/supabase/apify_scraper_runs/updateApifyScraperRun";
+
+const maybeSingle = vi.fn();
+const chain = {
+  update: vi.fn(() => chain),
+  eq: vi.fn(() => chain),
+  is: vi.fn(() => chain),
+  select: vi.fn(() => chain),
+  maybeSingle,
+};
+vi.mock("@/lib/supabase/serverClient", () => ({ default: { from: vi.fn(() => chain) } }));
+
+beforeEach(() => {
+  vi.clearAllMocks();
+});
+
+describe("updateApifyScraperRun", () => {
+  it("claims completion only for runs not yet completed — the idempotency guard", async () => {
+    maybeSingle.mockResolvedValue({ data: { run_id: "r1", batch_id: "b1" }, error: null });
+    const row = await updateApifyScraperRun("r1", ["https://x.com/a/status/1"]);
+    expect(chain.eq).toHaveBeenCalledWith("run_id", "r1");
+    expect(chain.is).toHaveBeenCalledWith("completed_at", null);
+    expect(row).toEqual({ run_id: "r1", batch_id: "b1" });
+  });
+
+  it("returns null on a webhook retry (row already completed) so the digest can never re-trigger", async () => {
+    maybeSingle.mockResolvedValue({ data: null, error: null });
+    expect(await updateApifyScraperRun("r1", [])).toBeNull();
+  });
+
+  it("returns null and logs on error", async () => {
+    maybeSingle.mockResolvedValue({ data: null, error: { message: "boom" } });
+    expect(await updateApifyScraperRun("r1", [])).toBeNull();
+  });
+});

lib/supabase/apify_scraper_runs/updateApifyScraperRun.ts

@@ -2,9 +2,11 @@ import supabase from "@/lib/supabase/serverClient";
 import type { Tables } from "@/types/database.types";
 
 /**
- * Marks a scrape run's results as processed and records which post URLs were
- * genuinely new. Returns the updated row (carrying batch_id) or null when the
- * run was never registered (legacy/non-batch runs).
+ * Claims a scrape run's completion: marks results processed and records which
+ * post URLs were genuinely new. The `completed_at IS NULL` guard makes the
+ * claim idempotent — an Apify webhook retry for an already-completed run
+ * returns null (so it can never re-trigger digest assembly or overwrite the
+ * recorded diff) exactly like a never-registered legacy run.
  */
 export async function updateApifyScraperRun(
   runId: string,
@@ -14,6 +16,7 @@ export async function updateApifyScraperRun(
     .from("apify_scraper_runs")
     .update({ completed_at: new Date().toISOString(), new_post_urls: newPostUrls })
     .eq("run_id", runId)
+    .is("completed_at", null)
     .select()
     .maybeSingle();
   if (error) {

lib/supabase/email_send_log/selectScrapeDigestLogs.ts

@@ -0,0 +1,37 @@
+import supabase from "@/lib/supabase/serverClient";
+
+/**
+ * Select scrape-digest send-log rows with optional filters.
+ *
+ * @param batchId - Rows recording exactly this batch's send (idempotency check).
+ * @param artistIds - Rows for any of these watched artist entities.
+ * @param since - Only rows created at or after this ISO timestamp (rate-cap lookback).
+ */
+export async function selectScrapeDigestLogs({
+  batchId,
+  artistIds,
+  since,
+}: {
+  batchId?: string;
+  artistIds?: string[];
+  since?: string;
+} = {}) {
+  if (artistIds !== undefined && artistIds.length === 0) return [];
+
+  let query = supabase.from("email_send_log").select("account_id, created_at");
+
+  if (batchId) {
+    query = query.eq("raw_body", `scrape-digest:${batchId}`);
+  } else {
+    query = query.like("raw_body", "scrape-digest:%");
+  }
+  if (artistIds) query = query.in("account_id", artistIds);
+  if (since) query = query.gte("created_at", since);
+
+  const { data, error } = await query;
+  if (error) {
+    console.error("[ERROR] selectScrapeDigestLogs:", error);
+    return [];
+  }
+  return data ?? [];
+}