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, + }; +}