From cc9261edcecb817828d0be7cad6b90967837b48a Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 13 Sep 2026 04:20:11 +0000 Subject: [PATCH] Public directory: the crawler's public half, keyless, for nichedb.dev Anthony: "from all our public data put it all in nichedb.dev too from outreachgraph.com crawler/scanner". nichedb ingests only by pulling, so this is the endpoint it pulls: GET /api/v1/public/directory, above the session guard, paged by since + cursor (keyset on updated, id), cached five minutes, sixty requests a minute per caller. The line it holds, in one query: a company or site by its domain; a person only when they publish their own profile, meaning an OpenProfile.md they serve (published_url) or a profile and a home page that point at each other with rel=me, which the openprofile job now records as openprofiles.corroborated (migration 0036). Never an email, a phone, a location, a bio, a score, a signal, a campaign, a note or a workspace id; a person known from a scraped handle alone is not listed. Topics come from the industry and technologies on a company, and only from the ## Topics line of a person's profile. Tests: 15 for the route, the withholding rules, paging, the cursor and the limiter; the full suite is 1506 passing. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_014cmNRtR2vL1p89dbVQ7FZJ --- README.md | 13 + apps/api/src/app.ts | 51 +++ apps/api/src/public-directory.test.ts | 390 +++++++++++++++++++ apps/api/src/public-directory.ts | 336 ++++++++++++++++ migrations/0036_openprofile_corroborated.sql | 22 ++ packages/pipeline/src/openprofile.ts | 16 +- 6 files changed, 824 insertions(+), 4 deletions(-) create mode 100644 apps/api/src/public-directory.test.ts create mode 100644 apps/api/src/public-directory.ts create mode 100644 migrations/0036_openprofile_corroborated.sql diff --git a/README.md b/README.md index d354d88..58bdbaa 100644 --- a/README.md +++ b/README.md @@ -190,3 +190,16 @@ production customer data is never copied into staging. See [`CLAUDE.md`](CLAUDE.md). The short version: TypeScript strict, ESM, kebab-case, colocated tests, forward-only migrations, and a set of non-negotiables that come from the PRD rather than from taste. + +## Public directory + +`GET /api/v1/public/directory` is the one keyless read: what the crawler learned +from pages that were already public, about things that are public by nature. +Companies and sites by their domain, and people only when they publish their own +profile (an OpenProfile.md they serve themselves, or a profile and a home page +that point at each other with rel=me). Never an email, a phone, a location, a +score, a signal, a campaign or a workspace. One row shape, +`{ id, kind: company|site|person, name, url, description, topics, country, openprofile, updated }`, +paged by `since` and an opaque `cursor`, cacheable for five minutes, sixty +requests a minute per caller. The rule lives in `apps/api/src/public-directory.ts` +and nowhere else; nichedb.dev reads it into its directory collection. diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 44b8540..2b159a1 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -166,6 +166,15 @@ import { verificationEmail, type Mailer, } from '@outreachgraph/email'; +import { + DIRECTORY_CACHE_SECONDS, + DIRECTORY_DEFAULT_LIMIT, + DIRECTORY_RATE_LIMIT, + DIRECTORY_RATE_WINDOW_MS, + FixedWindowLimiter, + clientKey, + listPublicDirectory, +} from './public-directory'; import { ApiError, canApprove, type AppEnv, type RequestActor } from './context'; import * as repo from './repository'; import { @@ -226,6 +235,11 @@ export interface AppOptions { readonly suggestSubreddits?: typeof suggestSubreddits; /** Public origin, used to build links that land in someone's inbox. */ readonly appUrl?: string; + /** + * Throttles the keyless public directory. Tests inject a small one; + * production leaves it unset and gets sixty requests a minute per caller. + */ + readonly publicDirectoryLimiter?: FixedWindowLimiter; /** * Sells credit packs over CoinPayPortal. Omit and the billing routes answer * 503 rather than 500 — a deployment without payment credentials is a @@ -722,6 +736,43 @@ export function createApp(options: AppOptions): Hono { } }); + /** + * The public directory: companies, sites and self-published people, keyless. + * + * Sits above the session guard because it is meant for readers with no + * account -- the nichedb.dev directory is the first. What it may say is + * decided in `public-directory.ts`, in one query, and nothing here adds a + * field to it. The page is cacheable by anyone for five minutes, and a + * caller is throttled per address so a runaway loop cannot turn a cheap read + * into a load; a reader walking the whole directory needs a handful of + * requests, not sixty a minute. + * + * Paging is by `since` and an opaque `cursor`, ascending by update time, so + * a reader can resume from where it stopped and pick up what changed since. + */ + const directoryLimiter = + options.publicDirectoryLimiter ?? + new FixedWindowLimiter(DIRECTORY_RATE_LIMIT, DIRECTORY_RATE_WINDOW_MS); + + api.get('/public/directory', async (c) => { + const verdict = directoryLimiter.take(clientKey(c.req.raw)); + c.header('x-ratelimit-remaining', String(verdict.remaining)); + if (!verdict.allowed) { + c.header('retry-after', String(verdict.retryAfterSeconds)); + throw new ApiError(429, 'rate_limited', 'too many requests; slow down and retry'); + } + + const rawLimit = c.req.query('limit'); + const page = await listPublicDirectory(options.db, { + since: c.req.query('since'), + cursor: c.req.query('cursor'), + limit: rawLimit === undefined ? DIRECTORY_DEFAULT_LIMIT : Number(rawLimit), + }); + + c.header('cache-control', `public, max-age=${DIRECTORY_CACHE_SECONDS}`); + return c.json(page); + }); + // Everything else under /api/v1 is authenticated and workspace-scoped. api.use('*', async (c, next) => { const actor = await resolveActor(c.req.raw); diff --git a/apps/api/src/public-directory.test.ts b/apps/api/src/public-directory.test.ts new file mode 100644 index 0000000..20aed6e --- /dev/null +++ b/apps/api/src/public-directory.test.ts @@ -0,0 +1,390 @@ +/** + * The public directory, and the line it holds. + * + * Two things matter here. That the route is reachable with no session and no + * token at all, since its reader has neither. And that what it says is exactly + * the public half: a company by its domain, a person only when they publish + * their own profile, and never an address, a location, a score or a workspace. + * The withholding tests are the ones that must not be loosened. + */ + +import { afterEach, describe, expect, test } from 'bun:test'; +import type { Hono } from 'hono'; +import { createApp } from './app'; +import type { AppEnv } from './context'; +import { + FixedWindowLimiter, + companyTopics, + decodeCursor, + encodeCursor, + listPublicDirectory, + topicsFromOpenProfile, +} from './public-directory'; +import { seedDatabase, SEED, type SeededDatabase } from './test-seed'; + +let active: SeededDatabase | undefined; + +afterEach(() => { + active?.cleanup(); + active = undefined; +}); + +const T0 = '2026-09-01T00:00:00.000Z'; +const T1 = '2026-09-02T00:00:00.000Z'; +const T2 = '2026-09-03T00:00:00.000Z'; + +/** An app whose authentication answers nobody, which is what a public reader is. */ +async function harness( + label: string, + limiter?: FixedWindowLimiter, +): Promise<{ app: Hono; seeded: SeededDatabase }> { + const seeded = await seedDatabase(label); + active = seeded; + const app = createApp({ + db: seeded.db, + authenticate: async () => undefined, + ...(limiter ? { publicDirectoryLimiter: limiter } : {}), + }); + return { app, seeded }; +} + +async function get(app: Hono, query = ''): Promise { + return app.request(`/api/v1/public/directory${query}`); +} + +interface PersonFixture { + id: string; + name: string; + status?: string; + kind?: string; + publishedUrl?: string | null; + corroborated?: boolean; + markdown?: string; + updatedAt?: string; +} + +async function person(seeded: SeededDatabase, p: PersonFixture): Promise { + const stamp = p.updatedAt ?? T1; + await seeded.db.execute({ + sql: `INSERT INTO people (id, display_name, current_company_id, current_title, location, + identity_confidence, status, kind, outreach_eligible, believed_minor, created_at, updated_at) + VALUES (?, ?, ?, 'Maintainer', 'Lisbon, Portugal', 0.9, ?, ?, 1, 0, ?, ?)`, + args: [p.id, p.name, SEED.companyId, p.status ?? 'active', p.kind ?? 'person', stamp, stamp], + }); + if (p.publishedUrl !== undefined || p.corroborated !== undefined || p.markdown) { + await seeded.db.execute({ + sql: `INSERT INTO openprofiles (person_id, markdown, sources_json, published_url, corroborated, generated_at) + VALUES (?, ?, '[]', ?, ?, ?)`, + args: [ + p.id, + p.markdown ?? `# ${p.name}\n\n- **Kind**: person\n- **Handle**: @${p.id}\n`, + p.publishedUrl ?? null, + p.corroborated ? 1 : 0, + stamp, + ], + }); + } +} + +async function identity( + seeded: SeededDatabase, + personId: string, + network: string, + handle: string, + profileUrl: string | null, + confidence = 0.9, +): Promise { + await seeded.db.execute({ + sql: `INSERT INTO social_identities (id, person_id, network, handle, platform_user_id, profile_url, + confidence, source_type, verified_by, first_seen_at, last_verified_at) + VALUES (?, ?, ?, ?, NULL, ?, ?, 'public_web', '[]', ?, ?)`, + args: [`sid_${personId}_${network}`, personId, network, handle, profileUrl, confidence, T1, T1], + }); +} + +describe('public directory', () => { + test('answers without a session or token, cacheable for five minutes', async () => { + const { app } = await harness('dir-keyless'); + const response = await get(app); + + expect(response.status).toBe(200); + expect(response.headers.get('cache-control')).toBe('public, max-age=300'); + expect(response.headers.get('access-control-allow-origin')).toBe('*'); + + const body = (await response.json()) as { items: unknown[]; next: string | null }; + expect(Array.isArray(body.items)).toBe(true); + expect(body.next).toBeNull(); + }); + + test('lists a crawled company by its domain, with industry and technologies as topics', async () => { + const { app, seeded } = await harness('dir-company'); + await seeded.db.execute({ + sql: `UPDATE companies SET technologies = '["Postgres", "bun"]', contact_email = 'hello@acme.com', + location = '1 Market St, San Francisco' WHERE id = ?`, + args: [SEED.companyId], + }); + + const body = (await (await get(app)).json()) as { items: Record[] }; + const acme = body.items.find((item) => item.id === SEED.companyId); + + expect(acme).toEqual({ + id: SEED.companyId, + kind: 'company', + name: 'Acme', + url: 'https://acme.com', + description: null, + topics: ['saas', 'postgres', 'bun'], + country: null, + openprofile: null, + updated: expect.any(String), + }); + }); + + test('a row named only after its domain is a site, not a company', async () => { + const { app, seeded } = await harness('dir-site'); + await seeded.db.execute({ + sql: `INSERT INTO companies (id, name, domain, technologies, created_at, updated_at) + VALUES ('co_site', 'example.org', 'example.org', '[]', ?, ?)`, + args: [T1, T1], + }); + + const body = (await (await get(app)).json()) as { items: Record[] }; + const site = body.items.find((item) => item.id === 'co_site'); + + expect(site?.kind).toBe('site'); + expect(site?.url).toBe('https://example.org'); + }); + + test('a company without a domain is not listed', async () => { + const { app, seeded } = await harness('dir-nodomain'); + await seeded.db.execute({ + sql: `INSERT INTO companies (id, name, domain, technologies, created_at, updated_at) + VALUES ('co_ghost', 'Ghost Ltd', NULL, '[]', ?, ?)`, + args: [T1, T1], + }); + + const body = (await (await get(app)).json()) as { items: { id: string }[] }; + expect(body.items.map((item) => item.id)).not.toContain('co_ghost'); + }); + + test('a person is listed only when they publish their own profile', async () => { + const { app, seeded } = await harness('dir-people'); + + // The seeded Jane has no OpenProfile at all: a prospect, not a publisher. + await person(seeded, { + id: 'per_assembled', + name: 'Assembled Only', + publishedUrl: null, + corroborated: false, + }); + await person(seeded, { + id: 'per_published', + name: 'Ada Publishes', + publishedUrl: 'https://ada.example/.well-known/openprofile.md', + markdown: + '# Ada\n\n- **Handle**: @ada\n- **Email**: ada@example.com\n\n## Topics\n\n- #rust, distributed systems, Rust\n', + }); + await identity(seeded, 'per_published', 'website', 'ada.example', 'https://ada.example'); + await identity(seeded, 'per_published', 'email', 'ada@example.com', null); + await person(seeded, { id: 'per_relme', name: 'Rel Me', corroborated: true }); + await identity( + seeded, + 'per_relme', + 'mastodon', + 'rel@hachyderm.io', + 'https://hachyderm.io/@rel', + ); + await person(seeded, { + id: 'per_suppressed', + name: 'Opted Out', + publishedUrl: 'https://gone.example/.well-known/openprofile.md', + status: 'suppressed', + }); + await person(seeded, { + id: 'per_inbox', + name: 'Acme team', + kind: 'company_inbox', + corroborated: true, + }); + + const body = (await (await get(app)).json()) as { items: Record[] }; + const ids = body.items.map((item) => item.id); + + expect(ids).toContain('per_published'); + expect(ids).toContain('per_relme'); + expect(ids).not.toContain(SEED.personId); + expect(ids).not.toContain('per_assembled'); + expect(ids).not.toContain('per_suppressed'); + expect(ids).not.toContain('per_inbox'); + + const ada = body.items.find((item) => item.id === 'per_published'); + expect(ada).toEqual({ + id: 'per_published', + kind: 'person', + name: 'Ada Publishes', + url: 'https://ada.example', + description: 'Maintainer at Acme', + topics: ['rust', 'distributed systems'], + country: null, + openprofile: 'https://ada.example/.well-known/openprofile.md', + updated: T1, + }); + + const rel = body.items.find((item) => item.id === 'per_relme'); + expect(rel?.url).toBe('https://hachyderm.io/@rel'); + expect(rel?.openprofile).toBeNull(); + }); + + test('never carries an address, a location, a score or a workspace', async () => { + const { app, seeded } = await harness('dir-withheld'); + await seeded.db.execute({ + sql: `UPDATE companies SET contact_email = 'support@acme.com', location = '1 Market St' WHERE id = ?`, + args: [SEED.companyId], + }); + await person(seeded, { + id: 'per_published', + name: 'Ada Publishes', + publishedUrl: 'https://ada.example/.well-known/openprofile.md', + markdown: '# Ada\n\n- **Email**: ada@example.com\n\n## Topics\n\n- rust\n', + }); + await identity(seeded, 'per_published', 'email', 'ada@example.com', null); + + const text = await (await get(app)).text(); + + expect(text).not.toContain('@acme.com'); + expect(text).not.toContain('ada@example.com'); + expect(text).not.toContain('Market St'); + expect(text).not.toContain('Lisbon'); + expect(text).not.toContain(SEED.workspaceId); + expect(text).not.toContain('identity_confidence'); + expect(text).not.toContain('score'); + + const body = JSON.parse(text) as { items: Record[] }; + for (const item of body.items) { + expect(Object.keys(item).sort()).toEqual([ + 'country', + 'description', + 'id', + 'kind', + 'name', + 'openprofile', + 'topics', + 'updated', + 'url', + ]); + } + }); + + test('pages by cursor without repeating or skipping, and filters by since', async () => { + const { app, seeded } = await harness('dir-paging'); + await seeded.db.execute({ + sql: `UPDATE companies SET updated_at = ? WHERE id = ?`, + args: [T0, SEED.companyId], + }); + await person(seeded, { + id: 'per_a', + name: 'A', + publishedUrl: 'https://a.example/.well-known/openprofile.md', + updatedAt: T1, + }); + await person(seeded, { + id: 'per_b', + name: 'B', + publishedUrl: 'https://b.example/.well-known/openprofile.md', + updatedAt: T2, + }); + + const first = (await (await get(app, '?limit=2')).json()) as { + items: { id: string; updated: string }[]; + next: string | null; + }; + expect(first.items.map((item) => item.id)).toEqual([SEED.companyId, 'per_a']); + expect(first.next).not.toBeNull(); + + const second = (await (await get(app, `?limit=2&cursor=${first.next}`)).json()) as { + items: { id: string }[]; + next: string | null; + }; + expect(second.items.map((item) => item.id)).toEqual(['per_b']); + expect(second.next).toBeNull(); + + const since = (await (await get(app, `?since=${encodeURIComponent(T1)}`)).json()) as { + items: { id: string }[]; + }; + expect(since.items.map((item) => item.id)).toEqual(['per_a', 'per_b']); + }); + + test('refuses a malformed since or cursor rather than guessing', async () => { + const { app } = await harness('dir-bad-input'); + + const since = await get(app, '?since=yesterday'); + expect(since.status).toBe(400); + + const cursor = await get(app, '?cursor=not-a-cursor'); + expect(cursor.status).toBe(400); + }); + + test('throttles a caller who exceeds the window and says when to retry', async () => { + const limiter = new FixedWindowLimiter(2, 60_000); + const { app } = await harness('dir-throttle', limiter); + + expect((await get(app)).status).toBe(200); + expect((await get(app)).status).toBe(200); + const third = await get(app); + expect(third.status).toBe(429); + expect(Number(third.headers.get('retry-after'))).toBeGreaterThan(0); + + const other = await app.request('/api/v1/public/directory', { + headers: { 'x-forwarded-for': '203.0.113.9, 10.0.0.1' }, + }); + expect(other.status).toBe(200); + }); + + test('other routes stay authenticated', async () => { + const { app } = await harness('dir-guard'); + const response = await app.request('/api/v1/prospects'); + expect(response.status).toBe(401); + }); +}); + +describe('directory helpers', () => { + test('the cursor round-trips and rejects nonsense', () => { + const cursor = encodeCursor(T1, 'co_x'); + expect(decodeCursor(cursor)).toEqual({ updated: T1, id: 'co_x' }); + expect(() => decodeCursor('')).toThrow(); + expect(() => decodeCursor(Buffer.from('no-separator').toString('base64url'))).toThrow(); + expect(() => decodeCursor(Buffer.from('nope|co_x').toString('base64url'))).toThrow(); + }); + + test('company topics merge industry and technologies, lower-cased and unique', () => { + expect(companyTopics('SaaS', '["Postgres", "saas", 7, " "]')).toEqual(['saas', 'postgres']); + expect(companyTopics(null, 'not json')).toEqual([]); + expect(companyTopics(null, null)).toEqual([]); + }); + + test('topics are read from the Topics section only', () => { + expect(topicsFromOpenProfile('# X\n\n- **Email**: x@y.z\n\n## Topics\n\n- #A, b ,B\n')).toEqual( + ['a', 'b'], + ); + expect(topicsFromOpenProfile('# X\n\n## Links\n\n- [a](https://a)\n')).toEqual([]); + expect(topicsFromOpenProfile(null)).toEqual([]); + }); + + test('the limiter resets when the window ends', () => { + let now = 0; + const limiter = new FixedWindowLimiter(1, 1000, () => now); + expect(limiter.take('a').allowed).toBe(true); + expect(limiter.take('a')).toMatchObject({ allowed: false, retryAfterSeconds: 1 }); + now = 1000; + expect(limiter.take('a').allowed).toBe(true); + }); + + test('the query helper defaults and clamps the limit', async () => { + const seeded = await seedDatabase('dir-limit'); + active = seeded; + const page = await listPublicDirectory(seeded.db, { limit: 0 }); + expect(page.items.length).toBeLessThanOrEqual(1); + const wide = await listPublicDirectory(seeded.db, { limit: 10_000 }); + expect(wide.next).toBeNull(); + }); +}); diff --git a/apps/api/src/public-directory.ts b/apps/api/src/public-directory.ts new file mode 100644 index 0000000..65389d0 --- /dev/null +++ b/apps/api/src/public-directory.ts @@ -0,0 +1,336 @@ +/** + * The public directory: what the crawler learned from pages that were already + * public, about things that are public by nature. + * + * Everything OutreachGraph knows arrived from the open web, but not everything + * it knows is public data. A company's home page is; the fact that a workspace + * is running a campaign against it is not. A person's own OpenProfile.md is; + * the bio a stranger pasted in from their timeline, the score the engine gave + * them and the address the enrichment sweep guessed are not. This module draws + * that line once, in one query, so a reader who wants the public half (the + * nichedb.dev directory is the first) gets exactly that half and nothing + * leaks through a second, looser path later. + * + * WHAT IS LISTED + * + * - A company or site: a `companies` row with a domain. `company` when the + * page named an organisation, `site` when it did not and the row is only + * the domain the crawler read. + * - A person, only when they publish their own profile: an `openprofiles` row + * that either holds an OpenProfile.md they serve themselves + * (`published_url`) or was corroborated, meaning the profile named a home + * page and the home page pointed back with rel=me. A person known from a + * scraped handle alone is not listed, whatever their confidence says. + * + * WHAT IS NOT + * + * No email address, no phone, no `contact_email`, no `location` (a street + * address for a company, a city for a person), no campaign membership, no + * signal, no score, no note, no workspace id, no avatar hot-linked from a + * network's CDN. The row shape below is the whole export; a field not in it + * is withheld, and adding one is a decision to make here, not in a caller. + * + * `companies` and `people` carry no workspace_id: they are the shared identity + * graph, so nothing here needs a scope and nothing here can leak one. + * + * PAGING + * + * Keyset on `(updated, id)` ascending, so a reader can walk forward from a + * `since` timestamp and resume from the opaque cursor without ever repeating + * or skipping a row that was updated while it read. The cursor is the last + * row's `(updated, id)` base64url-encoded; there is nothing to guess in it and + * nothing to hide. + */ + +import { queryAll, type Client } from '@outreachgraph/db'; +import { ApiError } from './context'; + +export type DirectoryKind = 'company' | 'site' | 'person'; + +export interface DirectoryItem { + readonly id: string; + readonly kind: DirectoryKind; + readonly name: string; + readonly url: string | null; + readonly description: string | null; + readonly topics: readonly string[]; + readonly country: string | null; + /** The OpenProfile.md the person serves themselves, or null. */ + readonly openprofile: string | null; + /** ISO timestamp of the last change to the row. */ + readonly updated: string; +} + +export interface DirectoryPage { + readonly items: readonly DirectoryItem[]; + readonly next: string | null; +} + +export interface DirectoryQuery { + /** Only rows updated at or after this ISO timestamp. */ + readonly since?: string | undefined; + /** Resume after the row a previous page ended on. */ + readonly cursor?: string | undefined; + readonly limit?: number | undefined; +} + +export const DIRECTORY_DEFAULT_LIMIT = 100; +export const DIRECTORY_MAX_LIMIT = 200; + +/** Seconds a page may be served from a shared cache. */ +export const DIRECTORY_CACHE_SECONDS = 300; + +interface DirectoryRow { + kind: string; + id: string; + name: string; + domain: string | null; + industry: string | null; + technologies: string | null; + title: string | null; + company_name: string | null; + markdown: string | null; + published_url: string | null; + home_url: string | null; + profile_url: string | null; + updated_at: string; +} + +/** + * One query, both tables, one order. + * + * The person half is where the rule lives. `openprofiles` is joined rather + * than left-joined because a person without one has no self-published + * profile by definition; `published_url` and `corroborated` are the two ways + * they can have one. `kind = 'person'` drops the company-inbox leads, which + * are a mailbox wearing a person row. `status = 'active'` drops the suppressed + * and the deleted; a tombstone must never resurface in a public list. + * + * The person's URL is their home page when a `website` identity was recorded + * for them, else the profile the home page vouched for. Email rows are + * excluded from both subqueries by network, not by pattern. + */ +const DIRECTORY_SQL = ` + SELECT * FROM ( + SELECT 'company' AS kind, c.id, c.name, c.domain, c.industry, c.technologies, + NULL AS title, NULL AS company_name, NULL AS markdown, NULL AS published_url, + NULL AS home_url, NULL AS profile_url, c.updated_at + FROM companies c + WHERE c.domain IS NOT NULL AND c.domain <> '' + UNION ALL + SELECT 'person' AS kind, p.id, p.display_name AS name, NULL AS domain, NULL AS industry, + NULL AS technologies, p.current_title AS title, co.name AS company_name, + o.markdown, o.published_url, + (SELECT s.profile_url FROM social_identities s + WHERE s.person_id = p.id AND s.network = 'website' + AND s.profile_url LIKE 'http%' + ORDER BY s.confidence DESC, s.first_seen_at ASC LIMIT 1) AS home_url, + (SELECT s.profile_url FROM social_identities s + WHERE s.person_id = p.id AND s.network NOT IN ('email', 'website') + AND s.profile_url LIKE 'http%' + ORDER BY s.confidence DESC, s.first_seen_at ASC LIMIT 1) AS profile_url, + MAX(p.updated_at, o.generated_at) AS updated_at + FROM people p + JOIN openprofiles o ON o.person_id = p.id + LEFT JOIN companies co ON co.id = p.current_company_id + WHERE p.status = 'active' AND p.kind = 'person' + AND (o.published_url IS NOT NULL OR o.corroborated = 1) + ) d + WHERE (? IS NULL OR d.updated_at >= ?) + AND (? IS NULL OR d.updated_at > ? OR (d.updated_at = ? AND d.id > ?)) + ORDER BY d.updated_at ASC, d.id ASC + LIMIT ? +`; + +export async function listPublicDirectory( + db: Client, + query: DirectoryQuery = {}, +): Promise { + const limit = clampLimit(query.limit); + const since = parseSince(query.since); + const after = query.cursor ? decodeCursor(query.cursor) : undefined; + + const rows = await queryAll(db, DIRECTORY_SQL, [ + since ?? null, + since ?? null, + after?.updated ?? null, + after?.updated ?? null, + after?.updated ?? null, + after?.id ?? null, + limit + 1, + ]); + + const page = rows.slice(0, limit).map(toItem); + const last = page[page.length - 1]; + const next = rows.length > limit && last ? encodeCursor(last.updated, last.id) : null; + + return { items: page, next }; +} + +function toItem(row: DirectoryRow): DirectoryItem { + if (row.kind === 'person') { + return { + id: row.id, + kind: 'person', + name: row.name, + url: row.home_url ?? row.profile_url ?? row.published_url, + description: describePerson(row.title, row.company_name), + topics: topicsFromOpenProfile(row.markdown), + country: null, + openprofile: row.published_url, + updated: row.updated_at, + }; + } + + const domain = (row.domain ?? '').trim().toLowerCase(); + // The crawler names a company after its domain when the page named nobody; + // that row is a site we read, not an organisation we can vouch for. + const kind: DirectoryKind = row.name.trim().toLowerCase() === domain ? 'site' : 'company'; + + return { + id: row.id, + kind, + name: row.name, + url: `https://${domain}`, + // Nothing the crawler keeps on the company row describes it: the OpenGraph + // description is read and used for drafting but never stored. Withheld + // rather than invented. + description: null, + topics: companyTopics(row.industry, row.technologies), + country: null, + openprofile: null, + updated: row.updated_at, + }; +} + +/** "VP Engineering at Acme", or just the title, or nothing. Never a bio. */ +function describePerson(title: string | null, company: string | null): string | null { + const t = title?.trim(); + const c = company?.trim(); + if (t && c) return `${t} at ${c}`; + return t || null; +} + +/** The industry plus every technology the row carries, lower-cased and unique. */ +export function companyTopics(industry: string | null, technologies: string | null): string[] { + const topics = new Set(); + const add = (value: unknown) => { + if (typeof value !== 'string') return; + const t = value.trim().toLowerCase(); + if (t) topics.add(t); + }; + add(industry); + if (technologies) { + try { + const parsed: unknown = JSON.parse(technologies); + if (Array.isArray(parsed)) for (const value of parsed) add(value); + } catch { + // A malformed list is nothing to export, not an error to raise. + } + } + return [...topics].slice(0, 40); +} + +/** + * The `## Topics` line of an OpenProfile.md: `- a, b, c`. + * + * The only part of the markdown that is read. Topics are what the person + * tagged themselves with, in their own file or their own bio; the rest of the + * document may carry an email line, so it is never exported wholesale. + */ +export function topicsFromOpenProfile(markdown: string | null): string[] { + if (!markdown) return []; + const match = /^## Topics\s*\n+\s*-\s*(.+)$/m.exec(markdown); + if (!match?.[1]) return []; + const topics = new Set(); + for (const raw of match[1].split(',')) { + const t = raw.trim().replace(/^#/, '').toLowerCase(); + if (t) topics.add(t); + } + return [...topics].slice(0, 40); +} + +function clampLimit(value: number | undefined): number { + if (value === undefined || Number.isNaN(value)) return DIRECTORY_DEFAULT_LIMIT; + return Math.min(Math.max(Math.floor(value), 1), DIRECTORY_MAX_LIMIT); +} + +function parseSince(value: string | undefined): string | undefined { + if (value === undefined || value === '') return undefined; + const ms = Date.parse(value); + if (Number.isNaN(ms)) throw ApiError.badRequest('since must be an ISO 8601 timestamp'); + return new Date(ms).toISOString(); +} + +export function encodeCursor(updated: string, id: string): string { + return Buffer.from(`${updated}|${id}`, 'utf8').toString('base64url'); +} + +export function decodeCursor(cursor: string): { updated: string; id: string } { + let decoded: string; + try { + decoded = Buffer.from(cursor, 'base64url').toString('utf8'); + } catch { + throw ApiError.badRequest('cursor is not valid'); + } + const separator = decoded.lastIndexOf('|'); + const updated = separator > 0 ? decoded.slice(0, separator) : ''; + const id = separator > 0 ? decoded.slice(separator + 1) : ''; + if (!updated || !id || Number.isNaN(Date.parse(updated))) { + throw ApiError.badRequest('cursor is not valid'); + } + return { updated, id }; +} + +/** + * A fixed window per caller, in memory. + * + * The deployment is one container pinned to one replica, so a map is the whole + * truth. The endpoint is cheap and cacheable for five minutes, so the limit is + * there to stop a runaway loop, not to meter anyone: a reader walking the + * whole directory at 200 a page needs a handful of requests. + */ +export class FixedWindowLimiter { + private readonly hits = new Map(); + + constructor( + private readonly limit: number, + private readonly windowMs: number, + private readonly clock: () => number = () => Date.now(), + ) {} + + /** Records one request and says whether it is within the window's allowance. */ + take(key: string): { allowed: boolean; retryAfterSeconds: number; remaining: number } { + const now = this.clock(); + const entry = this.hits.get(key); + if (!entry || entry.resetAt <= now) { + this.hits.set(key, { count: 1, resetAt: now + this.windowMs }); + if (this.hits.size > 10_000) this.sweep(now); + return { allowed: true, retryAfterSeconds: 0, remaining: this.limit - 1 }; + } + entry.count += 1; + const remaining = Math.max(this.limit - entry.count, 0); + if (entry.count > this.limit) { + return { + allowed: false, + retryAfterSeconds: Math.max(Math.ceil((entry.resetAt - now) / 1000), 1), + remaining, + }; + } + return { allowed: true, retryAfterSeconds: 0, remaining }; + } + + private sweep(now: number): void { + for (const [key, entry] of this.hits) if (entry.resetAt <= now) this.hits.delete(key); + } +} + +export const DIRECTORY_RATE_LIMIT = 60; +export const DIRECTORY_RATE_WINDOW_MS = 60_000; + +/** The caller as the edge saw it: first forwarded hop, else the one Bun reports. */ +export function clientKey(request: Request, fallback = 'unknown'): string { + const forwarded = request.headers.get('x-forwarded-for'); + const first = forwarded?.split(',')[0]?.trim(); + return first || request.headers.get('x-real-ip')?.trim() || fallback; +} diff --git a/migrations/0036_openprofile_corroborated.sql b/migrations/0036_openprofile_corroborated.sql new file mode 100644 index 0000000..f5790f3 --- /dev/null +++ b/migrations/0036_openprofile_corroborated.sql @@ -0,0 +1,22 @@ +-- 0036_openprofile_corroborated.sql +-- +-- Whether the person's home page vouched for the profile we started from. +-- +-- The openprofile job already decides this: it reads the profile, follows the +-- home page the profile names, and looks for a rel=me link pointing back. Two +-- sources agreeing is what lifts the person to 0.9 and turns the page's links +-- into their accounts. But the decision was only ever expressed as a +-- confidence, and 0.9 is also what a person crawled off a company page gets, +-- so nothing could later ask "did this person verify themselves" and get a +-- straight answer. +-- +-- The public directory needs that answer. It lists a person only when they +-- publish their own profile: an OpenProfile.md they serve themselves +-- (`published_url`), or a profile and a home page that point at each other. +-- Storing the bit keeps that rule a column test rather than an inference over +-- confidences that mean different things in different rows. +-- +-- Rows written before this column default to 0 and are re-decided the next +-- time the job runs for that person. + +ALTER TABLE openprofiles ADD COLUMN corroborated INTEGER NOT NULL DEFAULT 0; diff --git a/packages/pipeline/src/openprofile.ts b/packages/pipeline/src/openprofile.ts index 48a2038..3b92724 100644 --- a/packages/pipeline/src/openprofile.ts +++ b/packages/pipeline/src/openprofile.ts @@ -321,11 +321,19 @@ export async function runOpenProfileJob( const markdown = published?.markdown ?? buildOpenProfile({ ...merged, web: merged.web ?? web }); await db.execute({ - sql: `INSERT INTO openprofiles (person_id, markdown, sources_json, published_url, generated_at) - VALUES (?, ?, ?, ?, ?) + sql: `INSERT INTO openprofiles (person_id, markdown, sources_json, published_url, corroborated, generated_at) + VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(person_id) DO UPDATE SET markdown = excluded.markdown, sources_json = excluded.sources_json, - published_url = excluded.published_url, generated_at = excluded.generated_at`, - args: [personId, markdown, JSON.stringify(sources), published?.url ?? null, stamp], + published_url = excluded.published_url, corroborated = excluded.corroborated, + generated_at = excluded.generated_at`, + args: [ + personId, + markdown, + JSON.stringify(sources), + published?.url ?? null, + corroborated ? 1 : 0, + stamp, + ], }); // What the sources corroborate, kept where the rest of the product reads it.