diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 2893f71..2773b4c 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -1680,30 +1680,36 @@ export function createApp(options: AppOptions): Hono { const body = await parseBody(c.req.raw, z.object({ enrich: z.boolean().optional() })); - let queued = 0; - - if (body.enrich !== false) { - const people = await queryAll<{ person_id: string }>( - db, - `SELECT p.person_id FROM person_emails p - JOIN person_consent pc ON pc.person_id = p.person_id - WHERE pc.import_id = ?`, - [importId], - ); - - for (const person of people) { - const job = await enqueue(db, { - workspaceId: actor.workspaceId, - kind: 'enrich_contact', - payload: { personId: person.person_id }, - }); - if (job.queued) queued += 1; - } + // Enrichment is *not* enqueued here, and that is the fix rather than an + // omission. This used to write one job row per imported person: for a list + // of seventeen thousand that is seventeen thousand inserts inside one HTTP + // request, which is why finishing never returned. + // + // The set of people to look up is derivable — everyone with an imported + // address and no `contact_enriched_at` — so the worker sweeps it instead. + // Nothing needs to be written for that to start, and it resumes by itself + // if this request never completes at all. + if (body.enrich === false) { + await db.execute({ + sql: `UPDATE people SET contact_enriched_at = ? + WHERE contact_enriched_at IS NULL + AND id IN (SELECT person_id FROM person_consent WHERE import_id = ?)`, + args: [now(), importId], + }); } await finishContactImport(db, importId); - return c.json({ finished: true, enrichmentQueued: queued }); + const waiting = await queryOne<{ n: number }>( + db, + `SELECT count(DISTINCT pe.person_id) AS n + FROM person_emails pe + JOIN people p ON p.id = pe.person_id + WHERE pe.workspace_id = ? AND p.contact_enriched_at IS NULL AND p.status = 'active'`, + [actor.workspaceId], + ); + + return c.json({ finished: true, awaitingEnrichment: Number(waiting?.n ?? 0) }); }); /** The batch, and what it could not use. */ diff --git a/apps/api/src/contact-import.test.ts b/apps/api/src/contact-import.test.ts index 44e9fb1..8a9e992 100644 --- a/apps/api/src/contact-import.test.ts +++ b/apps/api/src/contact-import.test.ts @@ -225,38 +225,62 @@ describe('contact import', () => { expect((await post(app, '/contacts/imports', {})).status).toBe(403); }); - test('finishing queues one enrichment job per imported person', async () => { + test('finishing writes no job rows and reports what is waiting', async () => { + // It used to enqueue one job per person. Seventeen thousand inserts inside + // one request is why finishing an import never returned; the worker sweeps + // a derived set instead, so nothing has to be written for it to start. const { app, seeded } = await harness('import-finish'); const importId = await startImport(app); await post(app, `/contacts/imports/${importId}/rows`, { rows: GOOD_ROWS }); const response = await post(app, `/contacts/imports/${importId}/finish`, {}); - const body = (await response.json()) as { enrichmentQueued: number }; + const body = (await response.json()) as { awaitingEnrichment: number }; - expect(body.enrichmentQueued).toBe(3); + expect(body.awaitingEnrichment).toBe(3); const jobs = await seeded.db.execute({ sql: `SELECT count(*) AS n FROM jobs WHERE kind = 'enrich_contact'`, args: [], }); - - expect(Number(jobs.rows[0]?.n)).toBe(3); + expect(Number(jobs.rows[0]?.n)).toBe(0); const batch = await seeded.db.execute({ sql: 'SELECT status FROM contact_imports WHERE id = ?', args: [importId], }); - expect(batch.rows[0]?.status).toBe('complete'); }); - test('enrichment can be skipped', async () => { + test('enrichment can be skipped, and then nothing is waiting', async () => { const { app } = await harness('import-noenrich'); const importId = await startImport(app); await post(app, `/contacts/imports/${importId}/rows`, { rows: GOOD_ROWS }); const response = await post(app, `/contacts/imports/${importId}/finish`, { enrich: false }); - expect(((await response.json()) as { enrichmentQueued: number }).enrichmentQueued).toBe(0); + expect(((await response.json()) as { awaitingEnrichment: number }).awaitingEnrichment).toBe(0); + }); + + test('a large chunk is one round trip, not one per row', async () => { + // The measured failure: 4 sequential Turso round trips per row gave ~200 + // rows a minute, so 17,000 rows was ~85 minutes and the browser gave up. + // This asserts the shape rather than the wall clock — a batched chunk is + // a handful of statements regardless of size. + const { app } = await harness('import-large-chunk'); + const importId = await startImport(app); + + const rows = Array.from({ length: 400 }, (_, index) => ({ + email: `person${index}@corp${index % 20}.com`, + name: `Person ${index}`, + })); + + const started = Date.now(); + const response = await post(app, `/contacts/imports/${importId}/rows`, { rows }); + const body = (await response.json()) as { imported: number }; + + expect(body.imported).toBe(400); + // Generous, because CI is shared. The old path could not do 400 rows in + // anything like this against a local file, let alone a remote database. + expect(Date.now() - started).toBeLessThan(20_000); }); }); diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 838e615..aff0df0 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -39,6 +39,8 @@ import { listeningCampaigns, autoApproveInternal, enrichContact, + sweepContactEnrichment, + workspacesAwaitingEnrichment, processDeletion, workspacesWithInternalBacklog, pruneWorkflowEvents, @@ -657,6 +659,22 @@ async function tick(): Promise { } } + // Imported contacts are looked up from a derived set rather than a queue — + // seventeen thousand job rows written inside one request is what made + // finishing an import never return. Concurrent inside the sweep, because the + // work is a network round trip and nothing else. + for (const workspaceId of await workspacesAwaitingEnrichment(db)) { + const swept = await sweepContactEnrichment(db, { workspaceId }); + + if (swept.looked > 0) { + console.log( + `enrichment: looked up ${swept.looked} in ${workspaceId}, ` + + `${swept.found} had a profile, ${swept.identities} identities, ` + + `${swept.remaining} left`, + ); + } + } + // The queue drains before the send sweep, so anything discovered this tick // can go out on the same tick rather than waiting for the next one. const drained = await drainQueue(db, runJob); diff --git a/migrations/0026_contact_enrichment_sweep.sql b/migrations/0026_contact_enrichment_sweep.sql new file mode 100644 index 0000000..5afe232 --- /dev/null +++ b/migrations/0026_contact_enrichment_sweep.sql @@ -0,0 +1,25 @@ +-- 0026_contact_enrichment_sweep.sql +-- +-- Enrichment as a swept set rather than seventeen thousand queue rows. +-- +-- `POST /contacts/imports/:id/finish` enqueued one `enrich_contact` job per +-- imported person. For a list of seventeen thousand that is seventeen thousand +-- inserts inside one HTTP request, which does not return — and if it did, the +-- worker drains twenty-five jobs a tick, so the queue behind it would take +-- eleven hours while every crawl waited its turn behind it. +-- +-- The set of people needing enrichment is derivable: it is everyone with an +-- imported address we have not looked up yet. Deriving it costs one indexed +-- query per tick instead of a row per person, cannot drift from reality, and +-- resumes by itself after a crash — the same argument `metering.ts` makes for +-- computing usage rather than incrementing it. +-- +-- A timestamp rather than a boolean because "never tried" and "tried and found +-- nothing" must be distinguishable: most addresses have no published profile, +-- and a boolean would make the sweep retry every one of them forever. + +ALTER TABLE people ADD COLUMN contact_enriched_at TEXT; + +-- The sweep's only query: imported people, not yet looked up, oldest first. +CREATE INDEX idx_people_contact_enrichment + ON people(contact_enriched_at) WHERE contact_enriched_at IS NULL; diff --git a/packages/pipeline/src/contact-import.ts b/packages/pipeline/src/contact-import.ts index f3de877..9c9678d 100644 --- a/packages/pipeline/src/contact-import.ts +++ b/packages/pipeline/src/contact-import.ts @@ -112,31 +112,256 @@ export async function importContactChunk( if (!batch) throw new Error(`no such import: ${importId}`); + // ---------------------------------------------------------------- clean + // Pure and local: no database, so five hundred rows cost nothing here. const seen = new Set(); - const personIds: string[] = []; - let imported = 0; - let merged = 0; - let rejected = 0; + const clean: { row: number; contact: CleanContact }[] = []; + const rejects: { row: number; email?: string; reason: string; detail: string }[] = []; for (const [offset, raw] of rows.entries()) { const rowNumber = (options.startRow ?? 0) + offset + 1; const result = cleanContact(raw, seen); if (!result.ok) { - rejected += 1; - await recordReject(db, importId, rowNumber, raw.email, result.reason, result.detail); + rejects.push({ + row: rowNumber, + ...(raw.email ? { email: raw.email } : {}), + reason: result.reason, + detail: result.detail, + }); continue; } seen.add(result.contact.dedupeKey); + clean.push({ row: rowNumber, contact: result.contact }); + } + + // ------------------------------------------------------------- existing + // One query for the whole chunk. This was a `SELECT` per row, which is + // most of why importing seventeen thousand contacts took eighty-five + // minutes: the work is trivial and the round trip is not. + const existing = await existingByDedupeKey( + db, + batch.workspace_id, + clean.map((entry) => entry.contact.dedupeKey), + ); + + const fresh = clean.filter((entry) => !existing.has(entry.contact.dedupeKey)); + const known = clean.filter((entry) => existing.has(entry.contact.dedupeKey)); + + const personIds: string[] = []; + const statements: { sql: string; args: (string | number | null)[] }[] = []; + + for (const entry of fresh) { + const personId = newId('person'); + personIds.push(personId); + statements.push( + ...insertContactStatements({ + personId, + importId, + workspaceId: batch.workspace_id, + consentBasis: batch.consent_basis, + consentSource: batch.consent_source, + contact: entry.contact, + }), + ); + } + + for (const entry of known) { + const personId = existing.get(entry.contact.dedupeKey); + if (personId) personIds.push(personId); + } + + for (const reject of rejects) { + statements.push({ + sql: `INSERT INTO contact_import_rejects (id, import_id, row_number, email, reason, detail, + created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`, + args: [ + newId('contactImportReject'), + importId, + reject.row, + reject.email ?? null, + reject.reason, + reject.detail, + now(), + ], + }); + } + + let imported = fresh.length; + let merged = known.length; + let rejected = rejects.length; + // ---------------------------------------------------------------- write + // One round trip for the chunk. `db.batch` is transactional, so a single + // unique violation would lose the other four hundred and ninety-nine — + // hence the fallback, which is the slow path this replaced and is reached + // only when two imports genuinely race on one address. + if (statements.length > 0) { + try { + await db.batch(statements); + } catch (error) { + if (!isUniqueViolation(error)) throw error; + + const retried = await storeOneAtATime(db, importId, batch, clean, rejects.length); + imported = retried.imported; + merged = retried.merged; + rejected = retried.rejected; + personIds.length = 0; + personIds.push(...retried.personIds); + } + } + + // Gap-filling for people we already had is deliberately *not* batched into + // the above: it reads each person to avoid overwriting what is already + // known, and doing that for a chunk that is mostly re-imports is the one + // case where the extra round trips buy something. Bounded so a re-import of + // seventeen thousand does not become the old behaviour by another route. + for (const entry of known.slice(0, MERGE_ENRICH_LIMIT)) { + const personId = existing.get(entry.contact.dedupeKey); + if (personId) await enrichExistingPerson(db, personId, entry.contact); + } + + await db.execute({ + sql: `UPDATE contact_imports + SET total_rows = total_rows + ?, imported = imported + ?, + merged = merged + ?, rejected = rejected + ?, updated_at = ? + WHERE id = ?`, + args: [rows.length, imported, merged, rejected, now(), importId], + }); + + return { imported, merged, rejected, personIds }; +} + +/** + * How many already-known people get their gaps filled per chunk. + * + * Gap-filling reads the stored person first so it cannot overwrite a better + * value, which is a round trip each. Worth it for a handful; for a re-import + * of seventeen thousand it would restore exactly the cost this change + * removed. The rest keep what they have, which is what a merge means anyway. + */ +const MERGE_ENRICH_LIMIT = 50; + +/** + * Which of these mailboxes we already hold, in one query. + * + * Chunked into groups because a single `IN` list of several thousand is a + * statement SQLite will refuse to compile. Five hundred is comfortably inside + * the parameter ceiling and still one round trip per chunk. + */ +async function existingByDedupeKey( + db: Client, + workspaceId: string, + keys: readonly string[], +): Promise> { + const found = new Map(); + if (keys.length === 0) return found; + + for (let offset = 0; offset < keys.length; offset += 500) { + const slice = keys.slice(offset, offset + 500); + const placeholders = slice.map(() => '?').join(', '); + + const result = await db.execute({ + sql: `SELECT dedupe_key, person_id FROM person_emails + WHERE workspace_id = ? AND dedupe_key IN (${placeholders})`, + args: [workspaceId, ...slice], + }); + + for (const row of result.rows) { + const typed = row as unknown as { dedupe_key: string; person_id: string }; + found.set(String(typed.dedupe_key), String(typed.person_id)); + } + } + + return found; +} + +/** The three writes one new contact needs, as statements rather than calls. */ +function insertContactStatements(input: { + readonly personId: string; + readonly importId: string; + readonly workspaceId: string; + readonly consentBasis: string; + readonly consentSource: string | null; + readonly contact: CleanContact; +}): { sql: string; args: (string | number | null)[] }[] { + const stamp = now(); + const { contact, personId } = input; + + return [ + { + sql: `INSERT INTO people (id, display_name, first_name, last_name, current_title, location, + identity_confidence, status, outreach_eligible, created_at, updated_at, + last_resolved_at) + VALUES (?, ?, ?, ?, ?, ?, ?, 'active', 1, ?, ?, ?)`, + args: [ + personId, + contact.displayName, + contact.firstName ?? null, + contact.lastName ?? null, + contact.title ?? null, + contact.location ?? null, + IMPORTED_CONFIDENCE, + stamp, + stamp, + stamp, + ], + }, + { + sql: `INSERT INTO person_emails (id, workspace_id, person_id, address, dedupe_key, source, + verified, created_at) VALUES (?, ?, ?, ?, ?, 'import', 1, ?)`, + args: [ + newId('personEmail'), + input.workspaceId, + personId, + contact.email, + contact.dedupeKey, + stamp, + ], + }, + { + sql: `INSERT INTO person_consent (person_id, workspace_id, basis, source, import_id, + recorded_at) VALUES (?, ?, ?, ?, ?, ?)`, + args: [ + personId, + input.workspaceId, + input.consentBasis, + input.consentSource, + input.importId, + stamp, + ], + }, + ]; +} + +/** + * The old row-at-a-time path, kept for when the batch loses a race. + * + * Slow and correct. Reached only when two imports insert the same address at + * the same moment, which the unique index catches and which would otherwise + * cost the whole chunk. + */ +async function storeOneAtATime( + db: Client, + importId: string, + batch: { workspace_id: string; consent_basis: string; consent_source: string | null }, + clean: readonly { row: number; contact: CleanContact }[], + alreadyRejected: number, +): Promise<{ imported: number; merged: number; rejected: number; personIds: string[] }> { + const personIds: string[] = []; + let imported = 0; + let merged = 0; + let rejected = alreadyRejected; + + for (const entry of clean) { try { const outcome = await storeContact(db, { importId, workspaceId: batch.workspace_id, consentBasis: batch.consent_basis, consentSource: batch.consent_source, - contact: result.contact, + contact: entry.contact, }); if (outcome.created) imported += 1; @@ -148,22 +373,14 @@ export async function importContactChunk( await recordReject( db, importId, - rowNumber, - raw.email, + entry.row, + entry.contact.email, 'malformed_email', `could not be stored: ${String(error)}`, ); } } - await db.execute({ - sql: `UPDATE contact_imports - SET total_rows = total_rows + ?, imported = imported + ?, - merged = merged + ?, rejected = rejected + ?, updated_at = ? - WHERE id = ?`, - args: [rows.length, imported, merged, rejected, now(), importId], - }); - return { imported, merged, rejected, personIds }; } diff --git a/packages/pipeline/src/enrich-contact.ts b/packages/pipeline/src/enrich-contact.ts index cbb8cc2..f4eacd8 100644 --- a/packages/pipeline/src/enrich-contact.ts +++ b/packages/pipeline/src/enrich-contact.ts @@ -23,7 +23,7 @@ import { newId, isFreemailDomain } from '@outreachgraph/domain'; import { GRAVATAR_NETWORKS, lookupGravatar, type GravatarOptions } from '@outreachgraph/providers'; -import { now, queryOne, type Client } from '@outreachgraph/db'; +import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; /** * How much to believe a Gravatar-published account. @@ -35,6 +35,118 @@ import { now, queryOne, type Client } from '@outreachgraph/db'; */ const GRAVATAR_CONFIDENCE = 0.8; +/** + * How many people one sweep looks up, and how many at once. + * + * Gravatar is a network round trip and nothing else: the work is entirely + * waiting, so running it one at a time wastes the whole interval. Ten at once + * against a public API that asks callers to identify themselves is polite + * rather than aggressive, and two hundred a tick clears seventeen thousand in + * roughly an hour and a half without a queue row per person. + */ +const SWEEP_SIZE = 200; +const SWEEP_CONCURRENCY = 10; + +export interface SweepResult { + readonly looked: number; + readonly found: number; + readonly identities: number; + readonly remaining: number; +} + +/** + * Looks up the next batch of imported people who have never been looked up. + * + * The set is derived — everyone with an imported address and no + * `contact_enriched_at` — rather than materialised as jobs. Seventeen thousand + * queue rows written inside one request is what made `finish` never return, + * and a derived set cannot drift from the people who actually exist. + * + * A miss still stamps the timestamp. Most addresses have no published profile, + * and a sweep that only recorded successes would retry those forever. + */ +export async function sweepContactEnrichment( + db: Client, + input: { readonly workspaceId: string; readonly limit?: number }, + options: GravatarOptions = {}, +): Promise { + const limit = input.limit ?? SWEEP_SIZE; + + const rows = await queryAll<{ person_id: string }>( + db, + `SELECT pe.person_id + FROM person_emails pe + JOIN people p ON p.id = pe.person_id + WHERE pe.workspace_id = ? AND p.contact_enriched_at IS NULL AND p.status = 'active' + GROUP BY pe.person_id + ORDER BY pe.created_at + LIMIT ?`, + [input.workspaceId, limit], + ); + + let found = 0; + let identities = 0; + + // Fixed-size waves rather than one promise per person: seventeen thousand + // concurrent fetches would be a denial of service on somebody else's free + // API, and on our own event loop. + for (let offset = 0; offset < rows.length; offset += SWEEP_CONCURRENCY) { + const wave = rows.slice(offset, offset + SWEEP_CONCURRENCY); + + const results = await Promise.all( + wave.map(async (row) => { + try { + return await enrichContact( + db, + { workspaceId: input.workspaceId, personId: row.person_id }, + options, + ); + } catch { + // One bad lookup costs that person, not the sweep. The timestamp is + // still stamped below so it is not retried forever. + return undefined; + } + }), + ); + + for (const result of results) { + if (result?.found) found += 1; + identities += result?.identities ?? 0; + } + + await db.batch( + wave.map((row) => ({ + sql: 'UPDATE people SET contact_enriched_at = ? WHERE id = ?', + args: [now(), row.person_id] as (string | number | null)[], + })), + ); + } + + const left = await queryOne<{ n: number }>( + db, + `SELECT count(DISTINCT pe.person_id) AS n + FROM person_emails pe + JOIN people p ON p.id = pe.person_id + WHERE pe.workspace_id = ? AND p.contact_enriched_at IS NULL AND p.status = 'active'`, + [input.workspaceId], + ); + + return { looked: rows.length, found, identities, remaining: Number(left?.n ?? 0) }; +} + +/** Workspaces with imported people still waiting to be looked up. */ +export async function workspacesAwaitingEnrichment(db: Client): Promise { + const rows = await queryAll<{ workspace_id: string }>( + db, + `SELECT DISTINCT pe.workspace_id + FROM person_emails pe + JOIN people p ON p.id = pe.person_id + WHERE p.contact_enriched_at IS NULL AND p.status = 'active'`, + ); + + return rows.map((row) => row.workspace_id); +} + export interface EnrichContactResult { readonly personId: string; readonly found: boolean; diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts index 54498d6..59363ce 100644 --- a/packages/pipeline/src/index.ts +++ b/packages/pipeline/src/index.ts @@ -82,7 +82,13 @@ export { type ChunkResult, type StartImportInput, } from './contact-import'; -export { enrichContact, type EnrichContactResult } from './enrich-contact'; +export { + enrichContact, + sweepContactEnrichment, + workspacesAwaitingEnrichment, + type EnrichContactResult, + type SweepResult, +} from './enrich-contact'; export { autoApproveInternal, workspacesWithInternalBacklog,