From 8f66f557d5b6a1df9df5fcf1109388d16c9659ee Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 13 Sep 2026 13:22:52 +0000 Subject: [PATCH] Follow nichedb.dev for a campaign: its open collections become crawls, on a clock nichedb.dev's webrings, sites and profiles collections are lists of people's own sites, with who makes each one and what it publishes about itself. Anyone in there publishes on the open web in a machine-readable way, which is who an open-standards offering wants to talk to. The `discover_nichedb` job reads each collection since a cursor through nichedb's keyless items API (sort=updated, order=asc, since=), turns items into candidate sites (one per host; platforms, our own hosts and the l.ink pages skipped; a profile's site from its accounts or its OpenProfile), queues a `crawl_site` per new host under the campaign, and queues itself again with the newest updated_at as the next cursor, everyMs from now. Two alternating dedupe keys per campaign keep one next run outstanding while a run is still running. Needs no model. POST /campaigns/:id/discover/nichedb starts or re-tunes it (collections, everyHours, limit, since), GET reads the pending and running runs, DELETE stops it; a running job whose row was deleted under it finishes without a successor. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01XYae2mH3khdwiXUVzcVMDw --- apps/api/src/app.ts | 94 +++- apps/server/src/index.ts | 13 + apps/web/lib/activity.ts | 1 + packages/pipeline/src/index.ts | 10 + packages/pipeline/src/jobs.ts | 2 + .../pipeline/src/nichedb-discovery.test.ts | 245 +++++++++++ packages/pipeline/src/nichedb-discovery.ts | 401 ++++++++++++++++++ 7 files changed, 765 insertions(+), 1 deletion(-) create mode 100644 packages/pipeline/src/nichedb-discovery.test.ts create mode 100644 packages/pipeline/src/nichedb-discovery.ts diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 2b159a1..4d46428 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -150,7 +150,16 @@ import { type ShareNetwork, } from '@outreachgraph/domain'; import { draftForRecommendation, draftProfile, type TextModel } from '@outreachgraph/ai'; -import { batchStatus, enqueue, runPipeline } from '@outreachgraph/pipeline'; +import { + batchStatus, + enqueue, + nichedbDiscoveryStatus, + nichedbFirstDedupeKey, + NICHEDB_DEFAULT_COLLECTIONS, + NICHEDB_DEFAULT_EVERY_MS, + runPipeline, + stopNichedbDiscovery, +} from '@outreachgraph/pipeline'; import { GitHubProvider, SiteProvider, @@ -1053,6 +1062,89 @@ export function createApp(options: AppOptions): Hono { return c.json({ campaign }); }); + /** + * Follow nichedb.dev for a campaign: every few hours, read the open + * collections that are people's own sites (webring members, OpenSite + * records, profiles with a home page), queue a crawl per new site, and + * come back. One pending run per campaign; POST again to change the + * collections or the clock, DELETE to stop. + */ + api.get('/campaigns/:id/discover/nichedb', async (c) => { + const actor = c.get('actor'); + const db = c.get('db'); + const campaign = await repo.getCampaign(db, actor.workspaceId, c.req.param('id')); + if (!campaign) throw ApiError.notFound('campaign'); + const runs = await nichedbDiscoveryStatus(db, actor.workspaceId, campaign.id); + return c.json({ campaignId: campaign.id, following: runs.length > 0, runs }); + }); + + api.post('/campaigns/:id/discover/nichedb', async (c) => { + const actor = c.get('actor'); + const db = c.get('db'); + if (!canApprove(actor)) throw ApiError.forbidden('following nichedb'); + const campaign = await repo.getCampaign(db, actor.workspaceId, c.req.param('id')); + if (!campaign) throw ApiError.notFound('campaign'); + + const raw = safeJson(await c.req.raw.text()); + const collections = + Array.isArray(raw.collections) && raw.collections.length + ? raw.collections + .map((x: unknown) => + String(x) + .toLowerCase() + .replace(/[^a-z0-9-]/g, ''), + ) + .filter(Boolean) + : [...NICHEDB_DEFAULT_COLLECTIONS]; + const everyHours = Number(raw.everyHours); + const everyMs = + Number.isFinite(everyHours) && everyHours >= 0 + ? Math.round(everyHours * 3_600_000) + : NICHEDB_DEFAULT_EVERY_MS; + const limit = + Number.isFinite(Number(raw.limit)) && Number(raw.limit) > 0 + ? Math.min(200, Math.round(Number(raw.limit))) + : undefined; + const since = typeof raw.since === 'string' && raw.since ? raw.since : null; + + // A new request replaces the pending schedule rather than adding to it. + await stopNichedbDiscovery(db, actor.workspaceId, campaign.id); + const result = await enqueue(db, { + workspaceId: actor.workspaceId, + kind: 'discover_nichedb', + payload: { + campaignId: campaign.id, + collections, + since, + everyMs, + ...(limit ? { limit } : {}), + }, + dedupeKey: nichedbFirstDedupeKey(campaign.id), + }); + return c.json( + { + ok: true, + campaignId: campaign.id, + queued: result.queued, + jobId: result.id ?? null, + collections, + everyHours: everyMs / 3_600_000, + since, + }, + 202, + ); + }); + + api.delete('/campaigns/:id/discover/nichedb', async (c) => { + const actor = c.get('actor'); + const db = c.get('db'); + if (!canApprove(actor)) throw ApiError.forbidden('stopping nichedb discovery'); + const campaign = await repo.getCampaign(db, actor.workspaceId, c.req.param('id')); + if (!campaign) throw ApiError.notFound('campaign'); + const removed = await stopNichedbDiscovery(db, actor.workspaceId, campaign.id); + return c.json({ ok: true, campaignId: campaign.id, removed }); + }); + /** * The front door (PRD ยง8). * diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index fce5620..24cf4d5 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -54,6 +54,7 @@ import { runCadences, runCrawlJob, runDiscoveryJob, + runNichedbDiscoveryJob, runOpenProfileJob, loadImapCredentials, receiveReplies, @@ -572,6 +573,15 @@ async function discoverDomains(job: QueuedJob): Promise { ); } +/** nichedb.dev's open collections, read since a cursor, become crawls; needs no model. */ +async function discoverNichedb(job: QueuedJob): Promise { + const result = await runNichedbDiscoveryJob({ db }, job); + console.log( + `discover nichedb for ${result.campaignId}: read ${result.read}, ${result.candidates} sites, queued ${result.queued}` + + `${result.rescheduled ? ', next run queued' : ''}`, + ); +} + async function runJob(job: QueuedJob): Promise { switch (job.kind) { case 'crawl_site': @@ -580,6 +590,9 @@ async function runJob(job: QueuedJob): Promise { case 'discover_domains': await discoverDomains(job); return; + case 'discover_nichedb': + await discoverNichedb(job); + return; case 'rescore_prospect': { const { campaignId, personId } = job.payload as { campaignId?: string; personId?: string }; if (!campaignId || !personId) diff --git a/apps/web/lib/activity.ts b/apps/web/lib/activity.ts index 423d71e..3ac6695 100644 --- a/apps/web/lib/activity.ts +++ b/apps/web/lib/activity.ts @@ -16,6 +16,7 @@ import type { WorkflowStatusView } from './api'; const KIND_LABELS: Record = { crawl_site: 'reading sites', discover_domains: 'finding companies', + discover_nichedb: 'reading nichedb.dev for sites', rescore_prospect: 'rescoring', process_deletion: 'deleting', }; diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts index 59cb7cf..4277ee8 100644 --- a/packages/pipeline/src/index.ts +++ b/packages/pipeline/src/index.ts @@ -25,6 +25,16 @@ export { type RegenerateResult, } from './regenerate'; export { runDiscoveryJob, type DiscoveryJobDeps, type DiscoveryJobResult } from './discovery'; +export { + runNichedbDiscoveryJob, + nichedbDiscoveryStatus, + stopNichedbDiscovery, + firstDedupeKey as nichedbFirstDedupeKey, + DEFAULT_COLLECTIONS as NICHEDB_DEFAULT_COLLECTIONS, + DEFAULT_EVERY_MS as NICHEDB_DEFAULT_EVERY_MS, + type NichedbDiscoveryDeps, + type NichedbDiscoveryResult, +} from './nichedb-discovery'; export { runAutopilot, HoldLedger, diff --git a/packages/pipeline/src/jobs.ts b/packages/pipeline/src/jobs.ts index 4763271..f8c1e91 100644 --- a/packages/pipeline/src/jobs.ts +++ b/packages/pipeline/src/jobs.ts @@ -30,6 +30,8 @@ export const JOB_KINDS = [ 'crawl_site', /** Expand a keyword into real companies and queue a crawl for each. */ 'discover_domains', + /** Read nichedb.dev's open collections since a cursor, queue a crawl per new site, and queue itself again. */ + 'discover_nichedb', /** * Find the rest of an imported contact from their address alone. * diff --git a/packages/pipeline/src/nichedb-discovery.test.ts b/packages/pipeline/src/nichedb-discovery.test.ts new file mode 100644 index 0000000..8ac6d94 --- /dev/null +++ b/packages/pipeline/src/nichedb-discovery.test.ts @@ -0,0 +1,245 @@ +/** + * nichedb discovery: items in, crawls out, and the next run queued. + * + * Only the network is stubbed. The queue is the real one, so what the test + * asserts is the row a worker would claim next. + */ +import { afterEach, describe, expect, test } from 'bun:test'; +import { queryAll } from '@outreachgraph/db'; +import { seedDatabase, SEED, type SeededDatabase } from '../../../apps/api/src/test-seed'; +import { enqueue, claimNext } from './queue'; +import { + candidateUrl, + hostOf, + isSkippedHost, + nichedbDiscoveryStatus, + originOf, + runNichedbDiscoveryJob, + stopNichedbDiscovery, +} from './nichedb-discovery'; + +let seeded: SeededDatabase | undefined; + +afterEach(() => { + seeded?.cleanup(); + seeded = undefined; +}); + +const ITEMS: Record = { + webrings: [ + { + id: 1, + collection: 'webrings', + kind: 'ring', + url: 'https://rssamplifier.com/ring/small-web', + updated_at: '2026-09-13T10:00:00.000Z', + }, + { + id: 2, + collection: 'webrings', + kind: 'member', + url: 'https://chovy.com/', + updated_at: '2026-09-13T10:05:00.000Z', + data: { made_by: 'human' }, + }, + { + id: 3, + collection: 'webrings', + kind: 'member', + url: 'https://www.chovy.com/blog', + updated_at: '2026-09-13T10:06:00.000Z', + }, + { + id: 4, + collection: 'webrings', + kind: 'member', + url: 'https://github.com/profullstack', + updated_at: '2026-09-13T10:07:00.000Z', + }, + ], + sites: [ + { + id: 5, + collection: 'sites', + kind: 'page', + url: 'https://example-com.l.ink/', + updated_at: '2026-09-13T11:00:00.000Z', + }, + { + id: 6, + collection: 'sites', + kind: 'page', + url: 'https://david.weekly.org/about', + updated_at: '2026-09-13T11:30:00.000Z', + }, + ], + profiles: [ + { + id: 7, + collection: 'profiles', + kind: 'person', + url: 'https://nichedb.dev/c/profiles/ada-1', + updated_at: '2026-09-13T09:00:00.000Z', + data: { + accounts: [ + { network: 'github', url: 'https://github.com/ada' }, + { network: 'website', url: 'https://ada.example/' }, + ], + }, + }, + { + id: 8, + collection: 'profiles', + kind: 'person', + url: 'https://nichedb.dev/c/profiles/nobody-2', + updated_at: '2026-09-13T09:30:00.000Z', + data: {}, + }, + ], +}; + +function stubNichedb(calls: string[]) { + return async (url: string) => { + calls.push(url); + const parsed = new URL(url); + const collection = parsed.searchParams.get('collection') ?? ''; + const since = parsed.searchParams.get('since'); + const items = (ITEMS[collection] ?? []).filter( + (i) => !since || String((i as { updated_at: string }).updated_at) > since, + ); + return { count: items.length, items }; + }; +} + +describe('candidate urls', () => { + test('a webring member is its site; a ring is not; a profile is its website or nothing', () => { + expect(candidateUrl(ITEMS.webrings![0] as never)).toBeNull(); + expect(candidateUrl(ITEMS.webrings![1] as never)).toBe('https://chovy.com/'); + expect(candidateUrl(ITEMS.profiles![0] as never)).toBe('https://ada.example/'); + expect(candidateUrl(ITEMS.profiles![1] as never)).toBeNull(); + expect(candidateUrl(ITEMS.sites![1] as never)).toBe('https://david.weekly.org/about'); + }); + + test('hosts drop www, refuse junk, and skip platforms and our own', () => { + expect(hostOf('https://www.chovy.com/blog')).toBe('chovy.com'); + expect(hostOf('mailto:a@b.c')).toBeNull(); + expect(hostOf('https://localhost/')).toBeNull(); + expect(originOf('https://www.chovy.com/blog?x=1')).toBe('https://chovy.com'); + expect(isSkippedHost('github.com')).toBe(true); + expect(isSkippedHost('gist.github.com')).toBe(true); + expect(isSkippedHost('nichedb.dev')).toBe(true); + expect(isSkippedHost('example-com.l.ink')).toBe(true); + expect(isSkippedHost('david.weekly.org')).toBe(false); + }); +}); + +describe('discover_nichedb', () => { + test('reads each collection, queues one crawl per new site, and queues itself again', async () => { + seeded = await seedDatabase('nichedb-discovery'); + const { db } = seeded; + const calls: string[] = []; + + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'discover_nichedb', + payload: { campaignId: SEED.campaignId, everyMs: 3_600_000 }, + }); + const job = await claimNext(db, SEED.workspaceId); + expect(job?.kind).toBe('discover_nichedb'); + + const result = await runNichedbDiscoveryJob({ db, fetchJson: stubNichedb(calls) }, job!); + + expect(result.read).toBe(8); + // chovy.com once (www and path folded), david.weekly.org, ada.example. + // github.com and the l.ink page are skipped; the ring and the empty + // profile are not sites. + expect(result.candidates).toBe(3); + expect(result.queued).toBe(3); + expect(result.next).toBe('2026-09-13T11:30:00.000Z'); + expect(result.rescheduled).toBe(true); + expect(calls.length).toBe(3); + + const crawls = await queryAll<{ payload_json: string; dedupe_key: string; batch_id: string }>( + db, + `SELECT payload_json, dedupe_key, batch_id FROM jobs WHERE kind = 'crawl_site' ORDER BY dedupe_key`, + ); + expect(crawls.map((c) => c.dedupe_key)).toEqual([ + `crawl:${SEED.campaignId}:ada.example`, + `crawl:${SEED.campaignId}:chovy.com`, + `crawl:${SEED.campaignId}:david.weekly.org`, + ]); + expect(crawls.every((c) => c.batch_id === job!.id)).toBe(true); + expect(JSON.parse(crawls[1]!.payload_json)).toEqual({ + url: 'https://chovy.com', + campaignId: SEED.campaignId, + }); + + const pending = await nichedbDiscoveryStatus(db, SEED.workspaceId, SEED.campaignId); + expect(pending.length).toBe(2); + const next = pending.find((p) => p.status === 'pending'); + expect(next?.since).toBe('2026-09-13T11:30:00.000Z'); + }); + + test('the next run reads only what changed, and a second run cannot double the next', async () => { + seeded = await seedDatabase('nichedb-discovery-next'); + const { db } = seeded; + const calls: string[] = []; + + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'discover_nichedb', + payload: { + campaignId: SEED.campaignId, + collections: ['sites'], + since: '2026-09-13T11:00:00.000Z', + everyMs: 3_600_000, + }, + }); + const job = await claimNext(db, SEED.workspaceId); + const result = await runNichedbDiscoveryJob({ db, fetchJson: stubNichedb(calls) }, job!); + expect(result.read).toBe(1); + expect(result.queued).toBe(1); + expect(calls[0]).toContain('since=2026-09-13T11%3A00%3A00.000Z'); + + // Running it again while the next is still queued does not queue a third. + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'discover_nichedb', + payload: { campaignId: SEED.campaignId, collections: ['sites'], everyMs: 3_600_000 }, + dedupeKey: 'test:second', + }); + // The crawl the first run queued is older and claims first; skip past it. + let again = await claimNext(db, SEED.workspaceId); + while (again && again.kind !== 'discover_nichedb') + again = await claimNext(db, SEED.workspaceId); + expect(again?.kind).toBe('discover_nichedb'); + const second = await runNichedbDiscoveryJob({ db, fetchJson: stubNichedb(calls) }, again!); + expect(second.rescheduled).toBe(false); + }); + + test('stopping removes the pending run and a stopped job does not requeue', async () => { + seeded = await seedDatabase('nichedb-discovery-stop'); + const { db } = seeded; + + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'discover_nichedb', + payload: { campaignId: SEED.campaignId, collections: ['sites'], everyMs: 3_600_000 }, + }); + expect(await stopNichedbDiscovery(db, SEED.workspaceId, SEED.campaignId)).toBe(1); + expect(await nichedbDiscoveryStatus(db, SEED.workspaceId, SEED.campaignId)).toEqual([]); + expect(await claimNext(db, SEED.workspaceId)).toBeUndefined(); + + // A run whose own row was deleted under it finishes without a successor. + await enqueue(db, { + workspaceId: SEED.workspaceId, + kind: 'discover_nichedb', + payload: { campaignId: SEED.campaignId, collections: ['sites'], everyMs: 3_600_000 }, + }); + const job = await claimNext(db, SEED.workspaceId); + await db.execute({ sql: `DELETE FROM jobs WHERE id = ?`, args: [job!.id] }); + const result = await runNichedbDiscoveryJob({ db, fetchJson: stubNichedb([]) }, job!); + expect(result.rescheduled).toBe(false); + expect(await nichedbDiscoveryStatus(db, SEED.workspaceId, SEED.campaignId)).toEqual([]); + }); +}); diff --git a/packages/pipeline/src/nichedb-discovery.ts b/packages/pipeline/src/nichedb-discovery.ts new file mode 100644 index 0000000..0f379b8 --- /dev/null +++ b/packages/pipeline/src/nichedb-discovery.ts @@ -0,0 +1,401 @@ +/** + * The `discover_nichedb` job: nichedb.dev's open items become a queue of + * sites to read, on a clock. + * + * nichedb.dev is Profullstack's open data directory. Some of its collections + * are lists of people's own sites: webring members (with who makes each + * site), sites that publish an OpenSite record, and profiles with a home page. + * Anyone in there is somebody who publishes on the open web and says so in a + * machine-readable way, which is exactly who an open-standards offering wants + * to talk to. Its items API is keyless and paged by `since`, so a campaign can + * follow it the way a feed reader follows a feed: read what changed since last + * time, queue the new sites, come back later. + * + * Each run reads each collection since the cursor it was handed, turns items + * into candidate site URLs (one per host, platforms and our own hosts + * skipped), queues a `crawl_site` per new host under this campaign, and then + * queues itself again with the newest `updated_at` it saw as the next cursor, + * `everyMs` from now. The crawl does the rest: the site names its person, the + * pipeline scores them against the offering, and a card appears or does not. + * + * A job, not a tick sweep, so a workspace that never asked for it costs + * nothing, and so stopping it is deleting one pending row. + */ +import { type Client, queryAll } from '@outreachgraph/db'; +import { emitEvent } from './events'; +import { enqueue, type QueuedJob } from './queue'; + +export const NICHEDB_URL = 'https://nichedb.dev'; + +/** Collections whose items are people's own sites. */ +export const DEFAULT_COLLECTIONS: readonly string[] = ['webrings', 'sites', 'profiles']; + +/** Six hours: nichedb re-reads its sources hourly, and a site is not urgent. */ +export const DEFAULT_EVERY_MS = 6 * 60 * 60 * 1000; + +/** Sites queued per run, so one run cannot flood the crawl queue. */ +export const DEFAULT_LIMIT = 40; + +const PAGE = 200; +const MAX_PAGES_PER_COLLECTION = 5; + +/** + * Hosts that are never one person's own site: platforms whose crawl would + * name the platform, and the hosts this company runs, which name us. + */ +const SKIP_HOSTS = new Set([ + 'github.com', + 'gitlab.com', + 'twitter.com', + 'x.com', + 'linkedin.com', + 'facebook.com', + 'instagram.com', + 'youtube.com', + 'medium.com', + 'substack.com', + 'reddit.com', + 'wikipedia.org', + 'nichedb.dev', + 'rssamplifier.com', + 'profullstack.com', + 'logicsrc.com', + 'outreachgraph.com', + 'goviral.wiki', +]); + +export interface NichedbItem { + readonly id?: string | number; + readonly collection?: string; + readonly kind?: string; + readonly url?: string; + readonly updated_at?: string; + readonly data?: Record | null; +} + +export interface NichedbDiscoveryDeps { + readonly db: Client; + /** The network, for a test to stub. Defaults to fetch of nichedb's items API. */ + readonly fetchJson?: (url: string) => Promise; + /** Cap on sites queued per run. */ + readonly limit?: number; +} + +export interface NichedbDiscoveryResult { + readonly campaignId: string; + readonly collections: readonly string[]; + readonly read: number; + readonly candidates: number; + readonly queued: number; + readonly since: string | null; + readonly next: string | null; + readonly rescheduled: boolean; +} + +/** The host a URL is on, without a leading www, or null for anything odd. */ +export function hostOf(url: string): string | null { + try { + const parsed = new URL(url); + if (parsed.protocol !== 'https:' && parsed.protocol !== 'http:') return null; + const host = parsed.hostname.toLowerCase().replace(/^www\./, ''); + if (!/^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)+$/.test(host)) + return null; + return host; + } catch { + return null; + } +} + +/** Whether a host is a platform or one of ours, and so never a candidate. */ +export function isSkippedHost(host: string): boolean { + if (SKIP_HOSTS.has(host)) return true; + for (const skip of SKIP_HOSTS) if (host.endsWith(`.${skip}`)) return true; + // l.ink is the link shortener behind the sites collection's pages. + return host.endsWith('.l.ink'); +} + +/** + * The site an item is about, or null when the item is not about a site. + * + * A webring member's url is the member site. A site item's url is the page + * on the site. A profile's url is nichedb's own page about the person, so the + * site has to come from the record: the accounts it lists, or the OpenProfile + * it was read from. + */ +export function candidateUrl(item: NichedbItem): string | null { + const data = item.data ?? {}; + if (item.collection === 'profiles') { + const accounts = Array.isArray(data.accounts) + ? (data.accounts as Record[]) + : []; + for (const account of accounts) { + const network = String(account.network ?? account.kind ?? '').toLowerCase(); + const url = + typeof account.url === 'string' + ? account.url + : typeof account.value === 'string' + ? account.value + : ''; + if ( + (network === 'website' || + network === 'site' || + network === 'web' || + network === 'homepage') && + url + ) + return url; + } + const openprofile = data.openprofile; + if (typeof openprofile === 'string') return openprofile; + if ( + openprofile && + typeof openprofile === 'object' && + typeof (openprofile as Record).url === 'string' + ) { + return String((openprofile as Record).url); + } + return null; + } + if (item.collection === 'webrings' && item.kind !== 'member') return null; + return typeof item.url === 'string' ? item.url : null; +} + +/** The origin of a URL, which is what the crawl wants: the site, not the page. */ +export function originOf(url: string): string | null { + const host = hostOf(url); + if (!host) return null; + try { + return `${new URL(url).protocol}//${host}`; + } catch { + return null; + } +} + +/** The dedupe key for a campaign's first run, which the API uses to start it. */ +export function firstDedupeKey(campaignId: string): string { + return `nichedb:${campaignId}:a`; +} + +/** The other of the two keys, so a running job can queue its successor. */ +export function nextDedupeKey(campaignId: string, current: string | null): string { + return current === `nichedb:${campaignId}:a` + ? `nichedb:${campaignId}:b` + : `nichedb:${campaignId}:a`; +} + +async function defaultFetchJson(url: string): Promise { + const response = await fetch(url, { + headers: { + accept: 'application/json', + 'user-agent': 'outreachgraph-nichedb-discovery/1 (+https://outreachgraph.com)', + }, + signal: AbortSignal.timeout(30_000), + }); + if (!response.ok) throw new Error(`nichedb answered ${response.status} for ${url}`); + return response.json(); +} + +/** + * One collection since a cursor, oldest first, a few pages at most. + * + * @returns the items and the newest `updated_at` among them + */ +async function readCollection( + fetchJson: (url: string) => Promise, + collection: string, + since: string | null, +): Promise<{ items: NichedbItem[]; newest: string | null }> { + const items: NichedbItem[] = []; + let newest: string | null = since; + let cursor = since; + for (let page = 0; page < MAX_PAGES_PER_COLLECTION; page += 1) { + const params = new URLSearchParams({ + collection, + sort: 'updated', + order: 'asc', + limit: String(PAGE), + }); + if (cursor) params.set('since', cursor); + const body = (await fetchJson(`${NICHEDB_URL}/api/v1/items?${params}`)) as { + items?: NichedbItem[]; + } | null; + const got = Array.isArray(body?.items) ? body.items : []; + if (got.length === 0) break; + for (const item of got) { + items.push(item); + if (item.updated_at && (!newest || item.updated_at > newest)) newest = item.updated_at; + } + const last = got[got.length - 1]?.updated_at ?? null; + // No progress means the page is one timestamp wide; stop rather than loop. + if (!last || last === cursor || got.length < PAGE) break; + cursor = last; + } + return { items, newest }; +} + +/** + * Where a campaign's discovery stands: the pending or running job, if any. + */ +export async function nichedbDiscoveryStatus( + db: Client, + workspaceId: string, + campaignId: string, +): Promise<{ id: string; status: string; runAfter: string; since: string | null }[]> { + const rows = await queryAll<{ + id: string; + status: string; + run_after: string; + payload_json: string; + }>( + db, + `SELECT id, status, run_after, payload_json FROM jobs + WHERE workspace_id = ? AND kind = 'discover_nichedb' AND status IN ('pending', 'running') + AND payload_json LIKE ? + ORDER BY run_after ASC`, + [workspaceId, `%"campaignId":"${campaignId}"%`], + ); + return rows.map((r) => { + let since: string | null = null; + try { + const payload = JSON.parse(r.payload_json) as { since?: unknown }; + since = typeof payload.since === 'string' ? payload.since : null; + } catch { + since = null; + } + return { id: r.id, status: r.status, runAfter: r.run_after, since }; + }); +} + +/** + * Stop following nichedb for a campaign: the pending run goes; a running one + * finishes and, finding itself stopped, does not queue the next. + * + * @returns how many pending runs were removed + */ +export async function stopNichedbDiscovery( + db: Client, + workspaceId: string, + campaignId: string, +): Promise { + const result = await db.execute({ + sql: `DELETE FROM jobs + WHERE workspace_id = ? AND kind = 'discover_nichedb' AND status = 'pending' + AND payload_json LIKE ?`, + args: [workspaceId, `%"campaignId":"${campaignId}"%`], + }); + return Number(result.rowsAffected ?? 0); +} + +export async function runNichedbDiscoveryJob( + deps: NichedbDiscoveryDeps, + job: QueuedJob, +): Promise { + const payload = job.payload as { + campaignId?: string; + collections?: unknown; + since?: unknown; + everyMs?: unknown; + limit?: unknown; + }; + const campaignId = typeof payload.campaignId === 'string' ? payload.campaignId : ''; + if (!campaignId) throw new Error('discover_nichedb needs a campaignId'); + const collections = + Array.isArray(payload.collections) && payload.collections.length + ? payload.collections.map(String) + : [...DEFAULT_COLLECTIONS]; + const since = typeof payload.since === 'string' && payload.since ? payload.since : null; + const everyMs = + typeof payload.everyMs === 'number' && payload.everyMs >= 0 + ? payload.everyMs + : DEFAULT_EVERY_MS; + const limit = Math.max(1, Number(payload.limit ?? deps.limit ?? DEFAULT_LIMIT) || DEFAULT_LIMIT); + const fetchJson = deps.fetchJson ?? defaultFetchJson; + + await emitEvent(deps.db, { + workspaceId: job.workspaceId, + campaignId, + phase: 'discover', + message: `Reading nichedb.dev ${collections.join(', ')}${since ? ` since ${since}` : ''}`, + detail: { collections, since }, + }); + + let read = 0; + let next: string | null = since; + const hosts = new Map(); + for (const collection of collections) { + const { items, newest } = await readCollection(fetchJson, collection, since); + read += items.length; + if (newest && (!next || newest > next)) next = newest; + for (const item of items) { + const url = candidateUrl({ ...item, collection: item.collection ?? collection }); + if (!url) continue; + const host = hostOf(url); + const origin = originOf(url); + if (!host || !origin || isSkippedHost(host) || hosts.has(host)) continue; + hosts.set(host, origin); + } + } + + let queued = 0; + for (const [host, origin] of hosts) { + if (queued >= limit) break; + const result = await enqueue(deps.db, { + workspaceId: job.workspaceId, + kind: 'crawl_site', + payload: { url: origin, campaignId }, + // Scored per campaign, so the same site in two campaigns is two crawls; + // the same site twice in one campaign's queue is not. + dedupeKey: `crawl:${campaignId}:${host}`, + batchId: job.id, + }); + if (result.queued) queued += 1; + } + + // The next run. Not while the campaign has been stopped under us: a stop + // deletes the pending row, and a running job that then re-queued itself + // would be a daemon nobody can turn off. The running row is this job, so + // "stopped" is "this job's own row is gone". + let rescheduled = false; + if (everyMs > 0) { + const still = await queryAll<{ id: string; dedupe_key: string | null }>( + deps.db, + `SELECT id, dedupe_key FROM jobs WHERE id = ?`, + [job.id], + ); + if (still.length > 0) { + const result = await enqueue(deps.db, { + workspaceId: job.workspaceId, + kind: 'discover_nichedb', + payload: { ...payload, campaignId, collections, since: next, everyMs, limit }, + delayMs: everyMs, + // One outstanding next run per campaign. The dedupe index covers + // running jobs too, and this job is still running under its own key, + // so the next run takes the other of two keys: a run under `a` + // queues `b`, a run under `b` queues `a`, and a second run started by + // hand while a next is already pending finds the key taken. + dedupeKey: nextDedupeKey(campaignId, still[0]?.dedupe_key ?? null), + }); + rescheduled = result.queued; + } + } + + await emitEvent(deps.db, { + workspaceId: job.workspaceId, + campaignId, + phase: 'discover', + level: queued > 0 ? 'success' : 'info', + message: `nichedb.dev: read ${read} items, ${hosts.size} sites, queued ${queued} to read${rescheduled ? `, again in ${Math.round(everyMs / 3_600_000)}h` : ''}`, + detail: { read, candidates: hosts.size, queued, since, next, rescheduled }, + }); + + return { + campaignId, + collections, + read, + candidates: hosts.size, + queued, + since, + next, + rescheduled, + }; +}