pitbot: local FTS5 search mirror so our index makes the cut (0.13.1)

.gitignore

@@ -1,3 +1,4 @@
 node_modules
 .env*
 *.log
+data/

deploy/crawler/docker-compose.yml

@@ -48,9 +48,12 @@ services:
       obscura-1: { condition: service_started }
       obscura-2: { condition: service_started }
     ports: ['127.0.0.1:8807:8797']
+    # The local search mirror pit.world reads (/api/v1/search on 8807).
+    volumes: ['index:/data']
     mem_limit: 1536m
     restart: unless-stopped
     stop_grace_period: 30s
 
 volumes:
   redis:
+  index:

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pitworld",
-  "version": "0.13.0",
+  "version": "0.13.1",
   "mcpName": "world.pit/pit",
   "description": "pit.world MCP server: private web search and Moshpit name lookups for AI agents (stdio, no API key). Also a web app, API, CLI and TUI.",
   "license": "MIT",

services/crawler/Dockerfile

@@ -8,7 +8,8 @@ RUN bun install --frozen-lockfile --production
 COPY src ./src
 COPY services/crawler ./services/crawler
 # Dev boxes check out with umask 007; the runtime user must still read the code.
-RUN chmod -R a+rX /app
+RUN chmod -R a+rX /app && mkdir -p /data && chown bun /data
+ENV CRAWL_INDEX_PATH=/data/index.db
 ENV NODE_ENV=production CRAWL_HEALTH_PORT=8797
 USER bun
 # A blocked event loop or a dead Redis shows as unhealthy (watchdog-autoheal restarts it).

services/crawler/src/cli.ts

