From 360268b97baacbe56934fe3268b63c24672da905 Mon Sep 17 00:00:00 2001 From: Sweets Sweetman Date: Mon, 6 Jul 2026 16:55:39 -0500 Subject: [PATCH 1/4] =?UTF-8?q?feat(apify):=20notify=20only=20on=20genuine?= =?UTF-8?q?ly=20new=20posts=20=E2=80=94=20diff=20against=20stored=20posts?= =?UTF-8?q?=20first?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Instagram scrape alert fired whenever a scrape returned posts, with no comparison against posts already stored — every scrape re-announced the profile's recent feed as new (observed: 6 of 7 alerts in one day announced posts up to 10 days old). New reusable filterNewPostUrls diffs candidate URLs against posts BEFORE upsert; the alert is gated on a non-empty result. Persistence unchanged (recoupable/chat#1855, PR #3 of 6). Co-Authored-By: Claude Fable 5 --- ...ndleInstagramProfileScraperResults.test.ts | 38 ++++++++++++++++++- .../handleInstagramProfileScraperResults.ts | 16 ++++++-- .../__tests__/filterNewPostUrls.test.ts | 25 ++++++++++++ lib/socials/filterNewPostUrls.ts | 21 ++++++++++ 4 files changed, 95 insertions(+), 5 deletions(-) create mode 100644 lib/socials/__tests__/filterNewPostUrls.test.ts create mode 100644 lib/socials/filterNewPostUrls.ts diff --git a/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts b/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts index 9092a197b..bab22c438 100644 --- a/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts +++ b/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts @@ -12,6 +12,7 @@ import { selectAccountSocials } from "@/lib/supabase/account_socials/selectAccou import { getAccountArtistIds } from "@/lib/supabase/account_artist_ids/getAccountArtistIds"; import selectAccountEmails from "@/lib/supabase/account_emails/selectAccountEmails"; import { uploadLinkToArweave } from "@/lib/arweave/uploadLinkToArweave"; +import { filterNewPostUrls } from "@/lib/socials/filterNewPostUrls"; vi.mock("@/lib/apify/client", () => ({ default: { dataset: vi.fn() } })); @@ -26,6 +27,7 @@ vi.mock("../handleInstagramProfileFollowUpRuns", () => ({ handleInstagramProfileFollowUpRuns: vi.fn(), })); vi.mock("@/lib/apify/sendApifyWebhookEmail", () => ({ sendApifyWebhookEmail: vi.fn() })); +vi.mock("@/lib/socials/filterNewPostUrls", () => ({ filterNewPostUrls: vi.fn() })); vi.mock("@/lib/supabase/socials/upsertSocials", () => ({ upsertSocials: vi.fn() })); vi.mock("@/lib/supabase/socials/selectSocials", () => ({ selectSocials: vi.fn(), @@ -51,7 +53,11 @@ const payload = { } as never; describe("handleInstagramProfileScraperResults", () => { - beforeEach(() => vi.clearAllMocks()); + beforeEach(() => { + vi.clearAllMocks(); + // default: everything scraped counts as new (per-test overrides below) + vi.mocked(filterNewPostUrls).mockImplementation(async urls => urls); + }); it("short-circuits when the dataset has no latest posts", async () => { mockDataset([{ username: "alice" }]); @@ -92,4 +98,34 @@ describe("handleInstagramProfileScraperResults", () => { expect(result.social).toEqual({ id: "s1" }); expect(result.posts).toEqual(posts); }); + + it("skips the alert email when no scraped post is genuinely new (chat#1855)", async () => { + mockDataset([ + { + latestPosts: [{ url: "u1", timestamp: "t" }], + username: "alice", + url: "instagram.com/alice", + profilePicUrl: "https://a", + fullName: "Alice", + }, + ]); + vi.mocked(filterNewPostUrls).mockResolvedValue([]); // all already stored + vi.mocked(upsertPosts).mockResolvedValue({ data: null, error: null } as never); + vi.mocked(getPosts).mockResolvedValue([{ id: "p1", post_url: "u1" }] as never); + vi.mocked(uploadLinkToArweave).mockResolvedValue(null); + vi.mocked(upsertSocials).mockResolvedValue([] as never); + vi.mocked(selectSocials).mockResolvedValue([{ id: "s1" }] as never); + vi.mocked(selectAccountSocials).mockResolvedValue([{ account_id: "a1" }] as never); + vi.mocked(getAccountArtistIds).mockResolvedValue([{ account_id: "a1" }] as never); + vi.mocked(selectAccountEmails).mockResolvedValue([{ email: "x@y.com" }] as never); + + const result = await handleInstagramProfileScraperResults(payload); + + expect(sendApifyWebhookEmail).not.toHaveBeenCalled(); + // persistence is unaffected — only the notification is gated + expect(upsertPosts).toHaveBeenCalledOnce(); + expect(upsertSocialPosts).toHaveBeenCalledOnce(); + expect(handleInstagramProfileFollowUpRuns).toHaveBeenCalledOnce(); + expect(result.social).toEqual({ id: "s1" }); + }); }); diff --git a/lib/apify/instagram/handleInstagramProfileScraperResults.ts b/lib/apify/instagram/handleInstagramProfileScraperResults.ts index ff8a971c5..4774c17f0 100644 --- a/lib/apify/instagram/handleInstagramProfileScraperResults.ts +++ b/lib/apify/instagram/handleInstagramProfileScraperResults.ts @@ -14,6 +14,7 @@ import { uploadLinkToArweave } from "@/lib/arweave/uploadLinkToArweave"; import { getFetchableUrl } from "@/lib/arweave/getFetchableUrl"; import type { ApifyInstagramProfileResult } from "@/lib/apify/types"; import type { ApifyWebhookPayload } from "@/lib/apify/validateApifyWebhookRequest"; +import { filterNewPostUrls } from "@/lib/socials/filterNewPostUrls"; import type { TablesInsert } from "@/types/database.types"; /** @@ -40,6 +41,9 @@ export async function handleInstagramProfileScraperResults(parsed: ApifyWebhookP post.url ? [{ post_url: post.url, updated_at: post.timestamp }] : [], ); if (postRows.length === 0) return { posts: [], social: null }; + // Diff BEFORE upserting — afterwards every scraped post exists and nothing + // is distinguishable as new (chat#1855). + const newPostUrls = await filterNewPostUrls(postRows.map(p => p.post_url)); await upsertPosts(postRows); const posts = await getPosts({ postUrls: postRows.map(p => p.post_url) }); @@ -96,10 +100,14 @@ export async function handleInstagramProfileScraperResults(parsed: ApifyWebhookP // mail outage doesn't block comment scraping and vice versa. let sentEmails = null; try { - sentEmails = await sendApifyWebhookEmail( - firstResult, - accountEmails.map(e => e.email).filter(Boolean), - ); + // Only notify when the scrape actually found posts new to the platform — + // otherwise every scrape re-announces the profile's recent feed. + if (newPostUrls.length > 0) { + sentEmails = await sendApifyWebhookEmail( + firstResult, + accountEmails.map(e => e.email).filter(Boolean), + ); + } } catch (error) { console.error("[WARN] webhook email failed:", error); } diff --git a/lib/socials/__tests__/filterNewPostUrls.test.ts b/lib/socials/__tests__/filterNewPostUrls.test.ts new file mode 100644 index 000000000..0f4be5305 --- /dev/null +++ b/lib/socials/__tests__/filterNewPostUrls.test.ts @@ -0,0 +1,25 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { filterNewPostUrls } from "@/lib/socials/filterNewPostUrls"; +import { getPosts } from "@/lib/supabase/posts/getPosts"; + +vi.mock("@/lib/supabase/posts/getPosts", () => ({ getPosts: vi.fn() })); + +beforeEach(() => vi.clearAllMocks()); + +describe("filterNewPostUrls", () => { + it("returns only URLs not already stored in posts", async () => { + vi.mocked(getPosts).mockResolvedValue([{ post_url: "u1" }, { post_url: "u3" }] as never); + expect(await filterNewPostUrls(["u1", "u2", "u3", "u4"])).toEqual(["u2", "u4"]); + expect(getPosts).toHaveBeenCalledWith({ postUrls: ["u1", "u2", "u3", "u4"] }); + }); + + it("returns every URL when none are stored", async () => { + vi.mocked(getPosts).mockResolvedValue([] as never); + expect(await filterNewPostUrls(["u1"])).toEqual(["u1"]); + }); + + it("returns [] for empty input without querying", async () => { + expect(await filterNewPostUrls([])).toEqual([]); + expect(getPosts).not.toHaveBeenCalled(); + }); +}); diff --git a/lib/socials/filterNewPostUrls.ts b/lib/socials/filterNewPostUrls.ts new file mode 100644 index 000000000..c71d3e670 --- /dev/null +++ b/lib/socials/filterNewPostUrls.ts @@ -0,0 +1,21 @@ +import { getPosts } from "@/lib/supabase/posts/getPosts"; + +/** + * Returns the subset of post URLs that are genuinely new to the platform — + * i.e. not already present in `posts`. Must run BEFORE the scrape results are + * upserted (afterwards everything exists and nothing is "new"). + * + * Used to gate scrape-alert notifications: a scrape re-reads a profile's + * whole recent feed, so without this diff every scrape re-announces old + * posts as new (recoupable/chat#1855). + * + * @param postUrls - Candidate post URLs from a scrape result. + * @returns URLs with no existing `posts` row. + */ +export async function filterNewPostUrls(postUrls: string[]): Promise { + if (!postUrls.length) return []; + + const existing = await getPosts({ postUrls }); + const known = new Set((existing ?? []).map(post => post.post_url)); + return postUrls.filter(url => !known.has(url)); +} From 9a991be6295bd25e8da98f613dda573271ae269b Mon Sep 17 00:00:00 2001 From: Sweets Sweetman Date: Mon, 6 Jul 2026 17:05:48 -0500 Subject: [PATCH 2/4] feat(apify): one consolidated new-posts digest per scrape batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A roster scrape starts one Apify run per platform, each completing independently — extending per-platform alerts would mean 4+ emails per scrape. This registers every run under a batch_id at scrape start (apify_scraper_runs, columns from recoupable/database#41), records each webhook completion with its genuinely-new post URLs, and when the batch's last run completes sends ONE digest (per-platform sections, BCC-only) via the new digest module. Platforms with nothing new are omitted; a batch with nothing new sends nothing. Instagram's solo alert is suppressed for batch runs; legacy/non-batch runs keep today's behavior (recoupable/chat#1855, PR #4 of 6). Co-Authored-By: Claude Fable 5 --- .../__tests__/apifyWebhookHandler.test.ts | 41 +++++++++++ lib/apify/apifyWebhookHandler.ts | 18 +++++ .../__tests__/maybeSendScrapeDigest.test.ts | 73 +++++++++++++++++++ lib/apify/digest/getScrapeDigestRecipients.ts | 28 +++++++ lib/apify/digest/maybeSendScrapeDigest.ts | 27 +++++++ lib/apify/digest/sendScrapeDigestEmail.ts | 38 ++++++++++ ...ndleInstagramProfileScraperResults.test.ts | 36 +++++++++ .../handleInstagramProfileScraperResults.ts | 10 ++- lib/apify/scrapeProfileUrlBatch.ts | 18 +++-- .../handleTiktokProfileScraperResults.test.ts | 3 + .../handleTiktokProfileScraperResults.ts | 5 +- ...handleTwitterProfileScraperResults.test.ts | 3 + .../handleTwitterProfileScraperResults.ts | 5 +- lib/apify/validateApifyWebhookRequest.ts | 4 + .../postArtistSocialsScrapeHandler.test.ts | 9 ++- lib/artist/postArtistSocialsScrapeHandler.ts | 32 +++++++- .../completeApifyScraperRun.ts | 24 ++++++ .../insertApifyScraperRuns.ts | 19 +++++ .../selectApifyScraperRun.ts | 16 ++++ .../selectApifyScraperRunsByBatch.ts | 17 +++++ lib/supabase/apify_scraper_runs/types.ts | 12 +++ 21 files changed, 421 insertions(+), 17 deletions(-) create mode 100644 lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts create mode 100644 lib/apify/digest/getScrapeDigestRecipients.ts create mode 100644 lib/apify/digest/maybeSendScrapeDigest.ts create mode 100644 lib/apify/digest/sendScrapeDigestEmail.ts create mode 100644 lib/supabase/apify_scraper_runs/completeApifyScraperRun.ts create mode 100644 lib/supabase/apify_scraper_runs/insertApifyScraperRuns.ts create mode 100644 lib/supabase/apify_scraper_runs/selectApifyScraperRun.ts create mode 100644 lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch.ts create mode 100644 lib/supabase/apify_scraper_runs/types.ts diff --git a/lib/apify/__tests__/apifyWebhookHandler.test.ts b/lib/apify/__tests__/apifyWebhookHandler.test.ts index 98f7aa7d2..3cebad3ef 100644 --- a/lib/apify/__tests__/apifyWebhookHandler.test.ts +++ b/lib/apify/__tests__/apifyWebhookHandler.test.ts @@ -1,4 +1,6 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; +import { completeApifyScraperRun } from "@/lib/supabase/apify_scraper_runs/completeApifyScraperRun"; +import { maybeSendScrapeDigest } from "@/lib/apify/digest/maybeSendScrapeDigest"; import { NextRequest } from "next/server"; import { apifyWebhookHandler } from "../apifyWebhookHandler"; import { handleInstagramProfileScraperResults } from "../instagram/handleInstagramProfileScraperResults"; @@ -44,6 +46,13 @@ const baseBody = { resource: { defaultDatasetId: "ds_1" }, }; +vi.mock("@/lib/supabase/apify_scraper_runs/completeApifyScraperRun", () => ({ + completeApifyScraperRun: vi.fn(async () => null), +})); +vi.mock("@/lib/apify/digest/maybeSendScrapeDigest", () => ({ + maybeSendScrapeDigest: vi.fn(async () => null), +})); + describe("apifyWebhookHandler", () => { beforeEach(() => vi.clearAllMocks()); @@ -60,6 +69,38 @@ describe("apifyWebhookHandler", () => { expect(handleInstagramCommentsScraper).not.toHaveBeenCalled(); }); + it("records batch completion + triggers the digest check when the payload carries a run id (chat#1855)", async () => { + vi.mocked(handleInstagramProfileScraperResults).mockResolvedValue({ + posts: [], + newPostUrls: ["https://instagram.com/p/new1"], + } as never); + vi.mocked(completeApifyScraperRun).mockResolvedValue({ + run_id: "run-9", + batch_id: "batch-7", + } as never); + + const res = await apifyWebhookHandler( + makeRequest({ + ...baseBody, + eventData: { actorId: "dSCLg0C3YEZ83HzYX" }, + resource: { ...baseBody.resource, id: "run-9" }, + }), + ); + + expect(res.status).toBe(200); + expect(completeApifyScraperRun).toHaveBeenCalledWith("run-9", ["https://instagram.com/p/new1"]); + expect(maybeSendScrapeDigest).toHaveBeenCalledWith("batch-7"); + }); + + it("skips digest bookkeeping when the payload has no run id", async () => { + vi.mocked(handleInstagramProfileScraperResults).mockResolvedValue({ posts: [] } as never); + await apifyWebhookHandler( + makeRequest({ ...baseBody, eventData: { actorId: "dSCLg0C3YEZ83HzYX" } }), + ); + expect(completeApifyScraperRun).not.toHaveBeenCalled(); + expect(maybeSendScrapeDigest).not.toHaveBeenCalled(); + }); + it("dispatches comments scraper for the IG comments actor", async () => { vi.mocked(handleInstagramCommentsScraper).mockResolvedValue({ comments: [], diff --git a/lib/apify/apifyWebhookHandler.ts b/lib/apify/apifyWebhookHandler.ts index fdda46e5a..540950e65 100644 --- a/lib/apify/apifyWebhookHandler.ts +++ b/lib/apify/apifyWebhookHandler.ts @@ -1,6 +1,8 @@ import { NextRequest, NextResponse } from "next/server"; import { validateApifyWebhookRequest } from "@/lib/apify/validateApifyWebhookRequest"; import { getApifyResultHandler } from "@/lib/apify/getApifyResultHandler"; +import { completeApifyScraperRun } from "@/lib/supabase/apify_scraper_runs/completeApifyScraperRun"; +import { maybeSendScrapeDigest } from "@/lib/apify/digest/maybeSendScrapeDigest"; /** * Handler for `POST /api/apify`. Always responds 200 so Apify does not @@ -27,6 +29,22 @@ export async function apifyWebhookHandler(request: NextRequest): Promise ({ + selectApifyScraperRunsByBatch: vi.fn(), +})); +vi.mock("@/lib/apify/digest/getScrapeDigestRecipients", () => ({ + getScrapeDigestRecipients: vi.fn(), +})); +vi.mock("@/lib/apify/digest/sendScrapeDigestEmail", () => ({ + sendScrapeDigestEmail: vi.fn(), +})); + +const run = (over: Record) => ({ + run_id: "r1", + batch_id: "b1", + account_id: "acct-1", + social_id: "s1", + platform: "instagram", + completed_at: "2026-07-07T00:00:00Z", + new_post_urls: [], + ...over, +}); + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(getScrapeDigestRecipients).mockResolvedValue(["owner@example.com"]); + vi.mocked(sendScrapeDigestEmail).mockResolvedValue({ id: "email-1" } as never); +}); + +describe("maybeSendScrapeDigest", () => { + it("does nothing while sibling runs are still incomplete", async () => { + vi.mocked(selectApifyScraperRunsByBatch).mockResolvedValue([ + run({}), + run({ run_id: "r2", platform: "tiktok", completed_at: null }), + ] as never); + expect(await maybeSendScrapeDigest("b1")).toBeNull(); + expect(sendScrapeDigestEmail).not.toHaveBeenCalled(); + }); + + it("sends ONE digest with per-platform new posts when the batch completes", async () => { + vi.mocked(selectApifyScraperRunsByBatch).mockResolvedValue([ + run({ new_post_urls: ["https://instagram.com/p/1"] }), + run({ run_id: "r2", platform: "tiktok", new_post_urls: ["https://tiktok.com/v/2"] }), + run({ run_id: "r3", platform: "x", new_post_urls: [] }), + ] as never); + await maybeSendScrapeDigest("b1"); + expect(sendScrapeDigestEmail).toHaveBeenCalledOnce(); + const arg = vi.mocked(sendScrapeDigestEmail).mock.calls[0][0]; + expect(arg.sections).toEqual([ + { platform: "instagram", postUrls: ["https://instagram.com/p/1"] }, + { platform: "tiktok", postUrls: ["https://tiktok.com/v/2"] }, + ]); // x omitted — nothing new + expect(arg.emails).toEqual(["owner@example.com"]); + }); + + it("sends nothing when the batch completes with zero new posts anywhere", async () => { + vi.mocked(selectApifyScraperRunsByBatch).mockResolvedValue([ + run({}), + run({ run_id: "r2", platform: "tiktok" }), + ] as never); + expect(await maybeSendScrapeDigest("b1")).toBeNull(); + expect(sendScrapeDigestEmail).not.toHaveBeenCalled(); + }); + + it("no-ops for a null batch id (legacy runs)", async () => { + expect(await maybeSendScrapeDigest(null)).toBeNull(); + expect(selectApifyScraperRunsByBatch).not.toHaveBeenCalled(); + }); +}); diff --git a/lib/apify/digest/getScrapeDigestRecipients.ts b/lib/apify/digest/getScrapeDigestRecipients.ts new file mode 100644 index 000000000..a3c31cd90 --- /dev/null +++ b/lib/apify/digest/getScrapeDigestRecipients.ts @@ -0,0 +1,28 @@ +import { selectAccountSocials } from "@/lib/supabase/account_socials/selectAccountSocials"; +import { getAccountArtistIds } from "@/lib/supabase/account_artist_ids/getAccountArtistIds"; +import selectAccountEmails from "@/lib/supabase/account_emails/selectAccountEmails"; + +/** + * Resolves the digest recipients for a set of scraped socials: every owner + * 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 { + const uniqueSocialIds = Array.from(new Set(socialIds.filter(Boolean))); + if (!uniqueSocialIds.length) return []; + + const accountSocials = ( + await Promise.all( + uniqueSocialIds.map(socialId => selectAccountSocials({ socialId, limit: 10000 })), + ) + ).flat(); + + const accountArtistIds = await getAccountArtistIds({ + artistIds: accountSocials.map(a => a.account_id), + }); + const uniqueAccountIds = Array.from( + 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))); +} diff --git a/lib/apify/digest/maybeSendScrapeDigest.ts b/lib/apify/digest/maybeSendScrapeDigest.ts new file mode 100644 index 000000000..9247b9dd1 --- /dev/null +++ b/lib/apify/digest/maybeSendScrapeDigest.ts @@ -0,0 +1,27 @@ +import { selectApifyScraperRunsByBatch } from "@/lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch"; +import { getScrapeDigestRecipients } from "@/lib/apify/digest/getScrapeDigestRecipients"; +import { sendScrapeDigestEmail } from "@/lib/apify/digest/sendScrapeDigestEmail"; + +/** + * 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. + */ +export async function maybeSendScrapeDigest(batchId: string | null | undefined) { + if (!batchId) return null; + + const runs = await selectApifyScraperRunsByBatch(batchId); + if (!runs.length || runs.some(r => !r.completed_at)) return null; + + const sections = runs + .filter(r => (r.new_post_urls?.length ?? 0) > 0) + .map(r => ({ platform: r.platform ?? "other", postUrls: r.new_post_urls ?? [] })); + if (!sections.length) return null; + + const emails = await getScrapeDigestRecipients( + runs.map(r => r.social_id).filter((id): id is string => Boolean(id)), + ); + return await sendScrapeDigestEmail({ emails, sections }); +} diff --git a/lib/apify/digest/sendScrapeDigestEmail.ts b/lib/apify/digest/sendScrapeDigestEmail.ts new file mode 100644 index 000000000..c75df6943 --- /dev/null +++ b/lib/apify/digest/sendScrapeDigestEmail.ts @@ -0,0 +1,38 @@ +import { sendEmailWithResend } from "@/lib/emails/sendEmail"; +import { RECOUP_FROM_EMAIL } from "@/lib/const"; + +export type ScrapeDigestSection = { platform: string; postUrls: string[] }; +export type ScrapeDigestInput = { + emails: string[]; + sections: ScrapeDigestSection[]; + artistName?: string | null; +}; + +/** + * Sends the consolidated new-posts digest: one email per scrape batch, + * one section per platform that found genuinely new posts. Deterministic + * body, direct links to each new post. Recipients span tenants — BCC only + * (chat#1855). + */ +export async function sendScrapeDigestEmail({ emails, sections, artistName }: ScrapeDigestInput) { + if (!emails.length || !sections.length) return null; + + const total = sections.reduce((n, s) => n + s.postUrls.length, 0); + const who = artistName || "your artist"; + const sectionHtml = sections + .map( + s => + `

${s.platform}

    ${s.postUrls + .map(u => `
  • ${u}
    • `) + .join("")}
`, + ) + .join(""); + + return await sendEmailWithResend({ + from: RECOUP_FROM_EMAIL, + to: [RECOUP_FROM_EMAIL], + bcc: emails, + subject: `${who}: ${total} new post${total === 1 ? "" : "s"} found across ${sections.length} platform${sections.length === 1 ? "" : "s"}`, + html: `

New posts found for ${who}:

${sectionHtml}`, + }); +} diff --git a/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts b/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts index bab22c438..2a9850140 100644 --- a/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts +++ b/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts @@ -13,6 +13,7 @@ import { getAccountArtistIds } from "@/lib/supabase/account_artist_ids/getAccoun import selectAccountEmails from "@/lib/supabase/account_emails/selectAccountEmails"; import { uploadLinkToArweave } from "@/lib/arweave/uploadLinkToArweave"; import { filterNewPostUrls } from "@/lib/socials/filterNewPostUrls"; +import { selectApifyScraperRun } from "@/lib/supabase/apify_scraper_runs/selectApifyScraperRun"; vi.mock("@/lib/apify/client", () => ({ default: { dataset: vi.fn() } })); @@ -28,6 +29,9 @@ vi.mock("../handleInstagramProfileFollowUpRuns", () => ({ })); vi.mock("@/lib/apify/sendApifyWebhookEmail", () => ({ sendApifyWebhookEmail: vi.fn() })); vi.mock("@/lib/socials/filterNewPostUrls", () => ({ filterNewPostUrls: vi.fn() })); +vi.mock("@/lib/supabase/apify_scraper_runs/selectApifyScraperRun", () => ({ + selectApifyScraperRun: vi.fn(async () => null), +})); vi.mock("@/lib/supabase/socials/upsertSocials", () => ({ upsertSocials: vi.fn() })); vi.mock("@/lib/supabase/socials/selectSocials", () => ({ selectSocials: vi.fn(), @@ -128,4 +132,36 @@ describe("handleInstagramProfileScraperResults", () => { expect(handleInstagramProfileFollowUpRuns).toHaveBeenCalledOnce(); expect(result.social).toEqual({ id: "s1" }); }); + + it("suppresses the solo email for digest-batch runs (webhook layer sends ONE digest)", async () => { + mockDataset([ + { + latestPosts: [{ url: "u-new", timestamp: "t" }], + username: "alice", + url: "instagram.com/alice", + profilePicUrl: "https://a", + fullName: "Alice", + }, + ]); + vi.mocked(selectApifyScraperRun).mockResolvedValue({ + run_id: "run-1", + batch_id: "b1", + } as never); + vi.mocked(upsertPosts).mockResolvedValue({ data: null, error: null } as never); + vi.mocked(getPosts).mockResolvedValue([{ id: "p1", post_url: "u-new" }] as never); + vi.mocked(uploadLinkToArweave).mockResolvedValue(null); + vi.mocked(upsertSocials).mockResolvedValue([] as never); + vi.mocked(selectSocials).mockResolvedValue([{ id: "s1" }] as never); + vi.mocked(selectAccountSocials).mockResolvedValue([{ account_id: "a1" }] as never); + vi.mocked(getAccountArtistIds).mockResolvedValue([{ account_id: "a1" }] as never); + vi.mocked(selectAccountEmails).mockResolvedValue([{ email: "x@y.com" }] as never); + + const result = await handleInstagramProfileScraperResults({ + ...(payload as Record), + resource: { defaultDatasetId: "ds_1", id: "run-1" }, + } as never); + + expect(sendApifyWebhookEmail).not.toHaveBeenCalled(); // digest covers it + expect(result.newPostUrls).toEqual(["u-new"]); + }); }); diff --git a/lib/apify/instagram/handleInstagramProfileScraperResults.ts b/lib/apify/instagram/handleInstagramProfileScraperResults.ts index 4774c17f0..3244e8eb7 100644 --- a/lib/apify/instagram/handleInstagramProfileScraperResults.ts +++ b/lib/apify/instagram/handleInstagramProfileScraperResults.ts @@ -15,6 +15,7 @@ import { getFetchableUrl } from "@/lib/arweave/getFetchableUrl"; import type { ApifyInstagramProfileResult } from "@/lib/apify/types"; import type { ApifyWebhookPayload } from "@/lib/apify/validateApifyWebhookRequest"; import { filterNewPostUrls } from "@/lib/socials/filterNewPostUrls"; +import { selectApifyScraperRun } from "@/lib/supabase/apify_scraper_runs/selectApifyScraperRun"; import type { TablesInsert } from "@/types/database.types"; /** @@ -98,11 +99,16 @@ export async function handleInstagramProfileScraperResults(parsed: ApifyWebhookP // Email + follow-up scrape are independent side effects; isolate so a // mail outage doesn't block comment scraping and vice versa. + // Digest-batch runs get ONE consolidated email from the webhook layer — + // suppress the per-platform solo email for them (chat#1855). Legacy runs + // (no batch registration) keep the immediate alert. + const registeredRun = parsed.resource.id ? await selectApifyScraperRun(parsed.resource.id) : null; + let sentEmails = null; try { // Only notify when the scrape actually found posts new to the platform — // otherwise every scrape re-announces the profile's recent feed. - if (newPostUrls.length > 0) { + if (newPostUrls.length > 0 && !registeredRun?.batch_id) { sentEmails = await sendApifyWebhookEmail( firstResult, accountEmails.map(e => e.email).filter(Boolean), @@ -118,5 +124,5 @@ export async function handleInstagramProfileScraperResults(parsed: ApifyWebhookP console.error("[WARN] follow-up scrape failed:", error); } - return { posts, social, accountSocials, accountEmails, sentEmails }; + return { posts, social, accountSocials, accountEmails, sentEmails, newPostUrls }; } diff --git a/lib/apify/scrapeProfileUrlBatch.ts b/lib/apify/scrapeProfileUrlBatch.ts index 5b73ea7b3..0bf1c2fd5 100644 --- a/lib/apify/scrapeProfileUrlBatch.ts +++ b/lib/apify/scrapeProfileUrlBatch.ts @@ -1,5 +1,7 @@ import { ProfileScrapeResult, ScrapeProfileResult, scrapeProfileUrl } from "./scrapeProfileUrl"; +export type BatchProfileScrapeResult = ProfileScrapeResult & { profileUrl: string | null }; + type ScrapeProfileUrlBatchInput = { profileUrl: string | null | undefined; username: string | null | undefined; @@ -8,18 +10,22 @@ type ScrapeProfileUrlBatchInput = { export const scrapeProfileUrlBatch = async ( inputs: ScrapeProfileUrlBatchInput[], posts?: number, -): Promise => { +): Promise => { const results = await Promise.all( - inputs.map(({ profileUrl, username }) => - scrapeProfileUrl(profileUrl ?? null, username ?? "", posts), - ), + inputs.map(async ({ profileUrl, username }) => { + const result = await scrapeProfileUrl(profileUrl ?? null, username ?? "", posts); + return result ? { ...result, profileUrl: profileUrl ?? null } : null; + }), ); return results - .filter((result): result is ScrapeProfileResult => result !== null) - .map(({ runId, datasetId, error }) => ({ + .filter( + (result): result is ScrapeProfileResult & { profileUrl: string | null } => result !== null, + ) + .map(({ runId, datasetId, error, profileUrl }) => ({ runId, datasetId, error, + profileUrl, })); }; diff --git a/lib/apify/tiktok/__tests__/handleTiktokProfileScraperResults.test.ts b/lib/apify/tiktok/__tests__/handleTiktokProfileScraperResults.test.ts index 26c230684..9b84f8ca6 100644 --- a/lib/apify/tiktok/__tests__/handleTiktokProfileScraperResults.test.ts +++ b/lib/apify/tiktok/__tests__/handleTiktokProfileScraperResults.test.ts @@ -1,6 +1,9 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import { handleTiktokProfileScraperResults } from "@/lib/apify/tiktok/handleTiktokProfileScraperResults"; const listItems = vi.fn(); +vi.mock("@/lib/socials/filterNewPostUrls", () => ({ + filterNewPostUrls: vi.fn(async (urls: string[]) => urls), +})); vi.mock("@/lib/apify/client", () => ({ default: { dataset: vi.fn(() => ({ listItems })) } })); const upsertSocials = vi.fn(); vi.mock("@/lib/supabase/socials/upsertSocials", () => ({ diff --git a/lib/apify/tiktok/handleTiktokProfileScraperResults.ts b/lib/apify/tiktok/handleTiktokProfileScraperResults.ts index f7361d09b..d736efb6b 100644 --- a/lib/apify/tiktok/handleTiktokProfileScraperResults.ts +++ b/lib/apify/tiktok/handleTiktokProfileScraperResults.ts @@ -2,6 +2,7 @@ import apifyClient from "@/lib/apify/client"; import { upsertSocials } from "@/lib/supabase/socials/upsertSocials"; import { normalizeProfileUrl } from "@/lib/socials/normalizeProfileUrl"; import { persistPostsForSocial } from "@/lib/apify/persistPostsForSocial"; +import { filterNewPostUrls } from "@/lib/socials/filterNewPostUrls"; import { toIsoDate } from "@/lib/apify/toIsoDate"; import type { ApifyWebhookPayload } from "@/lib/apify/validateApifyWebhookRequest"; import type { TablesInsert } from "@/types/database.types"; @@ -47,7 +48,9 @@ export async function handleTiktokProfileScraperResults(parsed: ApifyWebhookPayl ? [{ post_url: item.webVideoUrl, updated_at: toIsoDate(item.createTimeISO) }] : [], ); + // Diff before persisting so the digest can report genuinely new posts (chat#1855). + const newPostUrls = await filterNewPostUrls(postRows.map(p => p.post_url)); const { posts } = await persistPostsForSocial({ postRows, profileUrl: social.profile_url }); - return { social, posts }; + return { social, posts, newPostUrls }; } diff --git a/lib/apify/twitter/__tests__/handleTwitterProfileScraperResults.test.ts b/lib/apify/twitter/__tests__/handleTwitterProfileScraperResults.test.ts index f1c03b78c..0497e058b 100644 --- a/lib/apify/twitter/__tests__/handleTwitterProfileScraperResults.test.ts +++ b/lib/apify/twitter/__tests__/handleTwitterProfileScraperResults.test.ts @@ -1,6 +1,9 @@ import { describe, it, expect, vi, beforeEach } from "vitest"; import { handleTwitterProfileScraperResults } from "@/lib/apify/twitter/handleTwitterProfileScraperResults"; const listItems = vi.fn(); +vi.mock("@/lib/socials/filterNewPostUrls", () => ({ + filterNewPostUrls: vi.fn(async (urls: string[]) => urls), +})); vi.mock("@/lib/apify/client", () => ({ default: { dataset: vi.fn(() => ({ listItems })) } })); const upsertSocials = vi.fn(); vi.mock("@/lib/supabase/socials/upsertSocials", () => ({ diff --git a/lib/apify/twitter/handleTwitterProfileScraperResults.ts b/lib/apify/twitter/handleTwitterProfileScraperResults.ts index c583cc9ca..4634fae0a 100644 --- a/lib/apify/twitter/handleTwitterProfileScraperResults.ts +++ b/lib/apify/twitter/handleTwitterProfileScraperResults.ts @@ -2,6 +2,7 @@ import apifyClient from "@/lib/apify/client"; import { upsertSocials } from "@/lib/supabase/socials/upsertSocials"; import { normalizeProfileUrl } from "@/lib/socials/normalizeProfileUrl"; import { persistPostsForSocial } from "@/lib/apify/persistPostsForSocial"; +import { filterNewPostUrls } from "@/lib/socials/filterNewPostUrls"; import { toIsoDate } from "@/lib/apify/toIsoDate"; import type { ApifyWebhookPayload } from "@/lib/apify/validateApifyWebhookRequest"; import type { TablesInsert } from "@/types/database.types"; @@ -52,7 +53,9 @@ export async function handleTwitterProfileScraperResults(parsed: ApifyWebhookPay const postRows: TablesInsert<"posts">[] = (items as TweetItem[]).flatMap(item => item.url ? [{ post_url: item.url, updated_at: toIsoDate(item.createdAt) }] : [], ); + // Diff before persisting so the digest can report genuinely new posts (chat#1855). + const newPostUrls = await filterNewPostUrls(postRows.map(p => p.post_url)); const { posts } = await persistPostsForSocial({ postRows, profileUrl: social.profile_url }); - return { social, posts }; + return { social, posts, newPostUrls }; } diff --git a/lib/apify/validateApifyWebhookRequest.ts b/lib/apify/validateApifyWebhookRequest.ts index 2a446b595..4ccb76eae 100644 --- a/lib/apify/validateApifyWebhookRequest.ts +++ b/lib/apify/validateApifyWebhookRequest.ts @@ -13,6 +13,10 @@ export const apifyWebhookPayloadSchema = z.object({ actorId: z.string().min(1, "eventData.actorId is required"), }), resource: z.object({ + // Run id — present on Apify's default payload; used to complete the + // digest-batch registration (chat#1855). Optional so trimmed payloads + // keep validating. + id: z.string().optional(), defaultDatasetId: z.string().min(1, "resource.defaultDatasetId is required"), }), }); diff --git a/lib/artist/__tests__/postArtistSocialsScrapeHandler.test.ts b/lib/artist/__tests__/postArtistSocialsScrapeHandler.test.ts index 9fc090e47..ee95e0235 100644 --- a/lib/artist/__tests__/postArtistSocialsScrapeHandler.test.ts +++ b/lib/artist/__tests__/postArtistSocialsScrapeHandler.test.ts @@ -9,6 +9,9 @@ import { checkAccountArtistAccess } from "@/lib/artists/checkAccountArtistAccess import { ensureSocialScrapeCredits } from "@/lib/socials/ensureSocialScrapeCredits"; import { deductSocialScrapeCredits } from "@/lib/socials/deductSocialScrapeCredits"; +vi.mock("@/lib/supabase/apify_scraper_runs/insertApifyScraperRuns", () => ({ + insertApifyScraperRuns: vi.fn(async () => ({ data: null, error: null })), +})); vi.mock("@/lib/apify/scrapeProfileUrlBatch", () => ({ scrapeProfileUrlBatch: vi.fn() })); vi.mock("@/lib/supabase/account_socials/selectAccountSocials", () => ({ selectAccountSocials: vi.fn(), @@ -42,8 +45,8 @@ describe("postArtistSocialsScrapeHandler", () => { vi.mocked(ensureSocialScrapeCredits).mockResolvedValue(null); vi.mocked(selectAccountSocials).mockResolvedValue(TWO_SOCIALS as never); vi.mocked(scrapeProfileUrlBatch).mockResolvedValue([ - { runId: "r1", datasetId: "d1", error: null }, - { runId: "r2", datasetId: "d2", error: null }, + { runId: "r1", datasetId: "d1", error: null, profileUrl: null }, + { runId: "r2", datasetId: "d2", error: null, profileUrl: null }, ]); }); @@ -99,7 +102,7 @@ describe("postArtistSocialsScrapeHandler", () => { it("forwards posts and deducts (5 + posts) per profile actually scraped", async () => { vi.mocked(scrapeProfileUrlBatch).mockResolvedValue([ - { runId: "r1", datasetId: "d1", error: null }, + { runId: "r1", datasetId: "d1", error: null, profileUrl: null }, ]); await postArtistSocialsScrapeHandler(makeRequest({ artist_account_id: ARTIST_ID, posts: 20 })); expect(ensureSocialScrapeCredits).toHaveBeenCalledWith(ACCOUNT_ID, 50); diff --git a/lib/artist/postArtistSocialsScrapeHandler.ts b/lib/artist/postArtistSocialsScrapeHandler.ts index 530670824..5028d2279 100644 --- a/lib/artist/postArtistSocialsScrapeHandler.ts +++ b/lib/artist/postArtistSocialsScrapeHandler.ts @@ -8,6 +8,8 @@ import { validateAuthContext } from "@/lib/auth/validateAuthContext"; import { checkAccountArtistAccess } from "@/lib/artists/checkAccountArtistAccess"; import { ensureSocialScrapeCredits } from "@/lib/socials/ensureSocialScrapeCredits"; import { deductSocialScrapeCredits } from "@/lib/socials/deductSocialScrapeCredits"; +import { insertApifyScraperRuns } from "@/lib/supabase/apify_scraper_runs/insertApifyScraperRuns"; +import { getSocialPlatformByLink } from "@/lib/artists/getSocialPlatformByLink"; import { getSocialScrapeCreditCost } from "@/lib/socials/getSocialScrapeCreditCost"; /** @@ -70,12 +72,34 @@ export async function postArtistSocialsScrapeHandler(request: NextRequest): Prom if (results.length) { await deductSocialScrapeCredits(authResult.accountId, costPerProfile * results.length); + + // Register the batch so per-platform webhook completions can assemble + // ONE consolidated new-posts digest instead of an email per platform + // (chat#1855). Registration failure must not fail the scrape. + const batchId = crypto.randomUUID(); + const socialByUrl = new Map( + socials.map(s => [s.social?.profile_url ?? "", s.social?.id ?? null]), + ); + await insertApifyScraperRuns( + results + .filter(r => r.runId) + .map(r => ({ + run_id: r.runId as string, + account_id: authResult.accountId, + social_id: r.profileUrl ? (socialByUrl.get(r.profileUrl) ?? null) : null, + platform: r.profileUrl ? getSocialPlatformByLink(r.profileUrl).toLowerCase() : null, + batch_id: batchId, + })), + ); } - return NextResponse.json(results, { - status: 200, - headers: getCorsHeaders(), - }); + return NextResponse.json( + results.map(({ runId, datasetId, error }) => ({ runId, datasetId, error })), + { + status: 200, + headers: getCorsHeaders(), + }, + ); } catch (error) { console.error("[ERROR] postArtistSocialsScrapeHandler error:", error); return NextResponse.json( diff --git a/lib/supabase/apify_scraper_runs/completeApifyScraperRun.ts b/lib/supabase/apify_scraper_runs/completeApifyScraperRun.ts new file mode 100644 index 000000000..fa8f346c3 --- /dev/null +++ b/lib/supabase/apify_scraper_runs/completeApifyScraperRun.ts @@ -0,0 +1,24 @@ +import supabase from "@/lib/supabase/serverClient"; +import type { ApifyScraperRunRow } from "@/lib/supabase/apify_scraper_runs/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). + */ +export async function completeApifyScraperRun( + runId: string, + newPostUrls: string[], +): Promise { + const { data, error } = await supabase + .from("apify_scraper_runs" as never) + .update({ completed_at: new Date().toISOString(), new_post_urls: newPostUrls } as never) + .eq("run_id", runId) + .select() + .maybeSingle(); + if (error) { + console.error("[ERROR] completeApifyScraperRun:", error); + return null; + } + return (data as ApifyScraperRunRow | null) ?? null; +} diff --git a/lib/supabase/apify_scraper_runs/insertApifyScraperRuns.ts b/lib/supabase/apify_scraper_runs/insertApifyScraperRuns.ts new file mode 100644 index 000000000..35a162908 --- /dev/null +++ b/lib/supabase/apify_scraper_runs/insertApifyScraperRuns.ts @@ -0,0 +1,19 @@ +import supabase from "@/lib/supabase/serverClient"; +import type { ApifyScraperRunRow } from "@/lib/supabase/apify_scraper_runs/types"; + +/** + * Records scrape runs at start time so per-platform webhook completions can + * find their digest-batch siblings (recoupable/chat#1855). + * + * @param runs - One row per started Apify run (run_id, account_id, batch_id, …). + */ +export async function insertApifyScraperRuns( + runs: Pick[], +) { + if (!runs.length) return { data: null, error: null }; + const { data, error } = await supabase + .from("apify_scraper_runs" as never) + .upsert(runs as never[], { onConflict: "run_id", ignoreDuplicates: true }); + if (error) console.error("[ERROR] insertApifyScraperRuns:", error); + return { data, error }; +} diff --git a/lib/supabase/apify_scraper_runs/selectApifyScraperRun.ts b/lib/supabase/apify_scraper_runs/selectApifyScraperRun.ts new file mode 100644 index 000000000..a9602a450 --- /dev/null +++ b/lib/supabase/apify_scraper_runs/selectApifyScraperRun.ts @@ -0,0 +1,16 @@ +import supabase from "@/lib/supabase/serverClient"; +import type { ApifyScraperRunRow } from "@/lib/supabase/apify_scraper_runs/types"; + +/** Returns the registered scrape run for an Apify run id, or null. */ +export async function selectApifyScraperRun(runId: string): Promise { + const { data, error } = await supabase + .from("apify_scraper_runs" as never) + .select("*") + .eq("run_id", runId) + .maybeSingle(); + if (error) { + console.error("[ERROR] selectApifyScraperRun:", error); + return null; + } + return (data as ApifyScraperRunRow | null) ?? null; +} diff --git a/lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch.ts b/lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch.ts new file mode 100644 index 000000000..0dd9457e6 --- /dev/null +++ b/lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch.ts @@ -0,0 +1,17 @@ +import supabase from "@/lib/supabase/serverClient"; +import type { ApifyScraperRunRow } from "@/lib/supabase/apify_scraper_runs/types"; + +/** Returns every scrape run registered under a digest batch. */ +export async function selectApifyScraperRunsByBatch( + batchId: string, +): Promise { + const { data, error } = await supabase + .from("apify_scraper_runs" as never) + .select("*") + .eq("batch_id", batchId); + if (error) { + console.error("[ERROR] selectApifyScraperRunsByBatch:", error); + return []; + } + return (data as ApifyScraperRunRow[] | null) ?? []; +} diff --git a/lib/supabase/apify_scraper_runs/types.ts b/lib/supabase/apify_scraper_runs/types.ts new file mode 100644 index 000000000..faa9d00cf --- /dev/null +++ b/lib/supabase/apify_scraper_runs/types.ts @@ -0,0 +1,12 @@ +/** apify_scraper_runs row incl. the digest-batch columns added in + * recoupable/database#41 (not yet in generated database.types). */ +export type ApifyScraperRunRow = { + run_id: string; + account_id: string; + social_id: string | null; + platform: string | null; + batch_id: string | null; + completed_at: string | null; + new_post_urls: string[] | null; + created_at?: string; +}; From 7960d93c474aa5c3b4c81ab3fc8f03e39afe6e8d Mon Sep 17 00:00:00 2001 From: Sweets Sweetman Date: Mon, 6 Jul 2026 17:09:26 -0500 Subject: [PATCH 3/4] feat(apify): deterministic digest template with direct links to each new post MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the per-send LLM email body (nondeterministic branding, vendor jargon reaching customers, no reliable post links) with a shared deterministic renderer: stable subject/branding, one section per platform, a direct link to every genuinely-new post, chat CTA secondary. Used by both the batch digest and the legacy solo Instagram alert, which now also requires new-post URLs and is BCC-only (recoupable/chat#1855, PR #5 of 6; supersedes the interim body from PR #4 and, for this file, the standalone BCC fix in api#758 — same invariant, tests included). Co-Authored-By: Claude Fable 5 --- .../__tests__/sendApifyWebhookEmail.test.ts | 62 +++++++++++++++++++ .../__tests__/renderScrapeDigestHtml.test.ts | 39 ++++++++++++ lib/apify/digest/renderScrapeDigestHtml.ts | 53 ++++++++++++++++ lib/apify/digest/sendScrapeDigestEmail.ts | 1 + ...ndleInstagramProfileScraperResults.test.ts | 2 +- .../handleInstagramProfileScraperResults.ts | 1 + lib/apify/sendApifyWebhookEmail.ts | 55 ++++++---------- 7 files changed, 177 insertions(+), 36 deletions(-) create mode 100644 lib/apify/__tests__/sendApifyWebhookEmail.test.ts create mode 100644 lib/apify/digest/__tests__/renderScrapeDigestHtml.test.ts create mode 100644 lib/apify/digest/renderScrapeDigestHtml.ts diff --git a/lib/apify/__tests__/sendApifyWebhookEmail.test.ts b/lib/apify/__tests__/sendApifyWebhookEmail.test.ts new file mode 100644 index 000000000..65d60d36f --- /dev/null +++ b/lib/apify/__tests__/sendApifyWebhookEmail.test.ts @@ -0,0 +1,62 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { sendApifyWebhookEmail } from "@/lib/apify/sendApifyWebhookEmail"; +import { RECOUP_FROM_EMAIL } from "@/lib/const"; + +const sendEmailWithResend = vi.fn(); +vi.mock("@/lib/emails/sendEmail", () => ({ + sendEmailWithResend: (...a: unknown[]) => sendEmailWithResend(...a), +})); +vi.mock("@/lib/composio/getFrontendBaseUrl", () => ({ + getFrontendBaseUrl: () => "https://chat.recoupable.dev", +})); + +const PROFILE = { + fullName: "Ashnikko", + username: "ashnikko", + url: "https://instagram.com/ashnikko", + profilePicUrl: "https://example.com/pic.jpg", + biography: "bio", + followersCount: 100, + followsCount: 10, + latestPosts: [], +} as never; +const NEW_URLS = ["https://instagram.com/p/new1"]; + +beforeEach(() => { + vi.clearAllMocks(); + sendEmailWithResend.mockResolvedValue({ id: "email-1" }); +}); + +describe("sendApifyWebhookEmail", () => { + it("BCCs recipients so no account can see another's address (chat#1855)", async () => { + const recipients = ["manager@customer-a.com", "label@customer-b.com"]; + await sendApifyWebhookEmail(PROFILE, recipients, NEW_URLS); + const payload = sendEmailWithResend.mock.calls[0][0] as Record; + expect(payload.bcc).toEqual(recipients); + expect(payload.to).toEqual([RECOUP_FROM_EMAIL]); + }); + + it("never carries more than our own address in to/cc, regardless of recipient count", async () => { + const recipients = Array.from({ length: 25 }, (_, i) => `user${i}@example.com`); + await sendApifyWebhookEmail(PROFILE, recipients, NEW_URLS); + const payload = sendEmailWithResend.mock.calls[0][0] as Record; + expect([payload.to, payload.cc].flat().filter(Boolean)).toEqual([RECOUP_FROM_EMAIL]); + }); + + it("links the new posts in the deterministic body — no vendor jargon", async () => { + await sendApifyWebhookEmail(PROFILE, ["a@b.com"], NEW_URLS); + const payload = sendEmailWithResend.mock.calls[0][0] as Record; + expect(payload.html).toContain('href="https://instagram.com/p/new1"'); + expect((payload.html + payload.subject).toLowerCase()).not.toContain("apify"); + }); + + it("returns null and sends nothing when there are no recipients", async () => { + expect(await sendApifyWebhookEmail(PROFILE, [], NEW_URLS)).toBeNull(); + expect(sendEmailWithResend).not.toHaveBeenCalled(); + }); + + it("returns null and sends nothing when no post is genuinely new", async () => { + expect(await sendApifyWebhookEmail(PROFILE, ["a@b.com"], [])).toBeNull(); + expect(sendEmailWithResend).not.toHaveBeenCalled(); + }); +}); diff --git a/lib/apify/digest/__tests__/renderScrapeDigestHtml.test.ts b/lib/apify/digest/__tests__/renderScrapeDigestHtml.test.ts new file mode 100644 index 000000000..3562f7310 --- /dev/null +++ b/lib/apify/digest/__tests__/renderScrapeDigestHtml.test.ts @@ -0,0 +1,39 @@ +import { describe, it, expect } from "vitest"; +import { renderScrapeDigestHtml } from "@/lib/apify/digest/renderScrapeDigestHtml"; + +const SECTIONS = [ + { + platform: "instagram", + postUrls: ["https://instagram.com/p/abc", "https://instagram.com/p/def"], + }, + { platform: "tiktok", postUrls: ["https://tiktok.com/@a/video/1"] }, +]; + +describe("renderScrapeDigestHtml", () => { + it("links every new post, grouped under its platform", () => { + const { html } = renderScrapeDigestHtml({ sections: SECTIONS, artistName: "Ashnikko" }); + for (const s of SECTIONS) for (const u of s.postUrls) expect(html).toContain(`href="${u}"`); + expect(html.toLowerCase()).toContain("instagram"); + expect(html.toLowerCase()).toContain("tiktok"); + }); + + it("is deterministic — identical input renders identical output", () => { + const a = renderScrapeDigestHtml({ sections: SECTIONS, artistName: "Ashnikko" }); + const b = renderScrapeDigestHtml({ sections: SECTIONS, artistName: "Ashnikko" }); + expect(a).toEqual(b); + }); + + it("never leaks internal vendor jargon to customers", () => { + const { html, subject } = renderScrapeDigestHtml({ + sections: SECTIONS, + artistName: "Ashnikko", + }); + expect((html + subject).toLowerCase()).not.toContain("apify"); + expect((html + subject).toLowerCase()).not.toContain("dataset"); + }); + + it("counts posts and platforms in the subject", () => { + const { subject } = renderScrapeDigestHtml({ sections: SECTIONS, artistName: "Ashnikko" }); + expect(subject).toBe("Ashnikko: 3 new posts across 2 platforms"); + }); +}); diff --git a/lib/apify/digest/renderScrapeDigestHtml.ts b/lib/apify/digest/renderScrapeDigestHtml.ts new file mode 100644 index 000000000..f0583a3a8 --- /dev/null +++ b/lib/apify/digest/renderScrapeDigestHtml.ts @@ -0,0 +1,53 @@ +import { getFrontendBaseUrl } from "@/lib/composio/getFrontendBaseUrl"; +import type { ScrapeDigestSection } from "@/lib/apify/digest/sendScrapeDigestEmail"; + +const PLATFORM_LABELS: Record = { + instagram: "Instagram", + tiktok: "TikTok", + x: "X", + twitter: "X", + youtube: "YouTube", + facebook: "Facebook", + threads: "Threads", +}; + +/** + * Deterministic house-style renderer for the new-posts digest (chat#1855). + * Replaces the per-send LLM body: identical input renders identical output, + * every genuinely-new post is linked directly, and internal vendor jargon + * never reaches customers. + */ +export function renderScrapeDigestHtml({ + sections, + artistName, +}: { + sections: ScrapeDigestSection[]; + artistName?: string | null; +}): { subject: string; html: string } { + const who = artistName || "Your artist"; + const total = sections.reduce((n, s) => n + s.postUrls.length, 0); + const subject = `${who}: ${total} new post${total === 1 ? "" : "s"} across ${sections.length} platform${sections.length === 1 ? "" : "s"}`; + + const sectionHtml = sections + .map(section => { + const label = PLATFORM_LABELS[section.platform.toLowerCase()] ?? section.platform; + const items = section.postUrls + .map(url => { + const href = url.startsWith("http") ? url : `https://${url}`; + return `
  • ${href}
    • `; + }) + .join(""); + return `

    ${label}

      ${items}
    `; + }) + .join(""); + + const chatUrl = `${getFrontendBaseUrl()}/?q=${encodeURIComponent("tell me about my artist's latest posts")}`; + const html = `
    +

    New posts from ${who}

    +

    We found ${total} new post${total === 1 ? "" : "s"} since we last checked:

    +${sectionHtml} +

    Ask Recoup about these posts →

    +
    `; + + return { subject, html }; +} diff --git a/lib/apify/digest/sendScrapeDigestEmail.ts b/lib/apify/digest/sendScrapeDigestEmail.ts index c75df6943..3ae915bbc 100644 --- a/lib/apify/digest/sendScrapeDigestEmail.ts +++ b/lib/apify/digest/sendScrapeDigestEmail.ts @@ -1,5 +1,6 @@ import { sendEmailWithResend } from "@/lib/emails/sendEmail"; import { RECOUP_FROM_EMAIL } from "@/lib/const"; +import { renderScrapeDigestHtml } from "@/lib/apify/digest/renderScrapeDigestHtml"; export type ScrapeDigestSection = { platform: string; postUrls: string[] }; export type ScrapeDigestInput = { diff --git a/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts b/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts index 2a9850140..d8310dbd5 100644 --- a/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts +++ b/lib/apify/instagram/__tests__/handleInstagramProfileScraperResults.test.ts @@ -97,7 +97,7 @@ describe("handleInstagramProfileScraperResults", () => { expect(upsertPosts).toHaveBeenCalledOnce(); expect(upsertSocialPosts).toHaveBeenCalledOnce(); - expect(sendApifyWebhookEmail).toHaveBeenCalledWith(expect.any(Object), ["x@y.com"]); + expect(sendApifyWebhookEmail).toHaveBeenCalledWith(expect.any(Object), ["x@y.com"], ["u1"]); expect(handleInstagramProfileFollowUpRuns).toHaveBeenCalledOnce(); expect(result.social).toEqual({ id: "s1" }); expect(result.posts).toEqual(posts); diff --git a/lib/apify/instagram/handleInstagramProfileScraperResults.ts b/lib/apify/instagram/handleInstagramProfileScraperResults.ts index 3244e8eb7..de38d4c44 100644 --- a/lib/apify/instagram/handleInstagramProfileScraperResults.ts +++ b/lib/apify/instagram/handleInstagramProfileScraperResults.ts @@ -112,6 +112,7 @@ export async function handleInstagramProfileScraperResults(parsed: ApifyWebhookP sentEmails = await sendApifyWebhookEmail( firstResult, accountEmails.map(e => e.email).filter(Boolean), + newPostUrls, ); } } catch (error) { diff --git a/lib/apify/sendApifyWebhookEmail.ts b/lib/apify/sendApifyWebhookEmail.ts index 051db1495..5503b3612 100644 --- a/lib/apify/sendApifyWebhookEmail.ts +++ b/lib/apify/sendApifyWebhookEmail.ts @@ -1,54 +1,39 @@ -import generateText from "@/lib/ai/generateText"; import { sendEmailWithResend } from "@/lib/emails/sendEmail"; import { RECOUP_FROM_EMAIL } from "@/lib/const"; -import { getFrontendBaseUrl } from "@/lib/composio/getFrontendBaseUrl"; +import { renderScrapeDigestHtml } from "@/lib/apify/digest/renderScrapeDigestHtml"; import type { ApifyInstagramProfileResult } from "@/lib/apify/types"; /** - * Sends an Apify-webhook summary email to the given recipients using - * Resend. Generates the email body via an LLM based on the first - * profile result. + * Sends the legacy single-platform Instagram alert (non-batch scrape runs). + * Batch runs get the consolidated digest instead. Body comes from the shared + * deterministic renderer — the per-send LLM body is gone (chat#1855): stable + * branding, direct links to the new posts, no vendor jargon. * * @param profile - First dataset result from the profile scraper. - * @param emails - Recipient email addresses. - * @returns The Resend response, or `null` when there are no recipients. + * @param emails - Recipient email addresses (cross-tenant — BCC only). + * @param newPostUrls - Post URLs genuinely new to the platform. + * @returns The Resend response, or `null` when there is nothing to send. */ export async function sendApifyWebhookEmail( profile: ApifyInstagramProfileResult, emails: string[], + newPostUrls: string[] = [], ) { - if (!emails?.length) return null; + if (!emails?.length || !newPostUrls.length) return null; - const prompt = `You have a new Apify dataset update. Here is the data: - -Key Data -Full Name: ${profile.fullName} -Username: ${profile.username} -Profile URL: ${profile.url} -Profile Picture: ${profile.profilePicUrl} -Biography: ${profile.biography} -Followers: ${profile.followersCount} -Following: ${profile.followsCount} -Latest Posts: ${(Array.isArray(profile.latestPosts) ? profile.latestPosts : []).map(p => JSON.stringify(p)).join(", ")} -`; - - const { text } = await generateText({ - system: `you are a record label services manager for Recoup. - write beautiful html email. - subject: New Apify Dataset Notification. you're notifying music managers about new posts being available for one of their roster musician's Instagram profile. - include a link to view the instagram profile. - call to action is to open a chat link to learn more about the latest posts using Recoup Chat (AI Agents): ${getFrontendBaseUrl()}/?q=tell%20me%20about%20my%20latest%20Ig%20posts - You'll be passed a dataset summary for a musician profile and their latest posts on instagram. - your goal is to get the recipient to click a cta link to open a chat link to learn more about the latest posts using Recoup Chat (AI Agents). - only include the email body html. - no headers or subject`, - prompt, + const { subject, html } = renderScrapeDigestHtml({ + sections: [{ platform: "instagram", postUrls: newPostUrls }], + artistName: profile.fullName ?? profile.username ?? null, }); + // Recipients span multiple customer accounts (the social's watchers are + // resolved cross-tenant), so they must NEVER share a visible To: line — + // BCC only, with ourselves as the required To: (chat#1855). return await sendEmailWithResend({ from: RECOUP_FROM_EMAIL, - to: emails, - subject: `${profile.fullName ?? profile.username ?? "Your artist"} has new posts on Instagram`, - html: text, + to: [RECOUP_FROM_EMAIL], + bcc: emails, + subject, + html, }); } From 68bd5159c4c4453120e5f7dd5a1de8243ac572a1 Mon Sep 17 00:00:00 2001 From: Sweets Sweetman Date: Mon, 6 Jul 2026 17:13:20 -0500 Subject: [PATCH 4/4] feat(apify): digest rate cap (1/artist/24h) + email_send_log audit trail MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Caps the digest at one per watched artist entity per 24h and records every digest send in email_send_log (the solo alert path bypassed the log entirely — the 7-email burst on chat#1855 was only discoverable via the Resend admin endpoint). Recipients resolver now also returns the watched artist entity ids for the cap key (recoupable/chat#1855, PR #6 of 6). Co-Authored-By: Claude Fable 5 --- .../__tests__/maybeSendScrapeDigest.test.ts | 26 ++++++++++++++++- lib/apify/digest/getScrapeDigestRecipients.ts | 11 +++++-- lib/apify/digest/maybeSendScrapeDigest.ts | 29 +++++++++++++++++-- .../selectRecentScrapeDigestLogs.ts | 20 +++++++++++++ 4 files changed, 80 insertions(+), 6 deletions(-) create mode 100644 lib/supabase/email_send_log/selectRecentScrapeDigestLogs.ts diff --git a/lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts b/lib/apify/digest/__tests__/maybeSendScrapeDigest.test.ts index aa5e32654..759d92181 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 { selectApifyScraperRunsByBatch } from "@/lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch"; import { getScrapeDigestRecipients } from "@/lib/apify/digest/getScrapeDigestRecipients"; import { sendScrapeDigestEmail } from "@/lib/apify/digest/sendScrapeDigestEmail"; +import { selectRecentScrapeDigestLogs } from "@/lib/supabase/email_send_log/selectRecentScrapeDigestLogs"; +import { logEmailAttempt } from "@/lib/emails/logEmailAttempt"; vi.mock("@/lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch", () => ({ selectApifyScraperRunsByBatch: vi.fn(), @@ -13,6 +15,10 @@ vi.mock("@/lib/apify/digest/getScrapeDigestRecipients", () => ({ vi.mock("@/lib/apify/digest/sendScrapeDigestEmail", () => ({ sendScrapeDigestEmail: vi.fn(), })); +vi.mock("@/lib/supabase/email_send_log/selectRecentScrapeDigestLogs", () => ({ + selectRecentScrapeDigestLogs: vi.fn(async () => []), +})); +vi.mock("@/lib/emails/logEmailAttempt", () => ({ logEmailAttempt: vi.fn(async () => {}) })); const run = (over: Record) => ({ run_id: "r1", @@ -27,7 +33,10 @@ const run = (over: Record) => ({ beforeEach(() => { vi.clearAllMocks(); - vi.mocked(getScrapeDigestRecipients).mockResolvedValue(["owner@example.com"]); + vi.mocked(getScrapeDigestRecipients).mockResolvedValue({ + emails: ["owner@example.com"], + artistIds: ["artist-1"], + } as never); vi.mocked(sendScrapeDigestEmail).mockResolvedValue({ id: "email-1" } as never); }); @@ -55,6 +64,21 @@ describe("maybeSendScrapeDigest", () => { { platform: "tiktok", postUrls: ["https://tiktok.com/v/2"] }, ]); // x omitted — nothing new expect(arg.emails).toEqual(["owner@example.com"]); + // audit trail: the send is recorded per watched artist entity + expect(logEmailAttempt).toHaveBeenCalledWith( + expect.objectContaining({ accountId: "artist-1", status: "sent" }), + ); + }); + + it("suppresses the digest when one was sent for the artist within 24h (rate cap)", async () => { + vi.mocked(selectApifyScraperRunsByBatch).mockResolvedValue([ + run({ new_post_urls: ["https://instagram.com/p/1"] }), + ] as never); + vi.mocked(selectRecentScrapeDigestLogs).mockResolvedValue([ + { account_id: "artist-1", created_at: "2026-07-07T00: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 () => { diff --git a/lib/apify/digest/getScrapeDigestRecipients.ts b/lib/apify/digest/getScrapeDigestRecipients.ts index a3c31cd90..56024982d 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 9247b9dd1..72a4e4e0f 100644 --- a/lib/apify/digest/maybeSendScrapeDigest.ts +++ b/lib/apify/digest/maybeSendScrapeDigest.ts @@ -1,6 +1,8 @@ import { selectApifyScraperRunsByBatch } from "@/lib/supabase/apify_scraper_runs/selectApifyScraperRunsByBatch"; import { getScrapeDigestRecipients } from "@/lib/apify/digest/getScrapeDigestRecipients"; import { sendScrapeDigestEmail } from "@/lib/apify/digest/sendScrapeDigestEmail"; +import { selectRecentScrapeDigestLogs } from "@/lib/supabase/email_send_log/selectRecentScrapeDigestLogs"; +import { logEmailAttempt } from "@/lib/emails/logEmailAttempt"; /** * Batch-completion check for the one-digest-per-scrape design (chat#1855): @@ -20,8 +22,31 @@ export async function maybeSendScrapeDigest(batchId: string | null | undefined) .map(r => ({ platform: r.platform ?? "other", postUrls: r.new_post_urls ?? [] })); 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)), ); - return await sendScrapeDigestEmail({ emails, sections }); + + // Rate cap: at most one digest per watched artist entity per 24h (chat#1855). + const since = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString(); + const recent = await selectRecentScrapeDigestLogs(artistIds, since); + if (recent.length > 0) return null; + + const sent = await sendScrapeDigestEmail({ emails, sections }); + + // Audit trail: record the send per artist entity so alerts are debuggable + // and the cap has something to read (the solo path never logged at all). + const resendId = (sent as { id?: string } | null)?.id; + if (sent) { + await Promise.all( + artistIds.map(artistId => + logEmailAttempt({ + rawBody: `scrape-digest:${batchId}`, + status: "sent", + accountId: artistId, + resendId, + }), + ), + ); + } + return sent; } diff --git a/lib/supabase/email_send_log/selectRecentScrapeDigestLogs.ts b/lib/supabase/email_send_log/selectRecentScrapeDigestLogs.ts new file mode 100644 index 000000000..a45631db0 --- /dev/null +++ b/lib/supabase/email_send_log/selectRecentScrapeDigestLogs.ts @@ -0,0 +1,20 @@ +import supabase from "@/lib/supabase/serverClient"; + +/** + * Returns scrape-digest send-log rows for any of the given artist entity ids + * newer than `since` — the digest rate cap's lookback (chat#1855). + */ +export async function selectRecentScrapeDigestLogs(artistIds: string[], since: string) { + if (!artistIds.length) return []; + const { data, error } = await supabase + .from("email_send_log") + .select("account_id, created_at") + .in("account_id", artistIds) + .like("raw_body", "scrape-digest:%") + .gte("created_at", since); + if (error) { + console.error("[ERROR] selectRecentScrapeDigestLogs:", error); + return []; + } + return data ?? []; +}