src/doc-index.ts

Functions

Imports

It imports src/doc-reads.ts, src/env.ts, src/limits.ts, src/log.ts. Imported by src/health.ts, src/tick.ts, src/tools/docs-read.ts.

Source

1import type { Env } from "./env";
2import { logEvent } from "./log";
3import { SEARCH_ROWS } from "./limits";
4import type { SearchFilter, SearchRow } from "./doc-reads";
5
6// The meaning index: every document, copied into a Cloudflare AI Search instance on
7// built-in storage so search can match by meaning as well as by keyword
8// (capsid/research/design-hybrid-search.md, step 1). D1 stays the source of truth.
9// The index is derived from it and can be rebuilt from nothing.
10//
11// It is fed by the five-minute tick, not by the write paths: every writer of
12// `documents` (documentUpsert, pathMutation, restore, docStatements) sets
13// updated_at, so one watermark over (updated_at, id) catches them all without a hook
14// in each. A move or delete leaves an index item behind; the daily reconcile removes
15// it, and search re-checks every hit against D1 in any case, so a stale item is never
16// served.
17//
18// Unbound (tests, local dev, a deploy without the binding), it does nothing and says so.
19
20export const INDEX_INSTANCE = "capsid-docs";
21// Documents sent per tick. 1,788 documents on 2026-10-15 backfill in about 90 minutes.
22export const INDEX_BATCH = 100;
23// Index items deleted per reconcile, so a mistake removes a bounded number.
24const RECONCILE_DELETE_MAX = 200;
25// AI Search lists at most 50 items a page.
26const LIST_PAGE = 50;
27
28// v2: the first live runs (2026-10-15) moved v1 past documents the index refused, so
29// the backfill starts again. Uploads are upserts, so re-sending a document is harmless.
30export const CURSOR_KEY = "doc-index:cursor:v2";
31const READY_KEY = "doc-index:instance";
32const RECONCILED_KEY = "doc-index:reconciled";
33
34// Five fields, AI Search's limit per instance. A filter reads the first 64 bytes of a
35// string, and namespace names are far shorter.
36const INDEX_METADATA = [
37  { field_name: "namespace", data_type: "text" },
38  { field_name: "type", data_type: "text" },
39  { field_name: "status", data_type: "text" },
40  { field_name: "doc_id", data_type: "number" },
41  { field_name: "canon", data_type: "boolean" },
42] as const;
43
44interface IndexRow {
45  id: number;
46  namespace: string;
47  path: string;
48  title: string | null;
49  body: string | null;
50  type: string | null;
51  status: string | null;
52  updated_at: string;
53}
54
55interface Cursor {
56  updated_at: string;
57  id: number;
58}
59
60// The current canon: the documents a ruling or convention lives in. Search boosts
61// these over a session log that quotes an old ruling (design section 3.3).
62export function isCanon(namespace: string, path: string): boolean {
63  if (path === "core.md") return true;
64  if (namespace !== "capsid") return false;
65  return (
66    /^conventions[^/]*\.md$/.test(path) ||
67    path === "archive/decisions-vol-4.md" ||
68    path === "archive/decisions-vol-8.md" ||
69    path.startsWith("rulings/")
70  );
71}
72
73// The item key. AI Search reads the type from the extension, so a path without .md
74// gets one; the key is never read back into a path (doc_id is).
75export function indexKey(namespace: string, path: string): string {
76  return `${namespace}/${path}${path.endsWith(".md") ? "" : ".md"}`;
77}
78
79function indexContent(row: IndexRow): string {
80  return `# ${row.title ?? row.path}\n\n${row.body ?? ""}`;
81}
82
83// Every value is sent as a string, as R2 custom metadata is: the live index refused
84// a number and a boolean as invalid_metadata_format (2026-10-15). The instance's schema
85// types them, so doc_id still filters as a number and canon as a boolean.
86function indexMetadata(row: IndexRow): Record<string, string> {
87  return {
88    namespace: row.namespace,
89    type: row.type ?? "note",
90    status: row.status ?? "published",
91    doc_id: String(row.id),
92    canon: String(isCanon(row.namespace, row.path)),
93  };
94}
95
96export interface IndexReport {
97  ran: boolean;
98  note: string;
99  sent: number;
100  failed: number;
101  deleted: number;
102}
103
104const idle = (note: string): IndexReport => ({ ran: false, note, sent: 0, failed: 0, deleted: 0 });
105
106async function ensureInstance(env: Env, ns: AiSearchNamespace): Promise<AiSearchInstance> {
107  const instance = ns.get(INDEX_INSTANCE);
108  if ((await env.APP_KV.get(READY_KEY)) === "ready") return instance;
109  try {
110    await instance.info();
111  } catch {
112    await ns.create({
113      id: INDEX_INSTANCE,
114      index_method: { vector: true, keyword: true },
115      fusion_method: "rrf",
116      custom_metadata: INDEX_METADATA.map((m) => ({ ...m })),
117    });
118    logEvent("log", "DOC_INDEX_CREATED", { message: `DOC_INDEX_CREATED ${INDEX_INSTANCE}` });
119  }
120  await env.APP_KV.put(READY_KEY, "ready");
121  return instance;
122}
123
124async function readCursor(env: Env): Promise<Cursor> {
125  const raw = await env.APP_KV.get(CURSOR_KEY);
126  if (!raw) return { updated_at: "", id: 0 };
127  const parsed = JSON.parse(raw) as Partial<Cursor>;
128  return { updated_at: String(parsed.updated_at ?? ""), id: Number(parsed.id ?? 0) };
129}
130
131// Documents changed since the cursor, oldest first. updated_at has one-second
132// resolution, so the id breaks ties and two writes in one second are both sent.
133async function changedSince(db: D1Database, cursor: Cursor, limit: number): Promise<IndexRow[]> {
134  const { results } = await db
135    .prepare(
136      `SELECT id, namespace, path, title, body, type, status, updated_at FROM documents
137       WHERE (updated_at, id) > (?1, ?2)
138       ORDER BY updated_at, id LIMIT ?3`
139    )
140    .bind(cursor.updated_at, cursor.id, limit)
141    .all<IndexRow>();
142  return results;
143}
144
145// One document to the index. A refusal is logged and reported, never thrown, so one
146// document AI Search will not take cannot stop the rest.
147// Returns null when sent, or the refusal's message.
148async function upload(instance: AiSearchInstance, row: IndexRow): Promise<string | null> {
149  try {
150    await instance.items.upload(indexKey(row.namespace, row.path), indexContent(row), { metadata: indexMetadata(row) });
151    return null;
152  } catch (err) {
153    const message = err instanceof Error ? err.message : String(err);
154    logEvent("error", "DOC_INDEX_UPLOAD_FAILED", { message: `DOC_INDEX_UPLOAD_FAILED ${row.namespace}/${row.path}: ${message}` });
155    return message;
156  }
157}
158
159/** Send what changed since the last tick. The cursor moves past every row tried, sent
160 *  or not: one document AI Search refuses must not hold back the rest. A refusal is
161 *  logged and counted, and the run is recorded as refused so it shows on Health. When
162 *  every row is refused the fault is the index's, not a document's, so the cursor stays
163 *  and the same rows are tried next tick. */
164async function sendChanged(env: Env, instance: AiSearchInstance): Promise<{ tried: number; sent: number; failed: number; refusal: string | null }> {
165  const rows = await changedSince(env.DB, await readCursor(env), INDEX_BATCH);
166  let sent = 0;
167  let failed = 0;
168  let refusal: string | null = null;
169  for (const row of rows) {
170    const refused = await upload(instance, row);
171    if (refused === null) sent++;
172    else {
173      failed++;
174      refusal ??= refused;
175    }
176  }
177  const last = rows.at(-1);
178  if (last && sent > 0) await env.APP_KV.put(CURSOR_KEY, JSON.stringify({ updated_at: last.updated_at, id: last.id }));
179  return { tried: rows.length, sent, failed, refusal };
180}
181
182/** Once a UTC day: remove index items whose document is gone or has moved, and
183 *  re-send documents the index lacks (an upload that failed). Each side is bounded;
184 *  the rest go on the next day's pass. */
185async function reconcile(env: Env, instance: AiSearchInstance): Promise<{ deleted: number; resent: number; failed: number }> {
186  const { results } = await env.DB.prepare("SELECT id, namespace, path FROM documents").all<{ id: number; namespace: string; path: string }>();
187  const live = new Map(results.map((r) => [r.id, indexKey(r.namespace, r.path)]));
188  const present = new Set<string>();
189  const stale: string[] = [];
190  for (let page = 1; ; page++) {
191    const listed = await instance.items.list({ page, per_page: LIST_PAGE });
192    for (const item of listed.result) {
193      if (live.get(Number(item.metadata?.doc_id)) === item.key) present.add(item.key);
194      else stale.push(item.id);
195    }
196    if (listed.result.length < LIST_PAGE) break;
197  }
198  let deleted = 0;
199  for (const id of stale.slice(0, RECONCILE_DELETE_MAX)) {
200    await instance.items.delete(id);
201    deleted++;
202  }
203  const missing = [...live].filter(([, key]) => !present.has(key)).map(([id]) => id).slice(0, INDEX_BATCH);
204  let resent = 0;
205  let failed = 0;
206  for (const id of missing) {
207    const row = await env.DB.prepare(`SELECT id, namespace, path, title, body, type, status, updated_at FROM documents WHERE id = ?1`).bind(id).first<IndexRow>();
208    if (!row) continue;
209    if ((await upload(instance, row)) === null) resent++;
210    else failed++;
211  }
212  return { deleted, resent, failed };
213}
214
215export async function docIndexTick(env: Env, now: Date): Promise<IndexReport> {
216  const ns = env.AI_SEARCH;
217  if (!ns) return idle("AI_SEARCH is not bound");
218  const instance = await ensureInstance(env, ns);
219  const { tried, sent, failed, refusal } = await sendChanged(env, instance);
220  const today = now.toISOString().slice(0, 10);
221  let deleted = 0;
222  let resent = 0;
223  let refused = 0;
224  // Not until the backfill is done (before then every document is "missing"), and not
225  // while the index refuses everything.
226  if (tried < INDEX_BATCH && (sent > 0 || failed === 0) && (await env.APP_KV.get(RECONCILED_KEY)) !== today) {
227    ({ deleted, resent, failed: refused } = await reconcile(env, instance));
228    await env.APP_KV.put(RECONCILED_KEY, today);
229  }
230  const allFailed = failed + refused;
231  if (sent === 0 && allFailed === 0 && deleted === 0 && resent === 0) return idle("nothing changed");
232  return {
233    ran: true,
234    note: `sent ${sent}, failed ${failed}${refusal ? ` (first refusal: ${refusal})` : ""}; reconcile removed ${deleted} stale item(s), re-sent ${resent}, failed ${refused}`,
235    sent: sent + resent,
236    failed: allFailed,
237    deleted,
238  };
239}
240
241// Hybrid search (step 2 of the design): meaning plus keyword, through the index. The
242// index decides only the order. Every hit is read back from D1 with the caller's
243// filters, so a document the caller may not see, or one deleted or moved since it was
244// indexed, is never returned. Anything that keeps the index from answering returns a
245// reason instead of rows, and the caller falls back to FTS5.
246//
247// Queries are counted per month in KV. The account's free 1,000 a month is shared with
248// the site's Ask, and a paid hybrid query costs $0.00075, so the cap holds the worst case
249// near $4 a month (Dustin's $5 cap, decision D1 of the design).
250export const HYBRID_MONTHLY_CAP = 5000;
251const RERANK_MODEL = "@cf/baai/bge-reranker-base";
252const SNIPPET_CHARS = 240;
253// Closed tasks and superseded rulings stay searchable on request but are hidden by
254// default, so an old answer does not sit beside the current one (design 3.2).
255const RETIRED = ["closed", "superseded"];
256
257const countKey = (now: Date) => `doc-index:queries:${now.toISOString().slice(0, 7)}`;
258
259/** This month's hybrid queries, for /health (design 3.5). */
260export async function aiSearchUsage(env: Env, now: Date = new Date()): Promise<{ month: string; hybrid: number; cap: number }> {
261  return { month: now.toISOString().slice(0, 7), hybrid: Number((await env.APP_KV.get(countKey(now))) ?? 0), cap: HYBRID_MONTHLY_CAP };
262}
263
264type Match = "keyword" | "meaning" | "both";
265export type HybridRow = SearchRow & { match: Match };
266
267function matchOf(chunk: AiSearchSearchResponse["chunks"][number]): Match {
268  const d = chunk.scoring_details;
269  const keyword = d?.keyword_rank !== undefined || (d?.keyword_score ?? 0) > 0;
270  const meaning = d?.vector_rank !== undefined || (d?.vector_score ?? 0) > 0;
271  return keyword && meaning ? "both" : keyword ? "keyword" : "meaning";
272}
273
274const joinMatch = (a: Match, b: Match): Match => (a === b ? a : "both");
275
276export async function hybridSearch(
277  env: Env,
278  query: string,
279  filter: SearchFilter & { includeRetired?: boolean },
280  now: Date = new Date()
281): Promise<HybridRow[] | string> {
282  const ns = env.AI_SEARCH;
283  if (!ns) return "AI_SEARCH is not bound";
284  if ((await env.APP_KV.get(READY_KEY)) !== "ready") return "the search index is not built yet";
285  const key = countKey(now);
286  const used = Number((await env.APP_KV.get(key)) ?? 0);
287  if (used >= HYBRID_MONTHLY_CAP) return `this month's ${HYBRID_MONTHLY_CAP} hybrid queries are used`;
288  await env.APP_KV.put(key, String(used + 1), { expirationTtl: 40 * 86_400 });
289
290  const filters: Record<string, { $eq: string } | { $nin: string[] }> = {};
291  if (filter.namespace) filters.namespace = { $eq: filter.namespace };
292  if (filter.type) filters.type = { $eq: filter.type };
293  if (!filter.includeRetired) filters.status = { $nin: RETIRED };
294  let chunks: AiSearchSearchResponse["chunks"];
295  try {
296    ({ chunks } = await ns.get(INDEX_INSTANCE).search({
297      query,
298      ai_search_options: {
299        retrieval: {
300          retrieval_type: "hybrid",
301          max_num_results: 50,
302          return_on_failure: false,
303          ...(Object.keys(filters).length > 0 ? { filters } : {}),
304          // The current canon ranks above a session log that quotes an old ruling (3.3).
305          boost_by: [{ field: "canon", direction: "desc" }],
306        },
307        // A cross-encoder reorders the fused candidates by how well each answers the
308        // query, the step the usual hybrid pipeline adds after fusion. Threshold 0: it
309        // reorders, it never drops a candidate the fusion found.
310        reranking: { enabled: true, model: RERANK_MODEL, match_threshold: 0 },
311      },
312    }));
313  } catch (err) {
314    const message = err instanceof Error ? err.message : String(err);
315    logEvent("warn", "DOC_INDEX_SEARCH_FAILED", { message: `DOC_INDEX_SEARCH_FAILED ${message}` });
316    return `the search index refused the query: ${message}`;
317  }
318
319  // Best chunk first, so a document's first chunk sets its rank and its snippet; every
320  // chunk of it adds to how it matched.
321  const ranked = new Map<number, { snippet: string; match: Match }>();
322  for (const chunk of chunks) {
323    const id = Number(chunk.item.metadata?.doc_id);
324    if (!Number.isInteger(id)) continue;
325    const seen = ranked.get(id);
326    if (seen) seen.match = joinMatch(seen.match, matchOf(chunk));
327    else ranked.set(id, { snippet: chunk.text.replace(/\s+/g, " ").trim().slice(0, SNIPPET_CHARS), match: matchOf(chunk) });
328  }
329  if (ranked.size === 0) return [];
330  const { results } = await env.DB.prepare(
331    `SELECT id, namespace, path, title, type, status, updated_at FROM documents
332     WHERE id IN (SELECT value FROM json_each(?1))
333       AND (?2 IS NULL OR namespace = ?2)
334       AND (?3 IS NULL OR type = ?3)
335       AND (?4 IS NULL OR path IN (SELECT value FROM json_each(?4)))
336       AND (?5 = 1 OR status IS NULL OR status NOT IN (SELECT value FROM json_each(?6)))`
337  )
338    .bind(JSON.stringify([...ranked.keys()]), filter.namespace, filter.type, filter.allowed, filter.includeRetired ? 1 : 0, JSON.stringify(RETIRED))
339    .all<Omit<SearchRow, "snippet">>();
340  const byId = new Map(results.map((r) => [r.id, r]));
341  const rows: HybridRow[] = [];
342  for (const [id, hit] of ranked) {
343    const row = byId.get(id);
344    if (row) rows.push({ ...row, ...hit });
345    if (rows.length > SEARCH_ROWS) break;
346  }
347  return rows;
348}