@@ -49,7 +49,15 @@ async function worker() {
         hostname: config.healthHost,
         port: config.healthPort,
         async fetch(request) {
-          const path = new URL(request.url).pathname
+          const url = new URL(request.url)
+          const path = url.pathname
+          if (path === '/api/v1/search') {
+            // The same shape as nichedb's search, so pit.world reads either.
+            const q = (url.searchParams.get('q') ?? '').slice(0, 512)
+            const limit = Number(url.searchParams.get('limit')) || 10
+            const items = q ? runtime.index.search(q, limit) : []
+            return Response.json({ q, count: items.length, items })
+          }
           if (path !== '/healthz' && path !== '/stats') return new Response('not found', { status: 404 })
           try {
             // /healthz is the cheap one (Docker asks every 30 s); /stats is what pit.world/status reads.

services/crawler/src/config.ts

@@ -41,6 +41,8 @@ export type Config = {
   roles: Role[]
   healthPort: number
   healthHost: string
+  /** The local search mirror (SQLite FTS5) the publish role writes and /api/v1/search reads. */
+  indexPath: string
   userAgent: string
 }
 
@@ -76,6 +78,7 @@ export function configFromEnv(env: Record<string, string | undefined> = process.
     healthPort: num(env.CRAWL_HEALTH_PORT, 8797, 0),
     // Loopback on a host; 0.0.0.0 inside a container, whose port is published on the host's loopback.
     healthHost: env.CRAWL_HEALTH_HOST ?? '127.0.0.1',
+    indexPath: env.CRAWL_INDEX_PATH ?? './data/index.db',
     userAgent: env.CRAWL_USER_AGENT ?? USER_AGENT,
   }
 }

services/crawler/src/crawler.ts

@@ -29,6 +29,8 @@ export type Deps = {
   resolver?: Resolver
   renderer?: { size: number; render(url: string): Promise<Rendered> } | null
   log?: (line: string) => void
+  /** The local search mirror; every published batch lands here first. */
+  index?: { add(entries: string[]): number } | null
 }
 
 export type PageOutcome = { delayMs?: number; outcome: string }
@@ -207,16 +209,26 @@ export function createCrawler(deps: Deps) {
     return { outcome: await accept(doc, { ...job, rendered: true }, robots) }
   }
 
-  /** Records to nichedb, 200 a call, until the outbox is empty or 20 calls are made. */
+  /**
+   * Records into the local search mirror, then to nichedb, 200 a call, until the
+   * outbox is empty or 20 calls are made. A failed call puts the batch back; the
+   * mirror is an upsert, so indexing it again is harmless. Without a nichedb key
+   * the records go to the mirror only.
+   */
   async function publish(): Promise<{ sent: number; added: number; rejected: number; waiting: boolean }> {
-    if (!config.nichedbKey) return { sent: 0, added: 0, rejected: 0, waiting: true }
+    if (!config.nichedbKey && !deps.index) return { sent: 0, added: 0, rejected: 0, waiting: true }
     const fetchImpl = deps.fetch ?? fetch
     let sent = 0
     let added = 0
     let rejected = 0
     for (let round = 0; round < 20; round++) {
       const raw = await frontier.take(BATCH)
       if (!raw.length) break
+      if (deps.index) {
+        const kept = deps.index.add(raw)
+        if (kept) await frontier.bump('indexed_local', kept)
+      }
+      if (!config.nichedbKey) continue
       let response: Response
       try {
         response = await fetchImpl(`${config.nichedbUrl}/api/v1/sites/records`, {
@@ -238,6 +250,7 @@ export function createCrawler(deps: Deps) {
       added += body.added ?? 0
       rejected += body.rejected?.length ?? 0
     }
+    if (!config.nichedbKey) return { sent: 0, added: 0, rejected: 0, waiting: false }
     if (sent) {
       await frontier.bump('published', sent)
       await frontier.bump('published_new', added)

services/crawler/src/index-store.ts

@@ -0,0 +1,146 @@
+/**
+ * The crawler's local search mirror: every record it publishes, in SQLite FTS5.
+ *
+ * nichedb.dev/c/sites is where the index is published and kept, open to anyone.
+ * But a full-text match there runs against the GIN index of nichedb's whole
+ * items table (millions of rows across every collection), about a second for
+ * a common word and several for three words, and pit.world gives a slow engine
+ * 2.5 s. So the crawler also keeps what it publishes here and answers the same
+ * `/api/v1/search` shape on loopback in milliseconds; pit.world reads this.
+ *
+ * Written by the publish role (one writer), read by the health server. A
+ * record is keyed by canonical address, so a re-crawl replaces it.
+ */
+import { Database } from 'bun:sqlite'
+import { mkdirSync } from 'node:fs'
+import { dirname } from 'node:path'
+
+export type IndexedRecord = {
+  record: { url: string; canonical: string; title: string; description: string | null; kind: string; language: string | null; fetched_at: string; site?: { name?: string } }
+  enrichment: { keywords?: string[]; excerpt?: string; language?: string | null }
+}
+
+export type IndexHit = { url: string; title: string; summary: string; site: string; kind: string; fetched_at: string; score: number }
+
+/**
+ * User words as an FTS5 query that cannot fail: each token quoted as a literal.
+ * Raw input is a query language (a hyphen is a column filter, a dot a syntax
+ * error), and a search box must never 500 on what someone typed.
+ */
+export function toFtsTerms(input: string): string[] {
+  return String(input)
+    .toLowerCase()
+    .split(/[^\p{L}\p{N}_.+#]+/u)
+    .map((t) => t.replace(/^[.+#]+|[.]+$/g, ''))
+    .filter(Boolean)
+    .slice(0, 32)
+    .map((t) => `"${t.replace(/"/g, '""')}"`)
+}
+
+export class IndexStore {
+  private db: Database
+
+  constructor(path: string) {
+    if (path !== ':memory:') mkdirSync(dirname(path), { recursive: true })
+    this.db = new Database(path, { create: true })
+    this.db.exec('PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL; PRAGMA busy_timeout = 5000;')
+    this.db.exec(`
+      CREATE TABLE IF NOT EXISTS pages (
+        id INTEGER PRIMARY KEY,
+        url TEXT NOT NULL UNIQUE,
+        site TEXT NOT NULL,
+        kind TEXT NOT NULL,
+        title TEXT NOT NULL,
+        description TEXT NOT NULL DEFAULT '',
+        keywords TEXT NOT NULL DEFAULT '',
+        body TEXT NOT NULL DEFAULT '',
+        language TEXT,
+        fetched_at TEXT NOT NULL
+      );
+      CREATE VIRTUAL TABLE IF NOT EXISTS pages_fts USING fts5(
+        title, description, keywords, body, site,
+        content='pages', content_rowid='id', tokenize='unicode61 remove_diacritics 2'
+      );
+      CREATE TRIGGER IF NOT EXISTS pages_ai AFTER INSERT ON pages BEGIN
+        INSERT INTO pages_fts(rowid, title, description, keywords, body, site) VALUES (new.id, new.title, new.description, new.keywords, new.body, new.site);
+      END;
+      CREATE TRIGGER IF NOT EXISTS pages_ad AFTER DELETE ON pages BEGIN
+        INSERT INTO pages_fts(pages_fts, rowid, title, description, keywords, body, site) VALUES ('delete', old.id, old.title, old.description, old.keywords, old.body, old.site);
+      END;
+      CREATE TRIGGER IF NOT EXISTS pages_au AFTER UPDATE ON pages BEGIN
+        INSERT INTO pages_fts(pages_fts, rowid, title, description, keywords, body, site) VALUES ('delete', old.id, old.title, old.description, old.keywords, old.body, old.site);
+        INSERT INTO pages_fts(rowid, title, description, keywords, body, site) VALUES (new.id, new.title, new.description, new.keywords, new.body, new.site);
+      END;
+    `)
+  }
+
+  /** Keep a batch (one transaction). Entries are the outbox's JSON lines or objects. */
+  add(entries: Array<IndexedRecord | string>): number {
+    const upsert = this.db.prepare(`
+      INSERT INTO pages (url, site, kind, title, description, keywords, body, language, fetched_at)
+      VALUES ($url, $site, $kind, $title, $description, $keywords, $body, $language, $fetched)
+      ON CONFLICT(url) DO UPDATE SET site = excluded.site, kind = excluded.kind, title = excluded.title,
+        description = excluded.description, keywords = excluded.keywords, body = excluded.body,
+        language = excluded.language, fetched_at = excluded.fetched_at
+    `)
+    let kept = 0
+    this.db.transaction(() => {
+      for (const raw of entries) {
+        let entry: IndexedRecord
+        try {
+          entry = typeof raw === 'string' ? (JSON.parse(raw) as IndexedRecord) : raw
+        } catch {
+          continue
+        }
+        const r = entry?.record
+        if (!r?.canonical) continue
+        let site = ''
+        try {
+          site = new URL(r.canonical).hostname.replace(/^www\./, '')
+        } catch {
+          continue
+        }
+        upsert.run({
+          $url: r.canonical,
+          $site: site,
+          $kind: r.kind ?? 'page',
+          $title: r.title ?? site,
+          $description: r.description ?? '',
+          $keywords: (entry.enrichment?.keywords ?? []).join(' '),
+          $body: entry.enrichment?.excerpt ?? '',
+          $language: entry.enrichment?.language ?? r.language ?? null,
+          $fetched: r.fetched_at ?? new Date().toISOString(),
+        })
+        kept++
+      }
+    })()
+    return kept
+  }
+
+  /** Best matches: every word if any page has them all, else any word. Title weighs most. */
+  search(query: string, limit = 10): IndexHit[] {
+    const terms = toFtsTerms(query)
+    if (!terms.length) return []
+    const run = (match: string) =>
+      this.db
+        .query(
+          `SELECT p.url, p.title, p.site, p.kind, p.fetched_at,
+                  CASE WHEN p.description != '' THEN p.description ELSE substr(p.body, 1, 300) END AS summary,
+                  bm25(pages_fts, 10.0, 4.0, 6.0, 1.0, 3.0) AS score
+           FROM pages_fts JOIN pages p ON p.id = pages_fts.rowid
+           WHERE pages_fts MATCH ?
+           ORDER BY score LIMIT ?`,
+        )
+        .all(match, Math.min(Math.max(1, limit), 50)) as IndexHit[]
+    const all = run(terms.join(' AND '))
+    return all.length || terms.length === 1 ? all : run(terms.join(' OR '))
+  }
+
+  count(): number {
+    return (this.db.query('SELECT count(*) AS n FROM pages').get() as { n: number }).n
+  }
+
+  close(): void {
+    this.db.close()
+  }
+}

services/crawler/src/queues.ts

@@ -21,6 +21,7 @@ import { Redis } from 'ioredis'
 import type { Config, Role } from './config.ts'
 import { createCrawler, type Crawler, type PageJob, type SiteJob } from './crawler.ts'
 import { RedisFrontier } from './frontier.ts'
+import { IndexStore } from './index-store.ts'
 import { ObscuraPool } from './obscura.ts'
 import { urlId } from './url.ts'
 
@@ -48,6 +49,7 @@ export type Runtime = {
   queues: Record<keyof typeof QUEUES, Queue>
   frontier: RedisFrontier
   redis: Redis
+  index: IndexStore
   close(): Promise<void>
 }
 
@@ -63,10 +65,12 @@ export function createRuntime(config: Config, log: (line: string) => void = cons
   }
   const frontier = new RedisFrontier(redis, config.maxPages, config.recrawlDays * 86_400_000)
   const pool = config.obscuraUrls.length ? new ObscuraPool(config.obscuraUrls, config.obscuraToken) : null
+  const index = new IndexStore(config.indexPath)
   const crawler = createCrawler({
     config,
     frontier,
     renderer: pool,
+    index,
     log,
     enqueue: {
       async sites(jobs: SiteJob[]) {
@@ -87,9 +91,11 @@ export function createRuntime(config: Config, log: (line: string) => void = cons
     queues,
     frontier,
     redis,
+    index,
     async close() {
       await Promise.all(Object.values(queues).map((q) => q.close()))
       redis.disconnect()
+      index.close()
     },
   }
 }
@@ -196,7 +202,8 @@ export async function snapshot(runtime: Runtime) {
   ])
   const queues: Record<string, Record<string, number>> = {}
   Object.keys(runtime.queues).forEach((k, i) => (queues[k] = counts[i]!))
-  return { stats, queues }
+  const withIndex: Record<string, number> = { ...stats, index_pages: runtime.index.count() }
+  return { stats: withIndex, queues }
 }
 
 /** Everything /status shows: the snapshot, rates over the hour and the day, the log tail. */

services/crawler/test/index-store.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, test } from 'bun:test'
+import { configFromEnv } from '../src/config.ts'
+import { createCrawler } from '../src/crawler.ts'
+import { MemoryFrontier } from '../src/frontier.ts'
+import { IndexStore, toFtsTerms } from '../src/index-store.ts'
+
+const entry = (url: string, title: string, body: string, keywords: string[] = []) => ({
+  record: { url, canonical: url, title, description: null, kind: 'page', language: 'en', fetched_at: '2026-11-02T00:00:00.000Z' },
+  enrichment: { keywords, excerpt: body },
+})
+
+describe('the local search mirror', () => {
+  test('whatever someone types is a valid query', () => {
+    const store = new IndexStore(':memory:')
+    store.add([entry('https://a.example/', 'Node.js catch-all errors', 'TypeError: x is not a function in C# and node.js 1.3.14')])
+    for (const q of ['catch-all', 'TypeError: x', 'foo(bar', 'node.js', '1.3.14', 'C#', 'why "this fails', '***', '-'])
+      expect(() => store.search(q)).not.toThrow()
+    expect(store.search('node.js')).toHaveLength(1)
+    expect(toFtsTerms('a "b" c-d')).toEqual(['"a"', '"b"', '"c"', '"d"'])
+  })
+
+  test('all words first, any word when nothing has them all; title outranks body', () => {
+    const store = new IndexStore(':memory:')
+    store.add([
+      entry('https://ovens.example/pizza', 'Wood-fired pizza ovens', 'how to build a brick oven', ['pizza', 'oven']),
+      entry('https://blog.example/x', 'A post about bread', 'we once ate pizza near an oven'),
+      entry('https://other.example/', 'Gardening', 'tomatoes'),
+    ])
+    expect(store.search('pizza oven').map((h) => h.url)).toEqual(['https://ovens.example/pizza', 'https://blog.example/x'])
+    expect(store.search('pizza tomatoes').map((h) => h.url).sort()).toEqual(['https://blog.example/x', 'https://other.example/', 'https://ovens.example/pizza'])
+    expect(store.search('pizza')[0]!.summary).toBe('how to build a brick oven')
+  })
+
+  test('a re-crawl replaces the page, it does not add a second one', () => {
+    const store = new IndexStore(':memory:')
+    store.add([JSON.stringify(entry('https://a.example/', 'Old title', 'old words'))])
+    store.add([JSON.stringify(entry('https://a.example/', 'New title', 'fresh words'))])
+    expect(store.count()).toBe(1)
+    expect(store.search('old')).toEqual([])
+    expect(store.search('fresh')[0]!.title).toBe('New title')
+  })
+
+  test('publishing fills the mirror, even with nichedb off', async () => {
+    const store = new IndexStore(':memory:')
+    const frontier = new MemoryFrontier(10)
+    const noop = async () => {}
+    const crawler = createCrawler({ config: configFromEnv({}), frontier, index: store, resolver: async () => true, enqueue: { sites: noop, pages: noop, render: noop } })
+    await frontier.push([entry('https://a.example/', 'Hello mirror', 'words')], 100)
+    expect(await crawler.publish()).toEqual({ sent: 0, added: 0, rejected: 0, waiting: false })
+    expect(store.search('mirror')).toHaveLength(1)
+    expect(frontier.counters.indexed_local).toBe(1)
+  })
+})

src/crawl-status.ts

@@ -40,7 +40,7 @@ export type CrawlStatus = {
   health: Health
   summary: string
   reachable: boolean
-  totals: { sites: number; pagesIndexed: number; pagesRendered: number; published: number; outbox: number; failed: number }
+  totals: { sites: number; pagesIndexed: number; pagesRendered: number; published: number; outbox: number; failed: number; searchable: number }
   lastHour: { pages: number; rendered: number; failed: number; published: number; sitesStarted: number }
   lastDay: { pages: number; rendered: number; failed: number; published: number; sitesStarted: number }
   jobs: Job[]
@@ -75,7 +75,7 @@ export function describe(report: CrawlerReport | null, now = new Date()): CrawlS
       health: 'offline',
       summary: 'The crawler did not answer. Searches still work; the index just is not growing right now.',
       reachable: false,
-      totals: { sites: 0, pagesIndexed: 0, pagesRendered: 0, published: 0, outbox: 0, failed: 0 },
+      totals: { sites: 0, pagesIndexed: 0, pagesRendered: 0, published: 0, outbox: 0, failed: 0, searchable: 0 },
       lastHour: { pages: 0, rendered: 0, failed: 0, published: 0, sitesStarted: 0 },
       lastDay: { pages: 0, rendered: 0, failed: 0, published: 0, sitesStarted: 0 },
       jobs: [],
@@ -137,6 +137,7 @@ export function describe(report: CrawlerReport | null, now = new Date()): CrawlS
       published: n(stats, 'published'),
       outbox: n(stats, 'outbox'),
       failed: n(stats, 'pages_failed', 'pages_errored', 'renders_errored'),
+      searchable: n(stats, 'index_pages'),
     },
     lastHour,
     lastDay,