Merge pull request #470 from shamanic-technologies/main

chore: sync hotfix v0.39.88 to staging

Author: KevinLourd

CLAUDE.md

@@ -438,6 +438,9 @@ window. `src/lib/linkedin-engagement-spec.ts` (pure) + `src/lib/linkedin-engagem
   A dead page (`linkedin_company_pages.posts_status = not_found`, re-checked after
   24h) is SKIPPED with a loud log + warn trace, the other pages served; every page
   dead = 422 `competitor_pages_unreadable` (all providers down = 502, retryable).
+  A page every provider FAILED on (`posts_failed_at`, BOX-LOAD-1028: diffco-us drew
+  ~1,000 reads/h from lead-service's serve loop) is skipped WITHOUT a call for
+  `POSTS_FAILED_RETRY_MS` (1h), same answer; an unreadable 200/4xx stops the chain.
   No provider resolves `showcase/eimmigration` (all three probed).
 - **`doneReason` says WHY a serve is done** (`/search/next`, engagement only, null
   while `done=false`): `no_recent_posts` = no competitor page has a post published
@@ -1137,6 +1140,9 @@ campaign. features-service prices workflows from runs, so it read 11x ROI.
   `actualCostInUsdCents`, descendants included. Other-org spend counts (the
   reveal cache is fleet-wide) and is split out as `paidByOtherOrgsCostInUsdCents`.
   A paying run runs-service does not return lands in `unpricedRunIds`, never as 0.
+- **Fast (BOX-LOAD-1028, was ~1.6 s mean, the top apollo statement):** each person's
+  paid rows are kept in memory per key (person id / email) for 10 min, empty included;
+  each SQL branch hits a covering index (migration 0047). Do not add a branch without one.
 - **Nothing is billed, declared or moved.** Declaring the cost again on the
   serving run would bill the org twice; the earlier payer stays the payer.
 - Prod replay 2026-10-17 (aldhanab campaigns a41c4e52 + 3de6bdb2): 382 served,
@@ -1504,9 +1510,9 @@ paying (/enrich, /match, firmographics, job postings, audience companies, Linked
 pages). A new caller follows the same shape: a lock taken before the cache check
 makes every hit a transaction + lock, and a waiter holds a pool connection and a
 Postgres backend for the whole wait. Where a miss is slow and NEVER cached (a
-LinkedIn page no provider can read is re-read by every serve, ~2.4 s under the
-lock), wrap the locked part in `singleFlight(key, …)`: concurrent callers in this
-process share the one read; a follower whose leader threw runs its own.
+LinkedIn page every provider fails on, before it is parked), wrap the locked part
+in `singleFlight(key, …)`: concurrent callers in this process share the one read; a
+follower whose leader threw runs its own.
 
 ## U+0000 in provider data is stripped at the DRIVER, never per field
 

drizzle/0047_box_load_apollo_waste.sql

@@ -0,0 +1,16 @@
+-- BOX-LOAD-1028-APOLLO-WASTE.
+-- (1) A competitor page every posts provider failed on is remembered for a short
+-- backoff (POSTS_FAILED_RETRY_MS) instead of being re-read by every serve: diffco-us
+-- drew ~1,000 failed reads an hour (2026-10-27/28, ~1.2 s each, tikhub 400).
+ALTER TABLE "linkedin_company_pages" ADD COLUMN IF NOT EXISTS "posts_failed_at" timestamp with time zone;
+--> statement-breakpoint
+ALTER TABLE "linkedin_company_pages" ADD COLUMN IF NOT EXISTS "posts_failure" text;
+--> statement-breakpoint
+-- (2) POST /internal/acquisition-costs: covering indexes matching each branch's
+-- predicate, so a lookup never reads the wide heap (~1.6 s mean per call before,
+-- mostly heap reads of apollo_people_enrichments and email_verifications).
+CREATE INDEX IF NOT EXISTS idx_enrichments_person_acq ON apollo_people_enrichments (apollo_person_id) INCLUDE (enrichment_run_id, org_id) WHERE enrichment_run_id IS NOT NULL;
+--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS idx_enrichments_email_norm_acq ON apollo_people_enrichments (lower(btrim(email))) INCLUDE (enrichment_run_id, org_id) WHERE email IS NOT NULL AND enrichment_run_id IS NOT NULL;
+--> statement-breakpoint
+CREATE INDEX IF NOT EXISTS idx_email_verifications_billed_acq ON email_verifications (email) INCLUDE (verify_run_id, org_id) WHERE billed AND verify_run_id IS NOT NULL;

drizzle/meta/_journal.json

@@ -330,6 +330,13 @@
       "when": 1781395200000,
       "tag": "0046_enrichments_email_norm",
       "breakpoints": true
+    },
+    {
+      "idx": 47,
+      "version": "7",
+      "when": 1781481600000,
+      "tag": "0047_box_load_apollo_waste",
+      "breakpoints": true
     }
   ]
 }

src/db/schema.ts

@@ -720,6 +720,11 @@ export const linkedinCompanyPages = pgTable("linkedin_company_pages", {
   // "ok" | "not_found" (no posts provider knows the page; re-checked after the refresh period). null = ok (pre-0036 rows).
   postsStatus: text("posts_status"),
   postsError: text("posts_error"),
+  // Last time EVERY posts provider failed on the page (none said "not found" for all,
+  // none answered posts): the page is skipped without a provider call until
+  // POSTS_FAILED_RETRY_MS has passed (migration 0047). posts_failure = that read's reason.
+  postsFailedAt: timestamp("posts_failed_at", { withTimezone: true }),
+  postsFailure: text("posts_failure"),
 });
 
 // Silver: one post of a company page. `published_at` is APPROXIMATE (LinkedIn

src/lib/acquisition-costs.ts

@@ -178,40 +178,101 @@ export function attributeAcquisitions(
 }
 
 interface RawRow {
-  idx: number | string;
+  by: "pid" | "email";
+  key: string;
   kind: AcquisitionKind;
   run_id: string;
   org_id: string;
 }
 
+type FoundAcquisition = Omit<AcquisitionRow, "idx">;
+
+/**
+ * BOX-LOAD-1028: lead-service re-asks the same orgs' whole served stock every few
+ * minutes (features-service polls it), ~1,300 people per call, and this lookup was
+ * the biggest apollo statement on the box (~1.6 s mean, 14k calls in 40 h). What an
+ * EARLIER run paid for a person only grows by new acquisitions, which start after
+ * the serves already asked about, so a person's paid rows are kept per lookup key
+ * (apollo person id / normalized email) for ACQUISITION_ROWS_TTL_MS, empty answers
+ * included, and only keys not seen recently go to Postgres. Same TTL as the run
+ * prices (RUN_COST_TTL_MS): a new acquisition shows within 10 minutes.
+ */
+export const ACQUISITION_ROWS_TTL_MS = 10 * 60_000;
+const ACQUISITION_ROWS_CACHE_MAX = 200_000;
+const acquisitionRowsCache = new Map<string, { rows: FoundAcquisition[]; until: number }>();
+const pidKey = (pid: string) => `pid:${pid}`;
+const emailKey = (email: string) => `email:${email}`;
+
+/** Test seam. */
+export function __resetAcquisitionRowsCache(): void {
+  acquisitionRowsCache.clear();
+}
+
 /** Every PAID acquisition of the asked people in our tables, any org. */
 export async function findAcquisitionRows(people: ServedLead[]): Promise<AcquisitionRow[]> {
-  const asked = people.map((p, idx) => ({ idx, pid: p.apolloPersonId ?? null, email: normEmail(p.email) }));
-  const rows = (await db.execute(sql`
-    WITH asked AS (
-      SELECT * FROM jsonb_to_recordset(${JSON.stringify(asked)}::jsonb) AS a(idx int, pid text, email text)
-    )
-    SELECT a.idx, 'apollo_reveal' AS kind, e.enrichment_run_id AS run_id, e.org_id::text AS org_id
-      FROM asked a JOIN apollo_people_enrichments e ON e.apollo_person_id = a.pid
-      WHERE a.pid IS NOT NULL AND e.enrichment_run_id IS NOT NULL
-    UNION
-    SELECT a.idx, 'apollo_reveal', e.enrichment_run_id, e.org_id::text
-      FROM asked a JOIN apollo_people_enrichments e ON lower(btrim(e.email)) = a.email
-      WHERE a.email IS NOT NULL AND e.enrichment_run_id IS NOT NULL
-    UNION
-    SELECT a.idx, 'email_find', f.find_run_id, f.org_id::text
-      FROM asked a JOIN email_findings f ON f.apollo_person_id = a.pid
-      WHERE a.pid IS NOT NULL AND f.find_run_id IS NOT NULL
-    UNION
-    SELECT a.idx, 'email_find', f.find_run_id, f.org_id::text
-      FROM asked a JOIN email_findings f ON lower(btrim(f.email)) = a.email
-      WHERE a.email IS NOT NULL AND f.find_run_id IS NOT NULL
-    UNION
-    SELECT a.idx, 'email_verification', v.verify_run_id, v.org_id::text
-      FROM asked a JOIN email_verifications v ON v.email = a.email
-      WHERE a.email IS NOT NULL AND v.billed AND v.verify_run_id IS NOT NULL
-  `)) as unknown as RawRow[];
-  return rows.map((r) => ({ idx: Number(r.idx), kind: r.kind, runId: r.run_id, orgId: r.org_id }));
+  const now = Date.now();
+  const asked = people.map((p) => ({ pid: p.apolloPersonId ?? null, email: normEmail(p.email) }));
+  const pids = new Set<string>();
+  const emails = new Set<string>();
+  for (const a of asked) {
+    if (a.pid !== null && !((acquisitionRowsCache.get(pidKey(a.pid))?.until ?? 0) > now)) pids.add(a.pid);
+    if (a.email !== null && !((acquisitionRowsCache.get(emailKey(a.email))?.until ?? 0) > now)) emails.add(a.email);
+  }
+
+  const fresh = new Map<string, FoundAcquisition[]>();
+  if (pids.size > 0 || emails.size > 0) {
+    // Each branch matches on exactly the predicate of its covering index
+    // (migration 0047), so every lookup is an index-only scan.
+    const rows = (await db.execute(sql`
+      WITH pids AS (
+        SELECT value AS pid FROM jsonb_array_elements_text(${JSON.stringify([...pids])}::jsonb)
+      ), emails AS (
+        SELECT value AS email FROM jsonb_array_elements_text(${JSON.stringify([...emails])}::jsonb)
+      )
+      SELECT 'pid' AS by, a.pid AS key, 'apollo_reveal' AS kind, e.enrichment_run_id AS run_id, e.org_id::text AS org_id
+        FROM pids a JOIN apollo_people_enrichments e ON e.apollo_person_id = a.pid
+        WHERE e.enrichment_run_id IS NOT NULL
+      UNION
+      SELECT 'email', a.email, 'apollo_reveal', e.enrichment_run_id, e.org_id::text
+        FROM emails a JOIN apollo_people_enrichments e ON lower(btrim(e.email)) = a.email
+        WHERE e.email IS NOT NULL AND e.enrichment_run_id IS NOT NULL
+      UNION
+      SELECT 'pid', a.pid, 'email_find', f.find_run_id, f.org_id::text
+        FROM pids a JOIN email_findings f ON f.apollo_person_id = a.pid
+        WHERE f.find_run_id IS NOT NULL
+      UNION
+      SELECT 'email', a.email, 'email_find', f.find_run_id, f.org_id::text
+        FROM emails a JOIN email_findings f ON lower(btrim(f.email)) = a.email
+        WHERE f.find_run_id IS NOT NULL
+      UNION
+      SELECT 'email', a.email, 'email_verification', v.verify_run_id, v.org_id::text
+        FROM emails a JOIN email_verifications v ON v.email = a.email
+        WHERE v.billed AND v.verify_run_id IS NOT NULL
+    `)) as unknown as RawRow[];
+    for (const pid of pids) fresh.set(pidKey(pid), []);
+    for (const email of emails) fresh.set(emailKey(email), []);
+    for (const r of rows) {
+      const key = r.by === "pid" ? pidKey(r.key) : emailKey(r.key);
+      fresh.get(key)?.push({ kind: r.kind, runId: r.run_id, orgId: r.org_id });
+    }
+    if (acquisitionRowsCache.size + fresh.size > ACQUISITION_ROWS_CACHE_MAX) acquisitionRowsCache.clear();
+    const until = Date.now() + ACQUISITION_ROWS_TTL_MS;
+    for (const [key, found] of fresh) acquisitionRowsCache.set(key, { rows: found, until });
+  }
+
+  const rowsOf = (key: string): FoundAcquisition[] => fresh.get(key) ?? acquisitionRowsCache.get(key)?.rows ?? [];
+  const out: AcquisitionRow[] = [];
+  asked.forEach((a, idx) => {
+    const seen = new Set<string>();
+    const found = [...(a.pid !== null ? rowsOf(pidKey(a.pid)) : []), ...(a.email !== null ? rowsOf(emailKey(a.email)) : [])];
+    for (const f of found) {
+      const k = `${f.kind}:${f.runId}:${f.orgId}`;
+      if (seen.has(k)) continue;
+      seen.add(k);
+      out.push({ idx, ...f });
+    }
+  });
+  return out;
 }
 
 /**

src/lib/linkedin-engagement.ts

@@ -81,6 +81,14 @@ export const PROFILE_MAX_MICRO = 5_000;
 
 export const POSTS_REFRESH_MS = 24 * 3_600_000;
 export const ENGAGEMENT_REFRESH_MS = 24 * 3_600_000;
+/**
+ * A page every provider FAILED on (not a "not found" from all of them) is skipped
+ * without a call for this long, then read again. Before (BOX-LOAD-1028) nothing
+ * remembered it and every serve re-read it: diffco-us drew ~1,000 failed reads an
+ * hour from lead-service's serve loop (tikhub 400 "Request failed. Please retry").
+ * Short on purpose: a provider blip only parks the page for an hour.
+ */
+export const POSTS_FAILED_RETRY_MS = 3_600_000;
 export const PROFILE_CACHE_MS = 30 * 86_400_000;
 /** 100 reactions + 100 comments per page; 5 pages = up to 500 of each per post per day. */
 export const MAX_ENGAGEMENT_PAGES_PER_POST = 5;
@@ -454,17 +462,26 @@ async function readPosts(meter: EngagementMeter, page: CompetitorPage, providers
       notFound.push(verdict.reason);
       continue;
     }
-    if (verdict.kind === "fail") throw new LinkedinTregError(verdict.reason);
+    // An answer we do not understand stops the chain (no other provider is paid) and
+    // fails the page for this serve; it is remembered like any failure.
+    if (verdict.kind === "fail") return { kind: "failed", reason: verdict.reason };
     return verdict;
   }
   if (notFound.length === providers.length) return { kind: "not_found", reason: `no posts provider knows the page: ${notFound.join(" | ")}` };
   return { kind: "failed", reason: `no posts provider could read the page: ${[...notFound, ...skipped].join(" | ")}` };
 }
 
 /** The page's remembered posts answer while it is fresh; null = due for a read. */
-function freshPostsOutcome(row: typeof linkedinCompanyPages.$inferSelect | undefined, page: CompetitorPage, now: Date): PageOutcome | null {
-  if (!row?.postsFetchedAt || now.getTime() - row.postsFetchedAt.getTime() >= POSTS_REFRESH_MS) return null;
-  return row.postsStatus === "not_found" ? { page, state: "not_found", reason: row.postsError ?? "not found" } : { page, state: "readable" };
+export function freshPostsOutcome(row: typeof linkedinCompanyPages.$inferSelect | undefined, page: CompetitorPage, now: Date): PageOutcome | null {
+  if (row?.postsFetchedAt && now.getTime() - row.postsFetchedAt.getTime() < POSTS_REFRESH_MS) {
+    return row.postsStatus === "not_found" ? { page, state: "not_found", reason: row.postsError ?? "not found" } : { page, state: "readable" };
+  }
+  // Due, but every provider failed on it a moment ago: skipped, no provider call.
+  const failedAt = row?.postsFailedAt;
+  if (failedAt && now.getTime() - failedAt.getTime() < POSTS_FAILED_RETRY_MS && (!row.postsFetchedAt || failedAt > row.postsFetchedAt)) {
+    return { page, state: "failed", reason: row.postsFailure ?? "every posts provider failed" };
+  }
+  return null;
 }
 
 async function ensurePosts(meter: EngagementMeter, page: CompetitorPage, now: Date): Promise<PageOutcome> {
@@ -473,9 +490,8 @@ async function ensurePosts(meter: EngagementMeter, page: CompetitorPage, now: Da
   const [seen] = await db.select().from(linkedinCompanyPages).where(eq(linkedinCompanyPages.slug, page.slug)).limit(1);
   const known = freshPostsOutcome(seen, page, now);
   if (known) return known;
-  // A page no provider can read is never remembered, so every serve re-reads it
-  // (~2.4 s of provider calls under the lock): concurrent serves of the same page
-  // share that one read instead of queuing on the lock and repeating it.
+  // Concurrent serves of a due page share one read instead of queuing on the lock
+  // and repeating it (a failed read is remembered only once it returns).
   const outcome = await singleFlight(`linkedin-page:${page.slug}`, () => readPageUnderLock(meter, page, now));
   return { ...outcome, page };
 }
@@ -488,10 +504,16 @@ async function readPageUnderLock(meter: EngagementMeter, page: CompetitorPage, n
     if (fresh) return fresh;
 
     const read = await readPosts(meter, page, POSTS_PROVIDERS);
-    if (read.kind === "failed") return { page, state: "failed", reason: read.reason };
+    if (read.kind === "failed") {
+      // Remembered for POSTS_FAILED_RETRY_MS so the next serves skip it without a call.
+      // The posts already listed (posts_fetched_at, posts_count, posts_status) stay.
+      const failed = { postsFailedAt: now, postsFailure: read.reason.slice(0, 1000) };
+      await tx.insert(linkedinCompanyPages).values({ slug: page.slug, url: page.url, ...failed }).onConflictDoUpdate({ target: linkedinCompanyPages.slug, set: failed });
+      return { page, state: "failed", reason: read.reason };
+    }
     if (read.kind === "not_found") {
       // A permanent answer: remembered for the refresh period so no serve pays to re-learn it.
-      const set = { postsFetchedAt: now, postsCount: 0, postsStatus: "not_found", postsError: read.reason.slice(0, 1000) };
+      const set = { postsFetchedAt: now, postsCount: 0, postsStatus: "not_found", postsError: read.reason.slice(0, 1000), postsFailedAt: null, postsFailure: null };
       await tx.insert(linkedinCompanyPages).values({ slug: page.slug, url: page.url, ...set }).onConflictDoUpdate({ target: linkedinCompanyPages.slug, set });
       return { page, state: "not_found", reason: read.reason };
     }
@@ -505,7 +527,7 @@ async function readPageUnderLock(meter: EngagementMeter, page: CompetitorPage, n
         .values({ postId, pageSlug: page.slug, postUrl: p.url ?? null, text: p.text ?? null, publishedAt: published && !Number.isNaN(published.getTime()) ? published : null })
         .onConflictDoUpdate({ target: linkedinCompanyPosts.postId, set: { postUrl: sql`excluded.post_url`, text: sql`excluded.text`, lastSeenAt: now } });
     }
-    const set = { postsFetchedAt: now, postsCount: read.posts.length, postsStatus: "ok", postsError: null };
+    const set = { postsFetchedAt: now, postsCount: read.posts.length, postsStatus: "ok", postsError: null, postsFailedAt: null, postsFailure: null };
     await tx.insert(linkedinCompanyPages).values({ slug: page.slug, url: page.url, ...set }).onConflictDoUpdate({ target: linkedinCompanyPages.slug, set });
     return { page, state: "readable" };
   });