Follow nichedb.dev for a campaign: its open collections become crawls, on a clock

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<AppEnv> {
     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).
    *

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<void> {
   );
 }
 
+/** nichedb.dev's open collections, read since a cursor, become crawls; needs no model. */
+async function discoverNichedb(job: QueuedJob): Promise<void> {
+  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<void> {
   switch (job.kind) {
     case 'crawl_site':
@@ -580,6 +590,9 @@ async function runJob(job: QueuedJob): Promise<void> {
     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)

apps/web/lib/activity.ts

@@ -16,6 +16,7 @@ import type { WorkflowStatusView } from './api';
 const KIND_LABELS: Record<string, string> = {
   crawl_site: 'reading sites',
   discover_domains: 'finding companies',
+  discover_nichedb: 'reading nichedb.dev for sites',
   rescore_prospect: 'rescoring',
   process_deletion: 'deleting',
 };

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,

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.
    *