diff --git a/.env.example b/.env.example index 92645a3..b109542 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,14 @@ APOLLO_API_KEY= PDL_API_KEY= GITHUB_TOKEN= + +# Finds a lead's photo (and their LinkedIn profile URL, as research) by +# searching their name, title and company. One image search per person, +# roughly a quarter of a cent each. Unset, only Gravatar supplies pictures. +VALUESERP_API_KEY= +# Ceiling on photo lookups per workspace per day. Default 300. +PHOTO_LOOKUPS_PER_DAY= + BLUESKY_IDENTIFIER= BLUESKY_APP_PASSWORD= X_API_KEY= diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 3946782..260fe9d 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -1683,7 +1683,8 @@ export function createApp(options: AppOptions): Hono { const limit = clampLimit(c.req.query('limit')); const rows = await c.get('db').execute({ - sql: `SELECT p.id, p.display_name, p.current_title, p.identity_confidence, p.status, + sql: `SELECT p.id, p.display_name, p.current_title, p.avatar_url, p.identity_confidence, + p.status, co.name AS current_company, cp.status AS prospect_status, cp.interaction_state, s.opportunity, s.icp_fit, s.intent, s.reachability, diff --git a/apps/api/src/repository.ts b/apps/api/src/repository.ts index 2a190a8..6cf8b98 100644 --- a/apps/api/src/repository.ts +++ b/apps/api/src/repository.ts @@ -28,6 +28,7 @@ export interface PersonRow { current_title: string | null; current_company_id: string | null; location: string | null; + avatar_url?: string | null; identity_confidence: number; status: string; outreach_eligible: number; @@ -190,7 +191,7 @@ export async function listPendingRecommendations( return queryAll( db, - `SELECT r.*, p.display_name, p.current_title, p.identity_confidence, + `SELECT r.*, p.display_name, p.current_title, p.avatar_url, p.identity_confidence, s.summary AS signal_summary, s.source_url AS signal_url, s.source_timestamp AS signal_at, d.body AS draft_body, d.subject AS draft_subject, @@ -475,6 +476,15 @@ export async function resolveContactAddress( ); if (personal?.handle) return { address: personal.handle.trim().toLowerCase(), shared: false }; + // An imported address is the person's own too: they gave it to us, with + // consent recorded on the import. Personal, not shared. + const imported = await queryOne<{ address: string }>( + db, + `SELECT address FROM person_emails WHERE person_id = ? ORDER BY created_at LIMIT 1`, + [personId], + ); + if (imported?.address) return { address: imported.address.trim().toLowerCase(), shared: false }; + const company = await queryOne<{ contact_email: string }>( db, `SELECT co.contact_email FROM people p @@ -578,10 +588,15 @@ export async function pendingAddressUsage( `WITH candidate AS ( SELECT r.id AS recommendation_id, r.campaign_id AS campaign_id, EXISTS (SELECT 1 FROM drafts d WHERE d.recommendation_id = r.id) AS has_draft, - (SELECT lower(trim(si.handle)) FROM social_identities si - WHERE si.person_id = r.person_id AND si.network = 'email' - AND si.handle IS NOT NULL AND trim(si.handle) <> '' - ORDER BY si.confidence DESC LIMIT 1) AS personal, + COALESCE( + (SELECT lower(trim(si.handle)) FROM social_identities si + WHERE si.person_id = r.person_id AND si.network = 'email' + AND si.handle IS NOT NULL AND trim(si.handle) <> '' + ORDER BY si.confidence DESC LIMIT 1), + (SELECT lower(trim(pe.address)) FROM person_emails pe + WHERE pe.person_id = r.person_id AND pe.workspace_id = r.workspace_id + ORDER BY pe.created_at LIMIT 1) + ) AS personal, (SELECT lower(trim(co.contact_email)) FROM people p JOIN companies co ON co.id = p.current_company_id WHERE p.id = r.person_id AND co.contact_email IS NOT NULL diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 8123014..42430d9 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -47,7 +47,10 @@ import { pruneWorkflowEvents, regenerateRecommendations, rescoreProspect, + reseedIdleCampaigns, runAutopilot, + sweepProfilePhotos, + workspacesAwaitingPhotos, runCadences, runCrawlJob, runDiscoveryJob, @@ -67,6 +70,7 @@ import { RedditSource, RssSource, SiteProvider, + ValueSerpClient, type FeedSource, } from '@outreachgraph/providers'; @@ -237,6 +241,21 @@ if (!encryptionKey) { console.log('no SECRET_ENCRYPTION_KEY: workspaces cannot connect their own sending mailbox'); } +/** + * Pictures of leads, by search. + * + * Optional, like every provider: unset, Gravatar is the only source and most + * people stay as initials. Each lookup is a paid request, so the sweep is + * capped per workspace per day as well as per tick. + */ +const photoFinder = process.env.VALUESERP_API_KEY + ? new ValueSerpClient({ apiKey: process.env.VALUESERP_API_KEY }) + : undefined; + +const photoLookupsPerDay = Number(process.env.PHOTO_LOOKUPS_PER_DAY ?? 300); + +if (!photoFinder) console.log('no VALUESERP_API_KEY: lead photos come from Gravatar only'); + /** * The feed clients for one campaign's own targets. * @@ -682,6 +701,42 @@ async function tick(): Promise { } } + // A face for the people about to be written to. Bounded per tick and per + // day inside the sweep; this loop only decides who is looked at. + if (photoFinder) { + for (const workspaceId of await workspacesAwaitingPhotos(db)) { + try { + const swept = await sweepProfilePhotos( + { db, finder: photoFinder, dailyCap: photoLookupsPerDay }, + { workspaceId }, + ); + + if (swept.looked > 0) { + console.log( + `photos: looked up ${swept.looked} in ${workspaceId}, ${swept.found} found, ` + + `${swept.profiles} LinkedIn profile(s) recorded, ${swept.remainingToday} left today`, + ); + } + } catch (error) { + console.error(`photo sweep failed for ${workspaceId}`, error); + } + } + } + + // Campaigns whose seed has not been read in a week are asked again. This is + // the only thing that puts work back into an idle queue: once the first crop + // of research cards is cleared, nothing else ever asks the seed for more. + for (const workspace of workspaces) { + try { + const reseeded = await reseedIdleCampaigns(db, { workspaceId: workspace.id }); + if (reseeded.queued > 0) { + console.log(`reseeded ${reseeded.queued} idle campaign(s) in ${workspace.id}`); + } + } catch (error) { + console.error(`reseed failed for ${workspace.id}`, error); + } + } + // 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, { isOutage: isBudgetExhausted }); diff --git a/apps/web/app/(app)/prospects/[id]/page.tsx b/apps/web/app/(app)/prospects/[id]/page.tsx index d697e94..d51c31a 100644 --- a/apps/web/app/(app)/prospects/[id]/page.tsx +++ b/apps/web/app/(app)/prospects/[id]/page.tsx @@ -1,5 +1,6 @@ import Link from 'next/link'; import { notFound, redirect } from 'next/navigation'; +import { Avatar } from '../../../../components/avatar'; import { EmailCandidates } from '../../../../components/email-candidates'; import { EnrolButton } from '../../../../components/enrol-button'; import { PageGuide } from '../../../../components/page-guide'; @@ -65,12 +66,15 @@ export default async function ProspectPage({ params }: { params: Promise<{ id: s ← Prospects -
-

{person.display_name}

-

{person.current_title ?? '—'}

-

- Identity confidence {Math.round((person.identity_confidence ?? 0) * 100)}% -

+
+ +
+

{person.display_name}

+

{person.current_title ?? '—'}

+

+ Identity confidence {Math.round((person.identity_confidence ?? 0) * 100)}% +

