diff --git a/lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts b/lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts index 0608e53d..52082107 100644 --- a/lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts +++ b/lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts @@ -3,6 +3,8 @@ 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(), @@ -10,6 +12,9 @@ vi.mock("@/lib/supabase/apify_scraper_runs/selectApifyScraperRuns", () => ({ 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) => ({ run_id: "r1", @@ -39,7 +45,11 @@ const run = (over: Record) => ({ 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 () => { diff --git a/lib/apify/digest/getScrapeDigestRecipients.ts b/lib/apify/digest/getScrapeDigestRecipients.ts index a3c31cd9..56024982 100644 --- a/lib/apify/digest/getScrapeDigestRecipients.ts +++ b/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 { +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 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))), + }; } diff --git a/lib/apify/digest/maybeSendScrapeDigest.ts b/lib/apify/digest/maybeSendScrapeDigest.ts index c68b5179..45de19f3 100644 --- a/lib/apify/digest/maybeSendScrapeDigest.ts +++ b/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:` + * 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; } diff --git a/lib/supabase/apify_scraper_runs/__tests__/updateApifyScraperRun.test.ts b/lib/supabase/apify_scraper_runs/__tests__/updateApifyScraperRun.test.ts new file mode 100644 index 00000000..ed1d2a16 --- /dev/null +++ b/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(); + }); +}); diff --git a/lib/supabase/apify_scraper_runs/updateApifyScraperRun.ts b/lib/supabase/apify_scraper_runs/updateApifyScraperRun.ts index bf217a92..2cc347aa 100644 --- a/lib/supabase/apify_scraper_runs/updateApifyScraperRun.ts +++ b/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) { diff --git a/lib/supabase/email_send_log/selectScrapeDigestLogs.ts b/lib/supabase/email_send_log/selectScrapeDigestLogs.ts new file mode 100644 index 00000000..542cec8a --- /dev/null +++ b/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 ?? []; +}