diff --git a/lib/apify/__tests__/apifyWebhookHandler.test.ts b/lib/apify/__tests__/apifyWebhookHandler.test.ts index 98f7aa7d..8b458f78 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 { updateApifyScraperRun } from "@/lib/supabase/apify_scraper_runs/updateApifyScraperRun"; +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/updateApifyScraperRun", () => ({ + updateApifyScraperRun: 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(updateApifyScraperRun).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(updateApifyScraperRun).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(updateApifyScraperRun).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 fdda46e5..5ddfb46c 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 { updateApifyScraperRun } from "@/lib/supabase/apify_scraper_runs/updateApifyScraperRun"; +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 ({ + selectApifyScraperRuns: 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(selectApifyScraperRuns).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(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"] }), + 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(selectApifyScraperRuns).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(selectApifyScraperRuns).not.toHaveBeenCalled(); + }); +}); diff --git a/lib/apify/digest/getScrapeDigestRecipients.ts b/lib/apify/digest/getScrapeDigestRecipients.ts new file mode 100644 index 00000000..a3c31cd9 --- /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 00000000..cd14c13f --- /dev/null +++ b/lib/apify/digest/maybeSendScrapeDigest.ts @@ -0,0 +1,28 @@ +import { selectApifyScraperRuns } from "@/lib/supabase/apify_scraper_runs/selectApifyScraperRuns"; +import { parseNewPostUrls } from "@/lib/apify/digest/parseNewPostUrls"; +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 selectApifyScraperRuns({ batchId }); + if (!runs.length || runs.some(r => !r.completed_at)) return null; + + const sections = runs + .map(r => ({ platform: r.platform ?? "other", postUrls: parseNewPostUrls(r.new_post_urls) })) + .filter(s => s.postUrls.length > 0); + 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/parseNewPostUrls.ts b/lib/apify/digest/parseNewPostUrls.ts new file mode 100644 index 00000000..90e89c70 --- /dev/null +++ b/lib/apify/digest/parseNewPostUrls.ts @@ -0,0 +1,15 @@ +import { z } from "zod"; +import type { Json } from "@/types/database.types"; + +const newPostUrlsSchema = z.array(z.string()); + +/** + * Parses the `new_post_urls` JSONB column into a string array. JSONB is the + * one column the generated types can't narrow past `Json`, so this is the + * runtime boundary: anything malformed degrades to "no new posts" instead of + * crashing the digest assembler. + */ +export function parseNewPostUrls(value: Json | null): string[] { + const result = newPostUrlsSchema.safeParse(value); + return result.success ? result.data : []; +} diff --git a/lib/apify/digest/sendScrapeDigestEmail.ts b/lib/apify/digest/sendScrapeDigestEmail.ts new file mode 100644 index 00000000..c75df694 --- /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 bab22c43..2a985014 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 4774c17f..3244e8eb 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 5b73ea7b..0bf1c2fd 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 26c23068..9b84f8ca 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 f7361d09..d736efb6 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 f1c03b78..0497e058 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 c583cc9c..4634fae0 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 2a446b59..4ccb76ea 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 9fc090e4..7e64203e 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/upsertApifyScraperRuns", () => ({ + upsertApifyScraperRuns: 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 53067082..1cfe9a94 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 { upsertApifyScraperRuns } from "@/lib/supabase/apify_scraper_runs/upsertApifyScraperRuns"; +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 upsertApifyScraperRuns( + 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/selectApifyScraperRun.ts b/lib/supabase/apify_scraper_runs/selectApifyScraperRun.ts new file mode 100644 index 00000000..dcbb7577 --- /dev/null +++ b/lib/supabase/apify_scraper_runs/selectApifyScraperRun.ts @@ -0,0 +1,18 @@ +import supabase from "@/lib/supabase/serverClient"; +import type { Tables } from "@/types/database.types"; + +/** Returns the registered scrape run for an Apify run id, or null. */ +export async function selectApifyScraperRun( + runId: string, +): Promise | null> { + const { data, error } = await supabase + .from("apify_scraper_runs") + .select("*") + .eq("run_id", runId) + .maybeSingle(); + if (error) { + console.error("[ERROR] selectApifyScraperRun:", error); + return null; + } + return data ?? null; +} diff --git a/lib/supabase/apify_scraper_runs/selectApifyScraperRuns.ts b/lib/supabase/apify_scraper_runs/selectApifyScraperRuns.ts new file mode 100644 index 00000000..f266da35 --- /dev/null +++ b/lib/supabase/apify_scraper_runs/selectApifyScraperRuns.ts @@ -0,0 +1,26 @@ +import supabase from "@/lib/supabase/serverClient"; +import type { Tables } from "@/types/database.types"; + +/** + * Select scrape runs with optional filters. + * + * @param batchId - Return the runs registered under this digest batch. + */ +export async function selectApifyScraperRuns({ + batchId, +}: { + batchId?: string; +} = {}): Promise[]> { + let query = supabase.from("apify_scraper_runs").select("*"); + + if (batchId) { + query = query.eq("batch_id", batchId); + } + + const { data, error } = await query; + if (error) { + console.error("[ERROR] selectApifyScraperRuns:", error); + return []; + } + return data ?? []; +} diff --git a/lib/supabase/apify_scraper_runs/updateApifyScraperRun.ts b/lib/supabase/apify_scraper_runs/updateApifyScraperRun.ts new file mode 100644 index 00000000..bf217a92 --- /dev/null +++ b/lib/supabase/apify_scraper_runs/updateApifyScraperRun.ts @@ -0,0 +1,24 @@ +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). + */ +export async function updateApifyScraperRun( + runId: string, + newPostUrls: string[], +): Promise | null> { + const { data, error } = await supabase + .from("apify_scraper_runs") + .update({ completed_at: new Date().toISOString(), new_post_urls: newPostUrls }) + .eq("run_id", runId) + .select() + .maybeSingle(); + if (error) { + console.error("[ERROR] updateApifyScraperRun:", error); + return null; + } + return data ?? null; +} diff --git a/lib/supabase/apify_scraper_runs/upsertApifyScraperRuns.ts b/lib/supabase/apify_scraper_runs/upsertApifyScraperRuns.ts new file mode 100644 index 00000000..2ec7f19c --- /dev/null +++ b/lib/supabase/apify_scraper_runs/upsertApifyScraperRuns.ts @@ -0,0 +1,17 @@ +import supabase from "@/lib/supabase/serverClient"; +import type { TablesInsert } from "@/types/database.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 upsertApifyScraperRuns(runs: TablesInsert<"apify_scraper_runs">[]) { + if (!runs.length) return { data: null, error: null }; + const { data, error } = await supabase + .from("apify_scraper_runs") + .upsert(runs, { onConflict: "run_id", ignoreDuplicates: true }); + if (error) console.error("[ERROR] upsertApifyScraperRuns:", error); + return { data, error }; +} diff --git a/types/database.types.ts b/types/database.types.ts index 5cdec74e..141ec932 100644 --- a/types/database.types.ts +++ b/types/database.types.ts @@ -737,6 +737,39 @@ export type Database = { }; Relationships: []; }; + apify_scraper_runs: { + Row: { + account_id: string; + batch_id: string | null; + completed_at: string | null; + created_at: string; + new_post_urls: Json | null; + platform: string | null; + run_id: string; + social_id: string | null; + }; + Insert: { + account_id: string; + batch_id?: string | null; + completed_at?: string | null; + created_at?: string; + new_post_urls?: Json | null; + platform?: string | null; + run_id: string; + social_id?: string | null; + }; + Update: { + account_id?: string; + batch_id?: string | null; + completed_at?: string | null; + created_at?: string; + new_post_urls?: Json | null; + platform?: string | null; + run_id?: string; + social_id?: string | null; + }; + Relationships: []; + }; app_store_link_clicked: { Row: { clientId: string | null; @@ -3829,27 +3862,6 @@ export type Database = { Returns: undefined; }; extract_domain: { Args: { email: string }; Returns: string }; - get_catalog_measurements_aggregate: { - Args: { p_catalog: string; p_artist?: string }; - Returns: { - measured_song_count: number; - total_streams: number; - }[]; - }; - get_catalog_measurements_page: { - Args: { - p_catalog: string; - p_artist?: string; - p_limit?: number; - p_offset?: number; - }; - Returns: { - isrc: string; - title: string | null; - playcount: number; - measured_at: string; - }[]; - }; get_account_invitations: { Args: { account_slug: string }; Returns: { @@ -3891,6 +3903,27 @@ export type Database = { Args: { artistid: string; email: string }; Returns: Json; }; + get_catalog_measurements_aggregate: { + Args: { p_artist?: string; p_catalog: string }; + Returns: { + measured_song_count: number; + total_streams: number; + }[]; + }; + get_catalog_measurements_page: { + Args: { + p_artist?: string; + p_catalog: string; + p_limit?: number; + p_offset?: number; + }; + Returns: { + isrc: string; + measured_at: string; + playcount: number; + title: string; + }[]; + }; get_config: { Args: never; Returns: Json }; get_credit_spend_digest: { Args: { p_limit?: number; p_since: string };