+
diff --git a/apps/web/app/(app)/prospects/page.tsx b/apps/web/app/(app)/prospects/page.tsx index 65f95b7..941f266 100644 --- a/apps/web/app/(app)/prospects/page.tsx +++ b/apps/web/app/(app)/prospects/page.tsx @@ -1,5 +1,6 @@ import Link from 'next/link'; import { redirect } from 'next/navigation'; +import { Avatar } from '../../../components/avatar'; import { PageGuide } from '../../../components/page-guide'; import { ApiUnavailableError, NotAuthenticatedError, fetchProspects } from '../../../lib/api'; import type { ProspectRow } from '../../../lib/types'; @@ -75,15 +76,19 @@ function ProspectItem({ person }: { person: ProspectRow }) { href={`/prospects/${person.id}`} className="border-border bg-surface-raised block rounded-2xl border p-4" > -
- {person.display_name} - - {person.opportunity ?? '—'} - +
+ +
+
+ {person.display_name} + + {person.opportunity ?? '—'} + +
+

{subtitle || '—'}

+
-

{subtitle || '—'}

-
Signals
diff --git a/apps/web/components/approval-card.tsx b/apps/web/components/approval-card.tsx index a96d7e2..98fde95 100644 --- a/apps/web/components/approval-card.tsx +++ b/apps/web/components/approval-card.tsx @@ -5,6 +5,7 @@ import { useState } from 'react'; // Imported from the server-safe modules, not lib/api: that pulls in // next/headers and cannot be bundled for the browser. import { relativeTime } from '../lib/format'; +import { Avatar } from './avatar'; import { ShareButtons } from './share-buttons'; import type { ApprovalCard as Card } from '../lib/types'; @@ -166,9 +167,12 @@ export function ApprovalCard({ card }: { card: Card }) { return (
-
-

{card.display_name}

-

{card.current_title ?? '—'}

+
+ +
+

{card.display_name}

+

{card.current_title ?? '—'}

+
diff --git a/apps/web/components/avatar.tsx b/apps/web/components/avatar.tsx new file mode 100644 index 0000000..66a9ab3 --- /dev/null +++ b/apps/web/components/avatar.tsx @@ -0,0 +1,61 @@ +/** + * A face, or the initials standing in for one. + * + * The picture is a URL the person published — Gravatar, their LinkedIn + * profile, their company's team page — and it is hot-linked rather than + * copied, so it can vanish. When it does, or when there never was one, the + * initials take the same space, so a list of people does not jump around + * depending on who has a photograph. + */ + +const SIZES = { + sm: 'h-8 w-8 text-xs', + md: 'h-10 w-10 text-sm', + lg: 'h-16 w-16 text-lg', +} as const; + +export function Avatar({ + name, + src, + size = 'md', + className = '', +}: { + name: string; + src?: string | null; + size?: keyof typeof SIZES; + className?: string; +}) { + const box = `${SIZES[size]} shrink-0 rounded-full ${className}`; + + if (src) { + return ( + // eslint-disable-next-line @next/next/no-img-element -- remote hosts are not known ahead of time + + ); + } + + return ( + + ); +} + +export function initials(name: string): string { + const parts = name + .trim() + .split(/\s+/) + .filter((part) => /[\p{L}\p{N}]/u.test(part)); + const first = parts[0]?.[0] ?? ''; + const last = parts.length > 1 ? (parts[parts.length - 1]?.[0] ?? '') : ''; + return `${first}${last}`.toUpperCase() || '?'; +} diff --git a/apps/web/lib/types.ts b/apps/web/lib/types.ts index d6edcaa..f2d4029 100644 --- a/apps/web/lib/types.ts +++ b/apps/web/lib/types.ts @@ -19,6 +19,7 @@ export interface ApprovalCard { person_id: string; display_name: string; current_title: string | null; + avatar_url?: string | null; action: string; network: string; priority: number; @@ -154,6 +155,7 @@ export interface ProspectRow { id: string; display_name: string; current_title: string | null; + avatar_url?: string | null; current_company: string | null; identity_confidence: number; prospect_status: string; @@ -196,6 +198,7 @@ export interface ProspectDetail { id: string; display_name: string; current_title: string | null; + avatar_url?: string | null; identity_confidence: number; status: string; }; diff --git a/migrations/0033_avatars_and_reseed.sql b/migrations/0033_avatars_and_reseed.sql new file mode 100644 index 0000000..00e255e --- /dev/null +++ b/migrations/0033_avatars_and_reseed.sql @@ -0,0 +1,29 @@ +-- 0033_avatars_and_reseed.sql +-- +-- Two things the digest could not show and one thing the pipeline could not do. +-- +-- A face. The digest and the queue list people by name and title, and a name +-- is a poor handle on a stranger. Gravatar already returned a `thumbnailUrl` +-- for every profile it found and the enrichment step threw it away; a search +-- adapter can find the rest. The URL is stored rather than the bytes: it is a +-- pointer to a picture the person published, not a copy of it, and it can be +-- dropped by clearing one column. +-- +-- `photo_looked_up_at` is a timestamp for the same reason `contact_enriched_at` +-- is: most lookups miss, and a boolean would make the sweep retry every miss +-- forever. A miss stamps the column; only a hit fills `avatar_url`. +-- +-- `reseeded_at` records the last time an active campaign's seed was read +-- again. Every campaign was crawled exactly once, at creation, and a directory +-- that gains members after that day never produced another lead: "sites read +-- 0, new people 0" for ten days straight while six campaigns sat active. + +ALTER TABLE people ADD COLUMN avatar_url TEXT; +-- gravatar | search | site — where the picture came from, for the audit trail. +ALTER TABLE people ADD COLUMN avatar_source TEXT; +ALTER TABLE people ADD COLUMN photo_looked_up_at TEXT; + +CREATE INDEX IF NOT EXISTS idx_people_photo_lookup + ON people(photo_looked_up_at) WHERE photo_looked_up_at IS NULL; + +ALTER TABLE campaigns ADD COLUMN reseeded_at TEXT; diff --git a/packages/email/src/notifications.ts b/packages/email/src/notifications.ts index 4836a38..fd70177 100644 --- a/packages/email/src/notifications.ts +++ b/packages/email/src/notifications.ts @@ -147,6 +147,8 @@ export interface DigestLead { readonly companyName?: string; readonly opportunity?: number; readonly sentTo?: string; + /** A picture they published, when one is known. HTML only; text reads fine without. */ + readonly avatarUrl?: string; } export interface DailyDigest { @@ -155,7 +157,22 @@ export interface DailyDigest { readonly sitesCrawled: number; readonly peopleFound: number; readonly messagesSent: number; + /** Every pending card, whoever will act on it. */ readonly awaitingApproval: number; + /** + * The pending cards autopilot will send itself, and how many of those are + * currently held by a limit. Absent when the workspace has no campaign on + * autopilot, in which case "awaiting approval" says everything. + */ + readonly autopilotQueue?: { readonly total: number; readonly held: number }; + /** + * Pending cards only a human can act on — LinkedIn, X, GitHub — where the + * product drafts and the person acts in the network's own interface. + */ + readonly needsYou?: { + readonly total: number; + readonly byNetwork: Readonly>; + }; readonly repliesReceived?: number; readonly leads: readonly DigestLead[]; /** Campaigns that produced nothing, and why, when the reason is knowable. */ @@ -178,11 +195,28 @@ export function dailyDigestEmail(to: string, digest: DailyDigest, appUrl: string : `${digest.peopleFound} new ${digest.peopleFound === 1 ? 'lead' : 'leads'}, ` + `${digest.messagesSent} sent · OutreachGraph`; + // "Awaiting approval: 1117" was the line that got this digest replied to + // with "why is it not sending?". Most of those cards were not waiting for + // anyone: autopilot was going to send them, and was holding them under a + // limit. The split says which is which, and the notes say what the limit is. + const queue = digest.autopilotQueue; + const needsYou = digest.needsYou; + const networks = needsYou + ? Object.entries(needsYou.byNetwork) + .filter(([, n]) => n > 0) + .map(([network, n]) => `${networkName(network)} ${n}`) + .join(', ') + : ''; + const counts = [ `Sites read: ${digest.sitesCrawled}`, `New people: ${digest.peopleFound}`, `Messages sent: ${digest.messagesSent}`, - `Awaiting approval: ${digest.awaitingApproval}`, + queue + ? `Autopilot queue: ${queue.total}${queue.held > 0 ? ` (${queue.held} held by a limit)` : ''}` + : '', + needsYou ? `Needs you: ${needsYou.total}${networks ? ` (${networks})` : ''}` : '', + !queue && !needsYou ? `Awaiting approval: ${digest.awaitingApproval}` : '', digest.repliesReceived !== undefined ? `Replies: ${digest.repliesReceived}` : '', ].filter(Boolean); @@ -219,19 +253,28 @@ export function dailyDigestEmail(to: string, digest: DailyDigest, appUrl: string row('Sites read', digest.sitesCrawled), row('New people', digest.peopleFound), row('Messages sent', digest.messagesSent), - row('Awaiting approval', digest.awaitingApproval), + queue + ? row( + 'Autopilot queue', + queue.total, + queue.held > 0 ? `${queue.held} held by a limit` : undefined, + ) + : '', + needsYou ? row('Needs you', needsYou.total, networks || undefined) : '', + !queue && !needsYou ? row('Awaiting approval', digest.awaitingApproval) : '', digest.repliesReceived !== undefined ? row('Replies', digest.repliesReceived) : '', '', quiet ? '

Nothing new came back today. Campaigns are still running.

' : digest.leads.length - ? `
    ${digest.leads + ? `
      ${digest.leads .map((lead) => { const where = [lead.title, lead.companyName].filter(Boolean).join(' at '); const score = lead.opportunity !== undefined ? ` [${lead.opportunity}]` : ''; const sent = lead.sentTo ? ' — written to' : ''; return ( - `
    • ` + + `
    • ${avatarHtml(lead)}` + + `` + `${escapeHtml(lead.personName)}` + `${where ? ` — ${escapeHtml(where)}` : ''}${score}${sent}
      • ` ); @@ -248,9 +291,44 @@ export function dailyDigestEmail(to: string, digest: DailyDigest, appUrl: string return { to, subject, text: text + foot.text, html }; } -function row(label: string, value: number): string { +function row(label: string, value: number, note?: string): string { return ( `${escapeHtml(label)}` + - `${value}` + `${value}` + + `${note ? ` ${escapeHtml(note)}` : ''}` + + '' ); } + +/** + * A small round picture before the name, when one is known. + * + * Inline styles and fixed dimensions because mail clients honour little else, + * and `alt=""` because the name follows immediately — a reader with images off + * loses nothing. + */ +function avatarHtml(lead: DigestLead): string { + if (!lead.avatarUrl) return ''; + return ( + `' + ); +} + +function networkName(network: string): string { + switch (network) { + case 'linkedin': + return 'LinkedIn'; + case 'x': + return 'X'; + case 'github': + return 'GitHub'; + case 'bluesky': + return 'Bluesky'; + case 'mastodon': + return 'Mastodon'; + default: + return network; + } +} diff --git a/packages/pipeline/src/autopilot.test.ts b/packages/pipeline/src/autopilot.test.ts index 9552bd2..222ea82 100644 --- a/packages/pipeline/src/autopilot.test.ts +++ b/packages/pipeline/src/autopilot.test.ts @@ -3,7 +3,7 @@ import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; import type { Mailer, Message, SendResult } from '@outreachgraph/email'; import type { GenerateResult, TextModel } from '@outreachgraph/ai'; import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; -import { runAutopilot } from './autopilot'; +import { HoldLedger, describeHold, runAutopilot } from './autopilot'; let seeded: SeededDatabase | undefined; @@ -551,3 +551,191 @@ describe('one mailbox, several colleagues', () => { expect(result.skipped).toHaveLength(1); }); }); + +describe('a queue held at the top', () => { + /** + * The production shape that sent six a day: the highest-priority cards all + * resolve to one shared inbox that has already been written to, and the + * card that could actually go out sits behind them. + */ + async function addHeldColleague(db: Client, name: string, priority: number): Promise { + const stamp = now(); + const personId = `per_${name.toLowerCase()}`; + + await db.execute({ + sql: `INSERT INTO people (id, display_name, current_company_id, status, believed_minor, + outreach_eligible, identity_confidence, created_at, updated_at) + VALUES (?, ?, ?, 'active', 0, 1, 0.95, ?, ?)`, + args: [personId, name, SEED.companyId, stamp, stamp], + }); + + await db.execute({ + sql: `INSERT INTO recommendations (id, workspace_id, campaign_id, person_id, action, network, + priority, reason, policy_status, policy_version, status, created_at) + VALUES (?, ?, ?, ?, 'send_email', 'email', ?, 'same company', + 'allow', '2026-01-01', 'pending', ?)`, + args: [ + `rec_${name.toLowerCase()}`, + SEED.workspaceId, + SEED.campaignId, + personId, + priority, + stamp, + ], + }); + + await db.execute({ + sql: `INSERT INTO drafts (id, workspace_id, recommendation_id, subject, body, checks_json, + created_at, updated_at) + VALUES (?, ?, ?, 'Hello', 'A grounded message.', '[]', ?, ?)`, + args: [ + `drf_${name.toLowerCase()}`, + SEED.workspaceId, + `rec_${name.toLowerCase()}`, + stamp, + stamp, + ], + }); + } + + async function inboxAlreadyWrittenTo(db: Client, address: string): Promise { + await db.execute({ + sql: `INSERT INTO interactions (id, workspace_id, person_id, campaign_id, network, + direction, state, body, contact_address, shared_inbox, occurred_at, recorded_at) + VALUES ('int_earlier', ?, ?, ?, 'email', 'outbound', 'contacted', 'earlier', + ?, 1, ?, ?)`, + args: [SEED.workspaceId, SEED.personId, SEED.campaignId, address, now(), now()], + }); + } + + test('a sendable card behind the held ones is still reached', async () => { + seeded = await seedDatabase('autopilot-starvation'); + const { db } = seeded; + + // Jane can be written to, at the lowest priority in the queue. + await makeSendable(db, { personEmail: 'jane@acme.com', companyEmail: 'support@acme.com' }); + await db.execute({ + sql: 'UPDATE recommendations SET priority = 1 WHERE id = ?', + args: [SEED.recommendationId], + }); + + // Three colleagues ahead of her, all resolving to an inbox on cooldown. + await inboxAlreadyWrittenTo(db, 'support@acme.com'); + await addHeldColleague(db, 'Held1', 50); + await addHeldColleague(db, 'Held2', 50); + await addHeldColleague(db, 'Held3', 50); + + // A cap smaller than the number of held cards. The old window was + // `cap - sent today` rows, so this read two held cards and stopped. + const stamp = now(); + await db.execute({ + sql: `INSERT INTO workspace_settings (workspace_id, autopilot_daily_cap, created_at, updated_at) + VALUES (?, 2, ?, ?)`, + args: [SEED.workspaceId, stamp, stamp], + }); + + const { sent, mailer } = recordingMailer(); + const result = await runAutopilot( + { db, mailer, holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent.map((message) => message.to)).toEqual(['jane@acme.com']); + expect(result.skipped).toHaveLength(3); + expect(result.skipped.every((skip) => /address|company inbox/i.test(skip.reason))).toBe(true); + }); + + test('the same hold is written down once, not every tick', async () => { + seeded = await seedDatabase('autopilot-hold-once'); + const { db } = seeded; + + await makeSendable(db, { companyEmail: 'support@acme.com' }); + await inboxAlreadyWrittenTo(db, 'support@acme.com'); + + const ledger = new HoldLedger(); + const { mailer } = recordingMailer(); + + const first = await runAutopilot({ db, mailer, holdLedger: ledger }, SEED.workspaceId); + const second = await runAutopilot({ db, mailer, holdLedger: ledger }, SEED.workspaceId); + + // Reported on every run — the log still says what is happening. + expect(first.skipped).toHaveLength(1); + expect(second.skipped).toHaveLength(1); + + // Written to the live feed once. + const events = await queryAll<{ message: string }>( + db, + `SELECT message FROM workflow_events WHERE phase = 'send' AND level = 'warn'`, + ); + expect(events).toHaveLength(1); + + // And the digest can say what is holding the queue. + expect(ledger.summary(SEED.workspaceId)).toEqual([ + { label: 'the address was written to within the cooldown', count: 1 }, + ]); + }); + + test('an address they gave us in an import is their own, not a shared one', async () => { + seeded = await seedDatabase('autopilot-imported-address'); + const { db } = seeded; + + // No published identity, no company inbox — only the imported row. + await makeSendable(db); + await db.execute({ + sql: `INSERT INTO person_emails (id, workspace_id, person_id, address, dedupe_key, source, + verified, created_at) VALUES ('pem_jane', ?, ?, 'jane@home.example', + 'jane@home.example', 'import', 1, ?)`, + args: [SEED.workspaceId, SEED.personId, now()], + }); + + const { sent, mailer } = recordingMailer(); + const result = await runAutopilot( + { db, mailer, holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + expect(sent.map((message) => message.to)).toEqual(['jane@home.example']); + expect(result.sent[0]?.toSharedInbox).toBe(false); + }); +}); + +describe('HoldLedger', () => { + test('a changed number is the same hold; a changed reason is a new one', () => { + const ledger = new HoldLedger(); + + expect(ledger.observe('wsp', 'rec', 'Only 9.85h since this address was last contacted')).toBe( + true, + ); + expect(ledger.observe('wsp', 'rec', 'Only 9.09h since this address was last contacted')).toBe( + false, + ); + expect(ledger.observe('wsp', 'rec', 'no drafted message')).toBe(true); + + ledger.release('wsp', 'rec'); + expect(ledger.observe('wsp', 'rec', 'no drafted message')).toBe(true); + }); + + test('forgets cards a full pass did not see', () => { + const ledger = new HoldLedger(); + ledger.observe('wsp', 'gone', 'no drafted message'); + ledger.observe('wsp', 'kept', 'no drafted message'); + + ledger.retain('wsp', new Set(['kept'])); + + expect(ledger.summary('wsp')).toEqual([{ label: 'no message written yet', count: 1 }]); + }); + + test('describes the engine’s refusals as short reasons', () => { + expect( + describeHold( + 'This prospect shares a company inbox that has already had 2 message(s) this week; the limit is 2.', + ), + ).toBe("the company inbox already had this week's messages"); + expect(describeHold('Weekly limit for this prospect reached (1/1).')).toBe( + 'already written to this week', + ); + expect(describeHold('something the engine has never said')).toBe( + 'something the engine has never said', + ); + }); +}); diff --git a/packages/pipeline/src/autopilot.ts b/packages/pipeline/src/autopilot.ts index e6beaed..5e8f8f3 100644 --- a/packages/pipeline/src/autopilot.ts +++ b/packages/pipeline/src/autopilot.ts @@ -31,7 +31,7 @@ import { INTERNAL_ACTION_KINDS, newId, type ActionKind, type Network } from '@outreachgraph/domain'; import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; -import { evaluatePolicy, isExecutable } from '@outreachgraph/policy'; +import { evaluateAddressLimits, evaluatePolicy, isExecutable } from '@outreachgraph/policy'; import type { Mailer } from '@outreachgraph/email'; import { draftForRecommendation, type TextModel } from '@outreachgraph/ai'; import { mailerForWorkspace } from './email-account'; @@ -85,6 +85,8 @@ export interface AutopilotDeps { */ readonly model?: TextModel; readonly now?: Date; + /** Where holds are remembered between runs. Defaults to the process-wide one. */ + readonly holdLedger?: HoldLedger; } export interface SentOutreach { @@ -141,11 +143,151 @@ interface Candidate { /** Attempts before a recommendation is left alone for a human to look at. */ const MAX_SEND_ATTEMPTS = 3; +/** + * How many pending cards one run will look at. + * + * This used to be `cap - today`, on the theory that a run can send at most + * that many and so need not read more. It was wrong in the one situation that + * matters: a queue where the top of the priority order cannot be sent. In + * production 597 email cards were pending against a cap of 200, and the 194 + * highest-priority ones all resolved to a handful of shared company inboxes + * already inside their weekly limit. Every tick read those 194, held every + * one, and stopped — while 172 cards with a personal address sat just past the + * window and were never looked at. Six messages a day out of a queue that + * could have sent a hundred. + * + * So the window is now bounded by a ceiling that exists only to keep one run + * finite, and the loop stops when the cap is spent rather than when the window + * is. A held card costs an in-memory check, not a slot. + */ +const CANDIDATE_CEILING = 2000; + +/** + * Why a card is currently held, remembered across runs. + * + * Autopilot runs every tick and a held card is held on every one of them, so + * writing the reason each time produced 18,532 identical "held back" rows in + * one day for 120 people — enough that reading the day's events timed out. + * The reason is written when it is new or when it changes, and released when + * the card is sent. Numbers are ignored when comparing, because "Only 9.85h + * since this address was last contacted" and "Only 9.09h" are the same hold. + * + * The ledger also answers "what is holding the queue up" for the digest, + * which is the question a queue that sends six a day most needs to answer. + */ +export class HoldLedger { + private readonly held = new Map>(); + + /** Records a hold. True when it is worth writing down — new, or changed. */ + observe(workspaceId: string, recommendationId: string, reason: string): boolean { + const key = holdKey(reason); + const entries = this.entriesFor(workspaceId); + const previous = entries.get(recommendationId); + entries.set(recommendationId, { reason, key }); + return previous?.key !== key; + } + + /** Forgets a card that was sent, or otherwise stopped being held. */ + release(workspaceId: string, recommendationId: string): void { + this.held.get(workspaceId)?.delete(recommendationId); + } + + /** + * Drops cards a completed run did not see — sent by hand, superseded, + * expired. Only called after a full pass, so a run that stopped at the cap + * does not forget the cards it never reached. + */ + retain(workspaceId: string, seen: ReadonlySet): void { + const entries = this.held.get(workspaceId); + if (!entries) return; + for (const id of [...entries.keys()]) { + if (!seen.has(id)) entries.delete(id); + } + } + + /** The holds in force, grouped by reason, largest group first. */ + summary(workspaceId: string): readonly HeldGroup[] { + const groups = new Map(); + + for (const entry of this.held.get(workspaceId)?.values() ?? []) { + const group = groups.get(entry.key); + if (group) group.count += 1; + else groups.set(entry.key, { label: describeHold(entry.reason), count: 1 }); + } + + return [...groups.values()].sort((a, b) => b.count - a.count); + } + + private entriesFor(workspaceId: string): Map { + let entries = this.held.get(workspaceId); + if (!entries) { + entries = new Map(); + this.held.set(workspaceId, entries); + } + return entries; + } +} + +interface HoldEntry { + readonly reason: string; + readonly key: string; +} + +export interface HeldGroup { + /** The reason as a short noun phrase, the same for one card or a hundred. */ + readonly label: string; + readonly count: number; +} + +/** The process-wide ledger; one container, one replica, one worker loop. */ +export const holdLedger = new HoldLedger(); + +/** What is holding a workspace's queue up right now. */ +export function heldSummary( + workspaceId: string, + ledger: HoldLedger = holdLedger, +): readonly HeldGroup[] { + return ledger.summary(workspaceId); +} + +function holdKey(reason: string): string { + return reason.replace(/\d+(?:\.\d+)?/g, '#'); +} + +/** + * The engine's refusal, reworded to follow a count. + * + * "This prospect shares a company inbox that has already had 2 message(s) + * this week" is right for one card and wrong for a hundred and twenty. Unknown + * reasons are passed through rather than dropped. + */ +export function describeHold(reason: string): string { + if (/shares a company inbox/i.test(reason)) { + return "the company inbox already had this week's messages"; + } + if (/weekly limit for this address/i.test(reason)) { + return "the address already had this week's messages"; + } + if (/the cooldown is/i.test(reason)) return 'the address was written to within the cooldown'; + if (/weekly limit for this prospect/i.test(reason)) return 'already written to this week'; + if (/no address published/i.test(reason)) return 'no address to write to'; + if (/no drafted message/i.test(reason)) return 'no message written yet'; + if (/quality checks/i.test(reason)) return 'the draft failed its quality checks'; + if (/giving up after/i.test(reason)) return 'sending failed repeatedly'; + if (/requires human approval/i.test(reason)) return 'waiting for your approval'; + if (/no mailbox is connected/i.test(reason)) return 'no mailbox connected'; + if (/budget/i.test(reason)) return "over the plan's monthly allowance"; + return reason; +} + /** * Sends everything due for one workspace. * * Ordered by priority so a daily cap spends itself on the best leads rather - * than on whichever rows the planner happened to return first. + * than on whichever rows the planner happened to return first. A personal + * address sorts ahead of a shared inbox at equal priority: the shared one is + * the more likely to be held, and the cap is better spent on a message that + * reaches the person it names. */ export async function runAutopilot( deps: AutopilotDeps, @@ -153,6 +295,7 @@ export async function runAutopilot( ): Promise { const { db } = deps; const at = deps.now ?? new Date(); + const ledger = deps.holdLedger ?? holdLedger; const sent: SentOutreach[] = []; const skipped: SkippedOutreach[] = []; @@ -174,6 +317,11 @@ export async function runAutopilot( return { sent, skipped, failed }; } + // The address is the person's own when one is known — published on a page + // they control, or given to us in an import they consented to — and the + // employer's shared inbox otherwise. Imported addresses live in + // `person_emails`, and until now this query only read `social_identities`, + // so sixteen thousand consented mailboxes were invisible to the sender. const candidates = await queryAll( db, `SELECT r.id AS recommendation_id, r.campaign_id, r.person_id, r.action, r.network, @@ -183,9 +331,15 @@ export async function runAutopilot( w.min_outreach_confidence, d.id AS draft_id, d.subject, d.body, d.checks_json, co.name AS company_name, co.contact_email AS company_contact_email, - (SELECT si.handle FROM social_identities si - WHERE si.person_id = p.id AND si.network = 'email' - ORDER BY si.confidence DESC LIMIT 1) AS person_email, + COALESCE( + (SELECT si.handle FROM social_identities si + WHERE si.person_id = p.id AND si.network = 'email' + AND si.handle IS NOT NULL AND trim(si.handle) <> '' + ORDER BY si.confidence DESC LIMIT 1), + (SELECT pe.address FROM person_emails pe + WHERE pe.person_id = p.id AND pe.workspace_id = r.workspace_id + ORDER BY pe.created_at LIMIT 1) + ) AS person_email, (SELECT COUNT(*) FROM actions a WHERE a.recommendation_id = r.id AND a.status = 'failed') AS failed_attempts FROM recommendations r @@ -201,20 +355,40 @@ export async function runAutopilot( AND r.action = 'send_email' AND r.network = 'email' AND p.status = 'active' - ORDER BY r.priority DESC, r.created_at ASC + ORDER BY r.priority DESC, (person_email IS NULL) ASC, r.created_at ASC LIMIT ?`, - [workspaceId, Math.max(cap - today, 0)], + [workspaceId, CANDIDATE_CEILING], ); + // What each mailbox has already had, read once per address per run. A + // hundred colleagues behind one `support@` cost one query and a hundred + // comparisons, not a hundred queries — and after a send the entry is + // updated in place so the next colleague sees the message that just left. + const addressUsage = new Map(); + + // Read once and refreshed after each send rather than once per candidate: a + // workspace can cross its monthly allowance partway through a sweep, and only + // a send can move it. + let budgetState = await budgetStatus(db, workspaceId, at); + + const seen = new Set(); + let completed = true; + for (const row of candidates) { - if (today >= cap) break; + if (today >= cap) { + completed = false; + break; + } + seen.add(row.recommendation_id); // Skips are reported, not swallowed. // // "No address published for this person" and "still requires human // approval" are the two reasons a campaign sits at a stage looking broken, // and neither is an error anywhere else in the system — so if they are not - // surfaced here they are not surfaced at all. + // surfaced here they are not surfaced at all. The event is written when + // the reason is new; the same hold on the next tick is remembered, not + // repeated. const note = async (reason: string): Promise => { skipped.push({ recommendationId: row.recommendation_id, @@ -222,6 +396,8 @@ export async function runAutopilot( reason, }); + if (!ledger.observe(workspaceId, row.recommendation_id, reason)) return; + await emitEvent(db, { workspaceId, campaignId: row.campaign_id, @@ -246,6 +422,111 @@ export async function runAutopilot( continue; } + const recipient = pickEmailRecipient(row); + if (!recipient) { + await note('no address published for this person or their company'); + continue; + } + + const budget = safeJson(row.budget_json); + const cooldown = + typeof budget.minHoursBetweenActions === 'number' + ? { minHoursBetweenActions: budget.minHoursBetweenActions } + : {}; + + // ------------------------------------------------------ address gates + // + // Counted against the mailbox as well as the person, and checked first, + // before anything that costs a query or a model call. This is the gate + // that holds most of a real queue: a prospect with no personal address + // falls back to their employer's shared inbox, so N colleagues are N + // separate people, each comfortably inside its own weekly limit, while one + // `support@` receives N messages. The engine grew these gates in #34 and + // only the human approval route fed them until #48; they are the same + // arithmetic the queue's badge uses, so what is shown as held is held. + const address = recipient.address.trim().toLowerCase(); + let usage = addressUsage.get(address); + if (!usage) { + usage = await addressCounts(db, workspaceId, recipient.address, at); + addressUsage.set(address, usage); + } + + const breaches = evaluateAddressLimits({ + actionsThisWeek: usage.thisWeek, + maxPerWeek: numberOr(budget.maxActionsPerAddressPerWeek, 1), + shared: recipient.shared, + ...(usage.hoursSinceLast !== undefined ? { hoursSinceLast: usage.hoursSinceLast } : {}), + ...(typeof budget.minHoursBetweenActions === 'number' + ? { cooldownHours: budget.minHoursBetweenActions } + : {}), + }); + + // The engine reports the last breach when several fire; so does this. + const breach = breaches[breaches.length - 1]; + if (breach) { + await note(breach.reason); + continue; + } + + // ------------------------------------------------------------- policy + // + // Re-evaluated from live rows, never from the stored snapshot. + const counts = await actionCounts(db, workspaceId, row.person_id, at); + + const decision = evaluatePolicy({ + network: row.network as Network, + action: row.action as ActionKind, + approvalMode: row.approval_mode as 'trusted_automation', + hasConnectedAccount: sender !== undefined, + personSuppressed: row.person_status === 'suppressed' || row.outreach_eligible === 0, + personBelievedMinor: row.believed_minor === 1, + personDeleted: row.person_status === 'deleted', + identityConfidence: row.identity_confidence, + minIdentityConfidence: row.min_outreach_confidence, + actionsToday: today, + maxActionsPerDay: Math.min(numberOr(budget.maxActionsPerDay, 50), cap), + actionsToThisProspectThisWeek: counts.thisProspect, + maxActionsPerProspectPerWeek: numberOr(budget.maxActionsPerProspectPerWeek, 1), + // The cooldown the campaign configured, not only the engine default. + ...cooldown, + ...(counts.hoursSinceLast !== undefined + ? { hoursSinceLastActionToProspect: counts.hoursSinceLast } + : {}), + // Already known to pass; supplied so the engine's answer is complete. + actionsToThisAddressThisWeek: usage.thisWeek, + maxActionsPerAddressPerWeek: numberOr(budget.maxActionsPerAddressPerWeek, 1), + addressShared: recipient.shared, + ...(usage.hoursSinceLast !== undefined + ? { hoursSinceLastActionToAddress: usage.hoursSinceLast } + : {}), + budgetExhausted: budgetState.exhausted, + }); + + // `approved: false` is the whole point. Autopilot holds no approval, so + // only a decision of plain `allow` gets through — `allow_with_approval` + // means a human still has to look at it, and reaching here with that would + // mean the capability matrix no longer marks email customer-managed. + // Sending anyway would be exactly what the approval default prevents. + if (!isExecutable(decision.decision, false)) { + await note( + decision.decision === 'allow_with_approval' + ? 'this action still requires human approval' + : decision.reason, + ); + continue; + } + + if (!sender) { + await note('no mailbox is connected, so nothing can be sent'); + continue; + } + + // -------------------------------------------------------------- draft + // + // Only now, once the card is known to be sendable. Drafting is a model + // call, and writing a message for a card the address gate was about to + // hold anyway paid for a hundred drafts a tick that went nowhere. + // // A recommendation with no draft has nothing to send — but "the composer // declined to write one" and "nobody ever tried" are different states, and // until now they produced the same warning on every tick forever. Drafting @@ -310,93 +591,6 @@ export async function runAutopilot( continue; } - const recipient = pickEmailRecipient(row); - if (!recipient) { - await note('no address published for this person or their company'); - continue; - } - - // ------------------------------------------------------------- policy - // - // Re-evaluated from live rows, never from the stored snapshot. - const counts = await actionCounts(db, workspaceId, row.person_id, at); - - // Counted against the mailbox as well as the person. - // - // Both limits are needed and neither substitutes for the other. The - // per-person limit answers "how often do we contact this human"; this one - // answers "how much mail does this mailbox get", and a prospect with no - // personal address falls back to their employer's shared inbox — so N - // colleagues are N separate people, each comfortably inside its own weekly - // limit, while one `support@` receives N messages. - // - // The policy engine grew these gates in #34, but only the human approval - // route in `app.ts` ever filled them in. They are optional inputs, so - // omitting them does not fail loudly — it silently disables them, and this - // is the unattended path that sends at volume. In production the manual - // route was protected and autopilot was not, which is how an address that - // had already been written to that afternoon was written to again hours - // after the fix shipped. - const addressUsage = await addressCounts(db, workspaceId, recipient.address, at); - const budget = safeJson(row.budget_json); - - // Read inside the loop rather than once per run: a workspace can cross its - // monthly allowance partway through a sweep, and a snapshot taken before - // the first send would let the rest of the batch through on a stale count. - const budgetState = await budgetStatus(db, workspaceId, at); - - const decision = evaluatePolicy({ - network: row.network as Network, - action: row.action as ActionKind, - approvalMode: row.approval_mode as 'trusted_automation', - hasConnectedAccount: sender !== undefined, - personSuppressed: row.person_status === 'suppressed' || row.outreach_eligible === 0, - personBelievedMinor: row.believed_minor === 1, - personDeleted: row.person_status === 'deleted', - identityConfidence: row.identity_confidence, - minIdentityConfidence: row.min_outreach_confidence, - actionsToday: today, - maxActionsPerDay: Math.min(numberOr(budget.maxActionsPerDay, 50), cap), - actionsToThisProspectThisWeek: counts.thisProspect, - maxActionsPerProspectPerWeek: numberOr(budget.maxActionsPerProspectPerWeek, 1), - // The cooldown the campaign configured, not only the engine default. - // `evaluateAddressLimits` already honoured this for the queue's badge, - // so leaving it out here made the preview and the refusal disagree about - // the same card. - ...(typeof budget.minHoursBetweenActions === 'number' - ? { minHoursBetweenActions: budget.minHoursBetweenActions } - : {}), - ...(counts.hoursSinceLast !== undefined - ? { hoursSinceLastActionToProspect: counts.hoursSinceLast } - : {}), - actionsToThisAddressThisWeek: addressUsage.thisWeek, - maxActionsPerAddressPerWeek: numberOr(budget.maxActionsPerAddressPerWeek, 1), - addressShared: recipient.shared, - ...(addressUsage.hoursSinceLast !== undefined - ? { hoursSinceLastActionToAddress: addressUsage.hoursSinceLast } - : {}), - budgetExhausted: budgetState.exhausted, - }); - - // `approved: false` is the whole point. Autopilot holds no approval, so - // only a decision of plain `allow` gets through — `allow_with_approval` - // means a human still has to look at it, and reaching here with that would - // mean the capability matrix no longer marks email customer-managed. - // Sending anyway would be exactly what the approval default prevents. - if (!isExecutable(decision.decision, false)) { - await note( - decision.decision === 'allow_with_approval' - ? 'this action still requires human approval' - : decision.reason, - ); - continue; - } - - if (!sender) { - await note('no mailbox is connected, so nothing can be sent'); - continue; - } - // --------------------------------------------------------------- send const actionId = newId('action'); const stamp = now(); @@ -485,6 +679,12 @@ export async function runAutopilot( }); today += 1; + ledger.release(workspaceId, row.recommendation_id); + + // The mailbox just received one; colleagues behind it on this run must + // see that without re-reading the table. + addressUsage.set(address, { thisWeek: usage.thisWeek + 1, hoursSinceLast: 0 }); + budgetState = await budgetStatus(db, workspaceId, at); } catch (error) { const message = error instanceof Error ? error.message : String(error); failed += 1; @@ -520,9 +720,16 @@ export async function runAutopilot( } } + if (completed) ledger.retain(workspaceId, seen); + return { sent, skipped, failed }; } +interface AddressUsage { + readonly thisWeek: number; + readonly hoursSinceLast?: number; +} + /** True when any recorded quality gate failed. Unparseable checks fail closed. */ function hasFailingCheck(checksJson: string | null): boolean { if (!checksJson) return false; @@ -612,7 +819,7 @@ async function addressCounts( workspaceId: string, address: string, at: Date, -): Promise<{ thisWeek: number; hoursSinceLast?: number }> { +): Promise { const weekAgo = new Date(at.getTime() - 7 * 24 * 3_600_000).toISOString(); const row = await queryOne<{ n: number; last_at: string | null }>( diff --git a/packages/pipeline/src/enrich-contact.ts b/packages/pipeline/src/enrich-contact.ts index c53899a..7c06025 100644 --- a/packages/pipeline/src/enrich-contact.ts +++ b/packages/pipeline/src/enrich-contact.ts @@ -252,6 +252,18 @@ export async function enrichContact( const filledName = await fillGaps(db, input.personId, profile); + // The picture they chose for exactly this purpose. Gravatar's whole point is + // an avatar looked up by address, and it was being fetched and thrown away. + // Never overwrites one already held; a later, better source wins by being + // first, not by being later. + if (profile.avatarUrl) { + await db.execute({ + sql: `UPDATE people SET avatar_url = ?, avatar_source = 'gravatar', updated_at = ? + WHERE id = ? AND avatar_url IS NULL`, + args: [sizedGravatar(profile.avatarUrl), now(), input.personId], + }); + } + return { personId: input.personId, found: true, @@ -261,6 +273,11 @@ export async function enrichContact( }; } +/** Gravatar serves 80px unless asked; the digest and the queue want more. */ +function sizedGravatar(url: string): string { + return url.includes('?') ? url : `${url}?s=256`; +} + /** * Stores one published account, once. * diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts index 8deba1e..31630ca 100644 --- a/packages/pipeline/src/index.ts +++ b/packages/pipeline/src/index.ts @@ -16,11 +16,23 @@ export { export { runDiscoveryJob, type DiscoveryJobDeps, type DiscoveryJobResult } from './discovery'; export { runAutopilot, + HoldLedger, + holdLedger, + heldSummary, + describeHold, type AutopilotDeps, type AutopilotResult, + type HeldGroup, type SentOutreach, type SkippedOutreach, } from './autopilot'; +export { + sweepProfilePhotos, + workspacesAwaitingPhotos, + type PhotoSweepDeps, + type PhotoSweepResult, +} from './photos'; +export { reseedIdleCampaigns, type ReseedResult } from './reseed'; export { runListening, listeningCampaigns, diff --git a/packages/pipeline/src/notify.test.ts b/packages/pipeline/src/notify.test.ts index 7b365bb..4328970 100644 --- a/packages/pipeline/src/notify.test.ts +++ b/packages/pipeline/src/notify.test.ts @@ -3,6 +3,8 @@ import { now, queryAll, type Client } from '@outreachgraph/db'; import type { Mailer, Message, SendResult } from '@outreachgraph/email'; import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; import { notifyAddress, loadNotifySettings, sendDailyDigest, sendLeadAlerts } from './notify'; +import { HoldLedger } from './autopilot'; +import { recordDiscovered } from './stages'; let seeded: SeededDatabase | undefined; @@ -229,3 +231,80 @@ describe('sendDailyDigest', () => { expect(sent).toHaveLength(0); }); }); + +describe('what the digest says about the queue', () => { + test('splits the queue into what autopilot will send and what needs a human', async () => { + seeded = await seedDatabase('digest-queue'); + const { db } = seeded; + await setDigestHour(db, 9); + + // One email card autopilot owns, on a campaign that opted in. + await db.execute({ + sql: `UPDATE campaigns SET approval_mode = 'trusted_automation' WHERE id = ?`, + args: [SEED.campaignId], + }); + await db.execute({ + sql: `UPDATE recommendations SET action = 'send_email', network = 'email' WHERE id = ?`, + args: [SEED.recommendationId], + }); + + // One LinkedIn card only a person can act on. + await db.execute({ + sql: `INSERT INTO recommendations (id, workspace_id, campaign_id, person_id, action, network, + priority, reason, policy_status, policy_version, status, created_at) + VALUES ('rec_li', ?, ?, ?, 'reply', 'linkedin', 10, 'they posted', + 'manual_only', '2026-01-01', 'pending', ?)`, + args: [SEED.workspaceId, SEED.campaignId, SEED.personId, now()], + }); + + // Autopilot remembers holding the email card. + const ledger = new HoldLedger(); + ledger.observe( + SEED.workspaceId, + SEED.recommendationId, + 'This prospect shares a company inbox that has already had 2 message(s) this week; the limit is 2.', + ); + + // Jane has a picture and surfaced today. + await db.execute({ + sql: `UPDATE people SET avatar_url = 'https://pics.example/jane.jpg' WHERE id = ?`, + args: [SEED.personId], + }); + await recordDiscovered(db, { + workspaceId: SEED.workspaceId, + campaignId: SEED.campaignId, + personId: SEED.personId, + }); + + const { sent, mailer } = recordingMailer(); + await sendDailyDigest( + { db, mailer, appUrl: APP_URL, now: at(12), holdLedger: ledger }, + SEED.workspaceId, + ); + + const mail = sent[0]; + expect(mail?.text).toContain('Autopilot queue: 1 (1 held by a limit)'); + expect(mail?.text).toContain('Needs you: 1 (LinkedIn 1)'); + expect(mail?.text).not.toContain('Awaiting approval'); + expect(mail?.text).toContain( + "1 held in the autopilot queue: the company inbox already had this week's messages.", + ); + expect(mail?.html).toContain(' { + seeded = await seedDatabase('digest-plain'); + const { db } = seeded; + await setDigestHour(db, 9); + + const { sent, mailer } = recordingMailer(); + await sendDailyDigest( + { db, mailer, appUrl: APP_URL, now: at(12), holdLedger: new HoldLedger() }, + SEED.workspaceId, + ); + + // The seed's one pending card is a reply on a manual network. + expect(sent[0]?.text).toContain('Needs you: 1'); + expect(sent[0]?.text).not.toContain('Autopilot queue'); + }); +}); diff --git a/packages/pipeline/src/notify.ts b/packages/pipeline/src/notify.ts index 47b25e7..b893f70 100644 --- a/packages/pipeline/src/notify.ts +++ b/packages/pipeline/src/notify.ts @@ -19,8 +19,9 @@ * a crash mid-loop cannot produce a second copy. */ -import { newId } from '@outreachgraph/domain'; +import { INTERNAL_ACTION_KINDS, newId } from '@outreachgraph/domain'; import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; +import { heldSummary, type HoldLedger } from './autopilot'; import { dailyDigestEmail, leadAlertEmail, @@ -36,6 +37,8 @@ export interface NotifyDeps { /** Absolute base for links in the mail. */ readonly appUrl: string; readonly now?: Date; + /** Where autopilot remembers what it is holding. Defaults to the process-wide one. */ + readonly holdLedger?: HoldLedger; } export interface NotifySettings { @@ -291,7 +294,9 @@ export async function sendDailyDigest(deps: NotifyDeps, workspaceId: string): Pr const since = `${today}T00:00:00.000Z`; - const [crawled, found, sent, awaiting, leads] = await Promise.all([ + const internal = INTERNAL_ACTION_KINDS.map(() => '?').join(', '); + + const [crawled, found, sent, awaiting, queue, manual, leads] = await Promise.all([ countOne( deps.db, `SELECT COUNT(*) AS n FROM jobs @@ -317,16 +322,42 @@ export async function sendDailyDigest(deps: NotifyDeps, workspaceId: string): Pr WHERE workspace_id = ? AND status = 'pending'`, [workspaceId], ), + // What autopilot will send on its own: email cards in a campaign that + // opted in. The only network the matrix lets a machine act on. + countOne( + deps.db, + `SELECT COUNT(*) AS n FROM recommendations r + JOIN campaigns c ON c.id = r.campaign_id + WHERE r.workspace_id = ? AND r.status = 'pending' + AND r.action = 'send_email' AND r.network = 'email' + AND c.approval_mode = 'trusted_automation' AND c.status != 'archived'`, + [workspaceId], + ), + // What only a human can do: everything outbound that is not the above. + // Research cards are internal and clear themselves, so they are neither. + queryAll<{ network: string; n: number }>( + deps.db, + `SELECT r.network, COUNT(*) AS n FROM recommendations r + JOIN campaigns c ON c.id = r.campaign_id + WHERE r.workspace_id = ? AND r.status = 'pending' + AND r.action NOT IN (${internal}) + AND NOT (r.action = 'send_email' AND r.network = 'email' + AND c.approval_mode = 'trusted_automation' AND c.status != 'archived') + GROUP BY r.network + ORDER BY n DESC`, + [workspaceId, ...INTERNAL_ACTION_KINDS], + ), queryAll<{ person_id: string; display_name: string; current_title: string | null; company_name: string | null; opportunity: number | null; + avatar_url: string | null; sent_today: number; }>( deps.db, - `SELECT p.id AS person_id, p.display_name, p.current_title, + `SELECT p.id AS person_id, p.display_name, p.current_title, p.avatar_url, co.name AS company_name, s.opportunity, (SELECT COUNT(*) FROM actions a WHERE a.person_id = p.id AND a.workspace_id = e.workspace_id @@ -350,16 +381,34 @@ export async function sendDailyDigest(deps: NotifyDeps, workspaceId: string): Pr ...(lead.current_title ? { title: lead.current_title } : {}), ...(lead.company_name ? { companyName: lead.company_name } : {}), ...(lead.opportunity !== null ? { opportunity: Math.round(lead.opportunity) } : {}), + ...(lead.avatar_url ? { avatarUrl: lead.avatar_url } : {}), ...(lead.sent_today > 0 ? { sentTo: 'sent' } : {}), })); + // Why the queue is not moving, from the sweep's own memory of what it held + // on its last pass. The question a digest that says "6 sent" has to answer + // is what happened to the other five hundred, and until now it did not. + const held = heldSummary(workspaceId, deps.holdLedger); + const heldCount = held.reduce((sum, group) => sum + group.count, 0); + + const notes = held.map((group) => `${group.count} held in the autopilot queue: ${group.label}.`); + + const byNetwork: Record = {}; + for (const row of manual) byNetwork[row.network] = Number(row.n); + const manualTotal = Object.values(byNetwork).reduce((sum, n) => sum + n, 0); + const digest: DailyDigest = { date: today, sitesCrawled: crawled, peopleFound: found, messagesSent: sent, awaitingApproval: awaiting, + ...(queue > 0 || heldCount > 0 + ? { autopilotQueue: { total: queue, held: Math.min(heldCount, queue) } } + : {}), + ...(manualTotal > 0 ? { needsYou: { total: manualTotal, byNetwork } } : {}), leads: digestLeads, + ...(notes.length > 0 ? { notes } : {}), }; try { diff --git a/packages/pipeline/src/photos.test.ts b/packages/pipeline/src/photos.test.ts new file mode 100644 index 0000000..05c9b87 --- /dev/null +++ b/packages/pipeline/src/photos.test.ts @@ -0,0 +1,126 @@ +/** + * Finding a face for a lead. + * + * What matters here is the bookkeeping around the lookup, not the lookup: a + * miss must be remembered so it is never paid for twice, a hit must record the + * profile it came from as research, and the daily ceiling must hold. + */ + +import { afterEach, describe, expect, test } from 'bun:test'; +import { queryOne } from '@outreachgraph/db'; +import type { ProfilePhoto, ProfilePhotoFinder, ProfilePhotoQuery } from '@outreachgraph/providers'; +import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; +import { sweepProfilePhotos, workspacesAwaitingPhotos } from './photos'; + +let seeded: SeededDatabase | undefined; + +afterEach(() => { + seeded?.cleanup(); + seeded = undefined; +}); + +function finder(answer: ProfilePhoto | undefined): { + asked: ProfilePhotoQuery[]; + finder: ProfilePhotoFinder; +} { + const asked: ProfilePhotoQuery[] = []; + return { + asked, + finder: { + findProfilePhoto: async (query) => { + asked.push(query); + return answer; + }, + }, + }; +} + +const LINKEDIN_HIT: ProfilePhoto = { + photoUrl: 'https://media.licdn.com/dms/image/jane.jpg', + pageUrl: 'https://www.linkedin.com/in/jane-smith-123/', + source: 'linkedin', +}; + +describe('sweepProfilePhotos', () => { + test('asks with everything known and records the picture and the profile', async () => { + seeded = await seedDatabase('photos-hit'); + const { db } = seeded; + const { asked, finder: found } = finder(LINKEDIN_HIT); + + const result = await sweepProfilePhotos( + { db, finder: found }, + { workspaceId: SEED.workspaceId }, + ); + + expect(result).toEqual({ looked: 1, found: 1, profiles: 1, remainingToday: 299 }); + expect(asked[0]).toEqual({ + name: 'Jane Smith', + title: 'VP Engineering', + company: 'Acme', + companyDomain: 'acme.com', + }); + + const person = await queryOne<{ avatar_url: string; avatar_source: string }>( + db, + 'SELECT avatar_url, avatar_source FROM people WHERE id = ?', + [SEED.personId], + ); + expect(person?.avatar_url).toBe(LINKEDIN_HIT.photoUrl); + expect(person?.avatar_source).toBe('search'); + + // Research, at a confidence no machine can act on. + const identity = await queryOne<{ handle: string; profile_url: string; confidence: number }>( + db, + `SELECT handle, profile_url, confidence FROM social_identities + WHERE person_id = ? AND network = 'linkedin'`, + [SEED.personId], + ); + expect(identity?.handle).toBe('jane-smith-123'); + expect(identity?.profile_url).toBe('https://www.linkedin.com/in/jane-smith-123/'); + expect(identity?.confidence).toBeLessThan(0.85); + + // Done: nothing left to look up. + expect(await workspacesAwaitingPhotos(db)).toEqual([]); + }); + + test('a miss is remembered and never paid for again', async () => { + seeded = await seedDatabase('photos-miss'); + const { db } = seeded; + const { asked, finder: missed } = finder(undefined); + + const first = await sweepProfilePhotos( + { db, finder: missed }, + { workspaceId: SEED.workspaceId }, + ); + const second = await sweepProfilePhotos( + { db, finder: missed }, + { workspaceId: SEED.workspaceId }, + ); + + expect(first.looked).toBe(1); + expect(second.looked).toBe(0); + expect(asked).toHaveLength(1); + + const person = await queryOne<{ avatar_url: string | null; photo_looked_up_at: string | null }>( + db, + 'SELECT avatar_url, photo_looked_up_at FROM people WHERE id = ?', + [SEED.personId], + ); + expect(person?.avatar_url).toBeNull(); + expect(person?.photo_looked_up_at).not.toBeNull(); + }); + + test('the daily ceiling stops the sweep before it asks', async () => { + seeded = await seedDatabase('photos-cap'); + const { db } = seeded; + const { asked, finder: found } = finder(LINKEDIN_HIT); + + const result = await sweepProfilePhotos( + { db, finder: found, dailyCap: 0 }, + { workspaceId: SEED.workspaceId }, + ); + + expect(result.looked).toBe(0); + expect(asked).toHaveLength(0); + }); +}); diff --git a/packages/pipeline/src/photos.ts b/packages/pipeline/src/photos.ts new file mode 100644 index 0000000..1d38665 --- /dev/null +++ b/packages/pipeline/src/photos.ts @@ -0,0 +1,229 @@ +/** + * A face for each lead. + * + * The digest and the queue list people by name and title, and a name is a + * poor handle on a stranger: "Mark Ramsey — Global Senior Pastor at Citipointe + * Church" reads as a line item, the same line with a photograph reads as a + * person. Gravatar supplies one for about a person in a hundred; for the rest + * a search finds the picture on their LinkedIn profile or their company's own + * team page, and only those — see `@outreachgraph/providers` for why the page + * matters more than the picture. + * + * Bounded twice, because every lookup costs money and most of them miss: + * + * - **Per tick.** A small batch, so a workspace that just enrolled a + * thousand people does not spend a thousand credits in one minute. + * - **Per day.** A ceiling on lookups per workspace per day, counted from + * the `photo_looked_up_at` stamps rather than from memory so a restart + * cannot reset it. + * + * And scoped to people the workspace is actually working — members of an + * active campaign — rather than everyone ever imported. A picture of someone + * nobody will write to is a credit spent on nothing. + */ + +import { newId } from '@outreachgraph/domain'; +import type { ProfilePhotoFinder } from '@outreachgraph/providers'; +import { isLinkedInProfile } from '@outreachgraph/providers'; +import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; + +/** Lookups per workspace per tick. */ +const SWEEP_SIZE = 20; +/** Lookups per workspace per day, absent a configured ceiling. */ +const DEFAULT_DAILY_CAP = 300; + +/** + * How much to believe a LinkedIn URL found by search. + * + * Below the outreach floor on purpose. The result carried the person's name + * and the company we already had for them, which is enough to show a human + * the profile as research and nowhere near enough to act on it. Confirmation + * comes from the human, in LinkedIn's own interface. + */ +const SEARCH_IDENTITY_CONFIDENCE = 0.6; + +export interface PhotoSweepDeps { + readonly db: Client; + readonly finder: ProfilePhotoFinder; + readonly limit?: number; + readonly dailyCap?: number; + readonly now?: Date; +} + +export interface PhotoSweepResult { + readonly looked: number; + readonly found: number; + /** LinkedIn profiles recorded as research identities along the way. */ + readonly profiles: number; + /** Lookups left under today's ceiling after this run. */ + readonly remainingToday: number; +} + +interface Subject { + readonly person_id: string; + readonly display_name: string; + readonly current_title: string | null; + readonly company_name: string | null; + readonly company_domain: string | null; +} + +/** + * Looks up the next batch of people in this workspace's campaigns who have no + * picture and have never been looked up. + * + * People with a message waiting to go out come first: they are the ones about + * to appear in the digest as "written to", and the ones a reviewer is about to + * open in the queue. + */ +export async function sweepProfilePhotos( + deps: PhotoSweepDeps, + input: { readonly workspaceId: string }, +): Promise { + const at = deps.now ?? new Date(); + const dailyCap = deps.dailyCap ?? DEFAULT_DAILY_CAP; + const dayStart = `${at.toISOString().slice(0, 10)}T00:00:00.000Z`; + + const spent = await queryOne<{ n: number }>( + deps.db, + `SELECT count(*) AS n FROM people p + WHERE p.photo_looked_up_at >= ? + AND EXISTS (SELECT 1 FROM campaign_people cp + WHERE cp.person_id = p.id AND cp.workspace_id = ?)`, + [dayStart, input.workspaceId], + ); + + const room = Math.max(dailyCap - Number(spent?.n ?? 0), 0); + const limit = Math.min(deps.limit ?? SWEEP_SIZE, room); + + if (limit === 0) return { looked: 0, found: 0, profiles: 0, remainingToday: 0 }; + + const subjects = await queryAll( + deps.db, + `SELECT p.id AS person_id, p.display_name, p.current_title, + co.name AS company_name, co.domain AS company_domain + FROM people p + JOIN campaign_people cp ON cp.person_id = p.id + JOIN campaigns c ON c.id = cp.campaign_id + LEFT JOIN companies co ON co.id = p.current_company_id + WHERE cp.workspace_id = ? AND c.status IN ('active', 'running') + AND p.status = 'active' + AND p.avatar_url IS NULL AND p.photo_looked_up_at IS NULL + GROUP BY p.id + ORDER BY EXISTS (SELECT 1 FROM recommendations r + WHERE r.person_id = p.id AND r.status = 'pending' + AND r.action NOT IN ('refresh_research', 'observe', 'wait')) DESC, + cp.updated_at DESC + LIMIT ?`, + [input.workspaceId, limit], + ); + + let found = 0; + let profiles = 0; + + for (const subject of subjects) { + const photo = await deps.finder + .findProfilePhoto({ + name: subject.display_name, + title: subject.current_title ?? undefined, + company: subject.company_name ?? undefined, + companyDomain: subject.company_domain ?? undefined, + }) + .catch(() => undefined); + + // A miss is stamped so it is not retried forever; only a hit fills the URL. + if (!photo) { + await deps.db.execute({ + sql: 'UPDATE people SET photo_looked_up_at = ? WHERE id = ?', + args: [now(), subject.person_id], + }); + continue; + } + + found += 1; + await deps.db.execute({ + sql: `UPDATE people SET avatar_url = ?, avatar_source = ?, photo_looked_up_at = ?, + updated_at = ? + WHERE id = ? AND avatar_url IS NULL`, + args: [ + photo.photoUrl, + photo.source === 'linkedin' ? 'search' : 'site', + now(), + now(), + subject.person_id, + ], + }); + + // The profile the picture came from is the more useful half of the + // answer: it is the page the human will open to act. Recorded as research, + // once, and never at a confidence that could let a machine use it. + if ( + photo.source === 'linkedin' && + (await recordProfile(deps.db, subject.person_id, photo.pageUrl)) + ) { + profiles += 1; + } + } + + return { + looked: subjects.length, + found, + profiles, + remainingToday: Math.max(room - subjects.length, 0), + }; +} + +/** Workspaces with campaign members still waiting for a picture. */ +export async function workspacesAwaitingPhotos(db: Client): Promise { + const rows = await queryAll<{ workspace_id: string }>( + db, + `SELECT DISTINCT cp.workspace_id + FROM campaign_people cp + JOIN campaigns c ON c.id = cp.campaign_id + JOIN people p ON p.id = cp.person_id + WHERE c.status IN ('active', 'running') AND p.status = 'active' + AND p.avatar_url IS NULL AND p.photo_looked_up_at IS NULL`, + ); + + return rows.map((row) => row.workspace_id); +} + +async function recordProfile(db: Client, personId: string, pageUrl: string): Promise { + let host: string; + let path: string; + try { + const url = new URL(pageUrl); + host = url.hostname.toLowerCase(); + path = url.pathname; + } catch { + return false; + } + if (!isLinkedInProfile(host, path)) return false; + + const existing = await queryOne<{ id: string }>( + db, + `SELECT id FROM social_identities WHERE person_id = ? AND network = 'linkedin' LIMIT 1`, + [personId], + ); + if (existing) return false; + + const handle = path.split('/').filter(Boolean)[1] ?? null; + const stamp = now(); + + await db.execute({ + sql: `INSERT INTO social_identities (id, person_id, network, handle, profile_url, confidence, + source_type, verified_by, first_seen_at, last_verified_at) + VALUES (?, ?, 'linkedin', ?, ?, ?, 'public_web', ?, ?, ?)`, + args: [ + newId('socialIdentity'), + personId, + handle, + `https://${host}${path}`, + SEARCH_IDENTITY_CONFIDENCE, + JSON.stringify(['search']), + stamp, + stamp, + ], + }); + + return true; +} diff --git a/packages/pipeline/src/pipeline.ts b/packages/pipeline/src/pipeline.ts index 0afeb74..bdb3fe7 100644 --- a/packages/pipeline/src/pipeline.ts +++ b/packages/pipeline/src/pipeline.ts @@ -926,6 +926,23 @@ async function createRecommendation( if (inbox?.contact_email) reachable.push('email'); } + // An address they gave us themselves counts before either of the above. + // + // Imported contacts keep their mailbox in `person_emails`, not in + // `social_identities`, and this list was built from the latter alone — so a + // consented, imported person with nothing else known about them was + // "unreachable" and dead-ended at research. Sixteen thousand of them, in + // production, enrolled in nothing and written to never. + if (!reachable.includes('email')) { + const imported = await queryOne<{ id: string }>( + db, + `SELECT id FROM person_emails WHERE person_id = ? AND workspace_id = ? LIMIT 1`, + [personId, workspaceId], + ); + + if (imported) reachable.push('email'); + } + const score = await queryOne<{ opportunity: number }>( db, 'SELECT opportunity FROM scores WHERE campaign_id = ? AND person_id = ?', diff --git a/packages/pipeline/src/reseed.test.ts b/packages/pipeline/src/reseed.test.ts new file mode 100644 index 0000000..52cdb72 --- /dev/null +++ b/packages/pipeline/src/reseed.test.ts @@ -0,0 +1,113 @@ +/** + * Asking a campaign's seed again. + * + * The cases are about restraint: only an idle campaign, only after the + * interval, only once per interval — and then the right kind of job for the + * kind of seed. + */ + +import { afterEach, describe, expect, test } from 'bun:test'; +import { now, queryAll, queryOne, type Client } from '@outreachgraph/db'; +import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; +import { reseedIdleCampaigns } from './reseed'; + +let seeded: SeededDatabase | undefined; + +afterEach(() => { + seeded?.cleanup(); + seeded = undefined; +}); + +const TEN_DAYS_AGO = new Date(Date.now() - 10 * 86_400_000).toISOString(); + +async function seededCampaign( + db: Client, + options: { kind: 'url' | 'keyword'; value: string; startedAt?: string }, +): Promise { + await db.execute({ + sql: `UPDATE campaigns SET status = 'active', seed_kind = ?, seed_value = ?, started_at = ? + WHERE id = ?`, + args: [options.kind, options.value, options.startedAt ?? TEN_DAYS_AGO, SEED.campaignId], + }); +} + +async function jobs(db: Client): Promise<{ kind: string; payload_json: string; status: string }[]> { + return queryAll(db, 'SELECT kind, payload_json, status FROM jobs WHERE workspace_id = ?', [ + SEED.workspaceId, + ]); +} + +describe('reseedIdleCampaigns', () => { + test('reads a URL seed again after the interval, once', async () => { + seeded = await seedDatabase('reseed-url'); + const { db } = seeded; + await seededCampaign(db, { kind: 'url', value: 'acme.com/team' }); + + const first = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(first).toEqual({ considered: 1, queued: 1 }); + + const queued = await jobs(db); + expect(queued).toHaveLength(1); + expect(queued[0]?.kind).toBe('crawl_site'); + expect(JSON.parse(queued[0]?.payload_json ?? '{}')).toEqual({ + url: 'https://acme.com/team', + campaignId: SEED.campaignId, + }); + + const campaign = await queryOne<{ reseeded_at: string | null }>( + db, + 'SELECT reseeded_at FROM campaigns WHERE id = ?', + [SEED.campaignId], + ); + expect(campaign?.reseeded_at).not.toBeNull(); + + // The stamp holds it for another interval. + const second = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(second).toEqual({ considered: 0, queued: 0 }); + }); + + test('a keyword seed is discovered again rather than crawled', async () => { + seeded = await seedDatabase('reseed-keyword'); + const { db } = seeded; + await seededCampaign(db, { kind: 'keyword', value: 'dental practices in Austin' }); + + await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + + const queued = await jobs(db); + expect(queued.map((job) => job.kind)).toEqual(['discover_domains']); + expect(JSON.parse(queued[0]?.payload_json ?? '{}').keyword).toBe('dental practices in Austin'); + }); + + test('a campaign seeded recently is left alone', async () => { + seeded = await seedDatabase('reseed-recent'); + const { db } = seeded; + await seededCampaign(db, { kind: 'url', value: 'acme.com', startedAt: now() }); + + const result = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(result.considered).toBe(0); + expect(await jobs(db)).toHaveLength(0); + }); + + test('a campaign with work outstanding is not idle', async () => { + seeded = await seedDatabase('reseed-busy'); + const { db } = seeded; + await seededCampaign(db, { kind: 'url', value: 'acme.com' }); + + await db.execute({ + sql: `INSERT INTO jobs (id, workspace_id, kind, payload_json, status, attempts, max_attempts, + run_after, created_at, updated_at) + VALUES ('job_busy', ?, 'crawl_site', ?, 'pending', 0, 5, ?, ?, ?)`, + args: [ + SEED.workspaceId, + JSON.stringify({ url: 'https://acme.com/about', campaignId: SEED.campaignId }), + now(), + now(), + now(), + ], + }); + + const result = await reseedIdleCampaigns(db, { workspaceId: SEED.workspaceId }); + expect(result.considered).toBe(0); + expect(await jobs(db)).toHaveLength(1); + }); +}); diff --git a/packages/pipeline/src/reseed.ts b/packages/pipeline/src/reseed.ts new file mode 100644 index 0000000..41abb0c --- /dev/null +++ b/packages/pipeline/src/reseed.ts @@ -0,0 +1,112 @@ +/** + * Reading a campaign's seed again. + * + * A campaign is seeded once, at creation: the URL is crawled, or the keyword + * is expanded into companies and those are crawled. Nothing ever returns to + * the seed. So a directory that gains members, a team page that hires, a + * market that a second discovery run would name differently — none of it + * reaches the product, and once the initial crop of research cards has been + * cleared the intake has nothing left to run on. The digest reported "sites + * read 0, new people 0" for ten consecutive days while six campaigns sat + * active. Nothing was broken; nothing was being asked. + * + * This asks again, on a slow clock, for campaigns that are active and idle. + * Idle means no job of theirs is pending or running, so a campaign still + * working through its first crop is left alone. The crawl dedupe key is + * partial — it only blocks a duplicate while the earlier job is outstanding — + * so the same seed queues cleanly, and the pipeline already recognises a + * person it has read before, so a re-read finds only what is new. + */ + +import { now, queryAll, type Client } from '@outreachgraph/db'; +import { emitEvent } from './events'; +import { enqueue } from './queue'; + +/** How long a campaign is left alone after its seed was last read. */ +const DEFAULT_EVERY_DAYS = 7; + +export interface ReseedResult { + readonly considered: number; + readonly queued: number; +} + +interface IdleCampaign { + readonly id: string; + readonly name: string; + readonly seed_kind: string; + readonly seed_value: string; +} + +export async function reseedIdleCampaigns( + db: Client, + input: { readonly workspaceId: string; readonly everyDays?: number; readonly now?: Date }, +): Promise { + const at = input.now ?? new Date(); + const days = input.everyDays ?? DEFAULT_EVERY_DAYS; + const cutoff = new Date(at.getTime() - days * 86_400_000).toISOString(); + + const idle = await queryAll( + db, + `SELECT c.id, c.name, c.seed_kind, c.seed_value + FROM campaigns c + WHERE c.workspace_id = ? + AND c.status IN ('active', 'running') + AND c.seed_kind IN ('url', 'keyword') + AND c.seed_value IS NOT NULL AND trim(c.seed_value) <> '' + AND COALESCE(c.reseeded_at, c.started_at, c.created_at) < ? + AND NOT EXISTS ( + SELECT 1 FROM jobs j + WHERE j.workspace_id = c.workspace_id + AND j.status IN ('pending', 'running') + AND j.payload_json LIKE '%' || c.id || '%') + ORDER BY COALESCE(c.reseeded_at, c.started_at, c.created_at) ASC`, + [input.workspaceId, cutoff], + ); + + let queued = 0; + + for (const campaign of idle) { + const seed = campaign.seed_value.trim(); + + const result = + campaign.seed_kind === 'url' + ? await enqueue(db, { + workspaceId: input.workspaceId, + kind: 'crawl_site', + payload: { + url: /^https?:\/\//i.test(seed) ? seed : `https://${seed}`, + campaignId: campaign.id, + }, + dedupeKey: `crawl:${campaign.id}:${seed.replace(/^https?:\/\//i, '')}`, + }) + : await enqueue(db, { + workspaceId: input.workspaceId, + kind: 'discover_domains', + payload: { keyword: seed, campaignId: campaign.id }, + dedupeKey: `discover:${campaign.id}`, + }); + + // Stamped whether or not a job was queued: a dedupe hit means the work is + // already outstanding, and asking again next tick would not change that. + await db.execute({ + sql: 'UPDATE campaigns SET reseeded_at = ?, updated_at = ? WHERE id = ?', + args: [now(), now(), campaign.id], + }); + + if (!result.queued) continue; + queued += 1; + + await emitEvent(db, { + workspaceId: input.workspaceId, + campaignId: campaign.id, + phase: 'intake', + message: + campaign.seed_kind === 'url' + ? `Reading ${seed} again for anyone new` + : `Looking again for companies matching “${seed}”`, + detail: { seedKind: campaign.seed_kind, seed, everyDays: days }, + }); + } + + return { considered: idle.length, queued }; +} diff --git a/packages/providers/src/index.ts b/packages/providers/src/index.ts index 2f01791..a668b71 100644 --- a/packages/providers/src/index.ts +++ b/packages/providers/src/index.ts @@ -28,6 +28,17 @@ export { export { deriveEvidence, type EvidenceContext } from './evidence'; +export { + ValueSerpClient, + carriesName, + corroborate, + isLinkedInProfile, + type ProfilePhoto, + type ProfilePhotoFinder, + type ProfilePhotoQuery, + type ValueSerpOptions, +} from './valueserp'; + export { FixtureProvider, FIXTURE_CANDIDATES, diff --git a/packages/providers/src/valueserp/index.ts b/packages/providers/src/valueserp/index.ts new file mode 100644 index 0000000..7e4cefc --- /dev/null +++ b/packages/providers/src/valueserp/index.ts @@ -0,0 +1,199 @@ +/** + * ValueSERP: a picture of the person, found the way a human would find one. + * + * Given a name, a title and a company, a person searches Google Images and + * takes the headshot from the LinkedIn profile or the company's team page. + * This does the same through a search API, and it is deliberately no cleverer + * than that: one query, the first result whose page corroborates the person, + * nothing otherwise. + * + * What makes it precise rather than merely plausible is the page the image + * sits on, not the image. A picture is accepted only when it was published on + * a LinkedIn profile whose title carries the person's name, or on the + * company's own domain. "A face that came up for this name" is not evidence of + * anything; "the picture on the team page of the company we already know they + * work at" is. Identity precision beats recall (PRD §9), and a wrong face on a + * lead is worse than no face. + * + * This reads search results. It never fetches LinkedIn itself, never logs in, + * and never acts there — the human still does that in LinkedIn's own + * interface. What it records is research: a URL the person published under + * their own name. + */ + +export interface ProfilePhotoQuery { + readonly name: string; + readonly title?: string | undefined; + readonly company?: string | undefined; + /** The company's own domain; a picture published there is accepted. */ + readonly companyDomain?: string | undefined; +} + +export interface ProfilePhoto { + readonly photoUrl: string; + /** The page the picture was published on — the corroborating evidence. */ + readonly pageUrl: string; + readonly source: 'linkedin' | 'site'; +} + +export interface ProfilePhotoFinder { + findProfilePhoto(query: ProfilePhotoQuery): Promise; +} + +export interface ValueSerpOptions { + readonly apiKey: string; + readonly baseUrl?: string; + readonly fetchImpl?: typeof fetch; + readonly timeoutMs?: number; +} + +const DEFAULT_BASE = 'https://api.valueserp.com'; + +/** One image result, as much of it as this adapter reads. */ +interface ImageResult { + readonly title?: string; + readonly link?: string; + readonly image?: string; + readonly original?: string; + readonly thumbnail?: string; + readonly domain?: string; + readonly source?: string; +} + +export class ValueSerpClient implements ProfilePhotoFinder { + private readonly apiKey: string; + private readonly baseUrl: string; + private readonly fetchImpl: typeof fetch; + private readonly timeoutMs: number; + + constructor(options: ValueSerpOptions) { + this.apiKey = options.apiKey; + this.baseUrl = (options.baseUrl ?? DEFAULT_BASE).replace(/\/$/, ''); + this.fetchImpl = options.fetchImpl ?? fetch; + this.timeoutMs = options.timeoutMs ?? 10_000; + } + + /** + * One search, one answer, or none. + * + * A miss is the common case and costs a credit either way, which is why the + * caller stamps the person as looked up regardless. Network failures are a + * miss too: this runs in a sweep, and one flaky response should cost that + * person's picture, not the run. + */ + async findProfilePhoto(query: ProfilePhotoQuery): Promise { + const name = query.name.trim(); + if (!name) return undefined; + + const terms = [`"${name}"`, query.company?.trim(), query.title?.trim()].filter(Boolean); + + const url = new URL('/search', this.baseUrl); + url.searchParams.set('api_key', this.apiKey); + url.searchParams.set('search_type', 'images'); + url.searchParams.set('q', terms.join(' ')); + + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), this.timeoutMs); + + try { + const response = await this.fetchImpl(url, { + signal: controller.signal, + headers: { accept: 'application/json' }, + }); + if (!response.ok) return undefined; + + const body = (await response.json()) as { image_results?: ImageResult[] }; + const results = Array.isArray(body.image_results) ? body.image_results : []; + + for (const result of results) { + const match = corroborate(result, query); + if (match) return match; + } + + return undefined; + } catch { + return undefined; + } finally { + clearTimeout(timer); + } + } +} + +/** + * Whether one result is evidence enough. + * + * Exported for the tests and for anyone auditing what "corroborated" means: + * the page is a LinkedIn profile titled with the person's name, or the page is + * on the company's own domain and titled with their name. + */ +export function corroborate( + result: ImageResult, + query: ProfilePhotoQuery, +): ProfilePhoto | undefined { + const photoUrl = firstHttp(result.image, result.original, result.thumbnail); + const pageUrl = firstHttp(result.link, result.source); + if (!photoUrl || !pageUrl) return undefined; + + const title = result.title ?? ''; + if (!carriesName(title, query.name)) return undefined; + + let host: string; + let path: string; + try { + const page = new URL(pageUrl); + host = page.hostname.toLowerCase(); + path = page.pathname; + } catch { + return undefined; + } + + if (isLinkedInProfile(host, path)) return { photoUrl, pageUrl, source: 'linkedin' }; + + const domain = query.companyDomain + ?.trim() + .toLowerCase() + .replace(/^www\./, ''); + if (domain && (host === domain || host.endsWith(`.${domain}`))) { + return { photoUrl, pageUrl, source: 'site' }; + } + + return undefined; +} + +/** `linkedin.com/in/` on any LinkedIn host — a profile, not a company page. */ +export function isLinkedInProfile(host: string, path: string): boolean { + const onLinkedIn = host === 'linkedin.com' || host.endsWith('.linkedin.com'); + return onLinkedIn && /^\/in\/[^/]+/.test(path); +} + +/** + * Every part of the name that could be a name, present in the title. + * + * Initials and particles ("J", "de") are too short to mean anything and are + * not required; "Mark Ramsey" needs both `mark` and `ramsey`. Case and + * diacritics are folded so "Klaudia Majcher" matches "KLAUDIA MAJCHER" and + * "Stefan Wienold" matches a title that spells it with a different accent. + */ +export function carriesName(title: string, name: string): boolean { + const haystack = fold(title); + const parts = fold(name) + .split(/[\s,]+/) + .filter((part) => part.length >= 2); + + if (parts.length === 0) return false; + return parts.every((part) => haystack.includes(part)); +} + +function fold(value: string): string { + return value + .normalize('NFD') + .replace(/[\u0300-\u036f]/g, '') + .toLowerCase(); +} + +function firstHttp(...candidates: (string | undefined)[]): string | undefined { + for (const candidate of candidates) { + if (typeof candidate === 'string' && /^https?:\/\//i.test(candidate)) return candidate; + } + return undefined; +} diff --git a/packages/providers/src/valueserp/valueserp.test.ts b/packages/providers/src/valueserp/valueserp.test.ts new file mode 100644 index 0000000..d45d4b3 --- /dev/null +++ b/packages/providers/src/valueserp/valueserp.test.ts @@ -0,0 +1,132 @@ +/** + * What counts as the right face. + * + * The adapter's whole value is in what it refuses. A search for a common name + * returns a page of strangers; only a result whose page vouches for the person + * — their LinkedIn profile, their employer's site — is worth attaching to them. + */ + +import { describe, expect, test } from 'bun:test'; +import { ValueSerpClient, carriesName, corroborate, isLinkedInProfile } from './index'; + +const JANE = { + name: 'Jane Smith', + title: 'VP Engineering', + company: 'Acme', + companyDomain: 'acme.com', +}; + +describe('corroborate', () => { + test('accepts a LinkedIn profile titled with the name', () => { + const match = corroborate( + { + title: 'Jane Smith - VP Engineering - Acme | LinkedIn', + link: 'https://www.linkedin.com/in/janesmith', + image: 'https://media.licdn.com/dms/image/jane.jpg', + }, + JANE, + ); + expect(match).toEqual({ + photoUrl: 'https://media.licdn.com/dms/image/jane.jpg', + pageUrl: 'https://www.linkedin.com/in/janesmith', + source: 'linkedin', + }); + }); + + test('accepts the company’s own site', () => { + const match = corroborate( + { + title: 'Our team — Jane Smith', + link: 'https://www.acme.com/team', + image: 'https://www.acme.com/img/jane.jpg', + }, + JANE, + ); + expect(match?.source).toBe('site'); + }); + + test('rejects a LinkedIn company page, a stranger’s site, and a title without the name', () => { + const image = 'https://cdn.example/jane.jpg'; + + expect( + corroborate( + { title: 'Jane Smith', link: 'https://www.linkedin.com/company/acme', image }, + JANE, + ), + ).toBeUndefined(); + expect( + corroborate({ title: 'Jane Smith', link: 'https://someblog.example/jane', image }, JANE), + ).toBeUndefined(); + expect( + corroborate( + { title: 'Jane Doe - Acme | LinkedIn', link: 'https://linkedin.com/in/jd', image }, + JANE, + ), + ).toBeUndefined(); + }); + + test('needs both an image and a page', () => { + expect( + corroborate({ title: 'Jane Smith | LinkedIn', link: 'https://linkedin.com/in/js' }, JANE), + ).toBeUndefined(); + }); +}); + +describe('carriesName', () => { + test('folds case and accents and ignores initials', () => { + expect(carriesName('KLAUDIA MAJCHER – Właściciel', 'Klaudia Majcher')).toBe(true); + expect(carriesName('Stefan Wienold | Vertriebstrainer', 'Stefan Wienöld')).toBe(true); + expect(carriesName('Mark J. Ramsey — Citipointe', 'Mark J Ramsey')).toBe(true); + expect(carriesName('Mark Ramsay — Citipointe', 'Mark Ramsey')).toBe(false); + }); +}); + +describe('isLinkedInProfile', () => { + test('is a person, on any LinkedIn host', () => { + expect(isLinkedInProfile('www.linkedin.com', '/in/jane')).toBe(true); + expect(isLinkedInProfile('uk.linkedin.com', '/in/jane/')).toBe(true); + expect(isLinkedInProfile('linkedin.com', '/company/acme')).toBe(false); + expect(isLinkedInProfile('notlinkedin.com', '/in/jane')).toBe(false); + }); +}); + +describe('ValueSerpClient', () => { + function respond(body: unknown, status = 200): { calls: URL[]; fetchImpl: typeof fetch } { + const calls: URL[] = []; + const fetchImpl = (async (input: string | URL | Request) => { + calls.push(new URL(String(input instanceof Request ? input.url : input))); + return new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json' }, + }); + }) as unknown as typeof fetch; + return { calls, fetchImpl }; + } + + test('searches images for the quoted name with title and company', async () => { + const { calls, fetchImpl } = respond({ + image_results: [ + { title: 'Someone Else', link: 'https://linkedin.com/in/else', image: 'https://x/1.jpg' }, + { + title: 'Jane Smith - Acme | LinkedIn', + link: 'https://linkedin.com/in/jane', + image: 'https://media.licdn.com/jane.jpg', + }, + ], + }); + + const client = new ValueSerpClient({ apiKey: 'k', fetchImpl }); + const photo = await client.findProfilePhoto(JANE); + + expect(photo?.photoUrl).toBe('https://media.licdn.com/jane.jpg'); + expect(calls[0]?.searchParams.get('search_type')).toBe('images'); + expect(calls[0]?.searchParams.get('q')).toBe('"Jane Smith" Acme VP Engineering'); + expect(calls[0]?.searchParams.get('api_key')).toBe('k'); + }); + + test('a refused request is a miss, not an error', async () => { + const { fetchImpl } = respond({ request_info: { success: false } }, 401); + const client = new ValueSerpClient({ apiKey: 'k', fetchImpl }); + expect(await client.findProfilePhoto(JANE)).toBeUndefined(); + }); +});