Import seventeen thousand contacts in minutes, not eighty-five

apps/api/src/app.ts

@@ -1680,30 +1680,36 @@ export function createApp(options: AppOptions): Hono<AppEnv> {
 
     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. */

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);
   });
 });

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<void> {
     }
   }
 
+  // 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);

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;