Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
616 linesCodeBlameRaw
1/**
2 * The semantic index: pages and projects' docs files as passages
3 * (src/chunks.ts) in D1 (`doc_chunks`, with full text in
4 * `doc_chunks_fts`) and as vectors in the vector store (src/vectors.ts).
5 *
6 * A page is indexed a little after its Markdown is saved (the room,
7 * src/room.ts, in the background, so editing never waits on it); a
8 * project's docs files after they are read (src/repo-spaces.ts). Only
9 * passages whose text changed are embedded; one that only moved keeps its
10 * vector. Embedding is capped per workspace and hour (EMBED_PER_HOUR):
11 * past the cap passages are kept for words, and a catch-up run on the
12 * queue embeds them later. A failure leaves them for the next save or run.
13 * Nothing here throws to its caller.
14 *
15 * The backfill indexes a workspace's existing pages and files in batches
16 * on the queue (`docs.index` jobs, on the artifacts service's own events queue).
17 */
18import { chunkId, chunkMarkdown, embedText, repoFileId, textHash } from "./chunks.ts";
19import { overFreeBudget } from "./embed-billing.ts";
20import { scopesOf, type FolioRow, FOLIO_COLUMNS } from "./folios/access-store.ts";
21import { kindModel } from "./kinds/index.ts";
22import { cloudflareEmbedder, cloudflareVectors, type Embedder, type VectorMetadata, type VectorStore } from "./vectors.ts";
23
24/** Passages embedded per workspace per hour at most. Logged when reached. */
25export const EMBED_PER_HOUR = 200;
26/** Documents one backfill job indexes before handing on to the next. */
27const BACKFILL_BATCH = 20;
28/** A backfill not finished after this long is taken to have died and may start again. */
29const BACKFILL_STALE_MS = 2 * 60 * 60 * 1000;
30/** How long a run waits when the hour's cap is reached. */
31const CAP_DELAY_SECONDS = 3600;
32
33/**
34 * A job on the queue: index more of a workspace's docs and folios; or
35 * bring a large folio subtree's access, rooms and index scope up to date
36 * after a move or a sharing change (src/folios/service.ts).
37 */
38export type DocsJob = { type: "docs.index"; workspace_id: string } | { type: "folios.reacl"; folio_id: string };
39
40export type IndexEnv = {
41 DB: D1Database;
42 AI?: Ai;
43 VECTORS?: Vectorize;
44 /** Folios' semantic index, Vectorize `g1t-folios`. Optional: without it folios match words only. */
45 FOLIO_VECTORS?: Vectorize;
46 JOBS?: Queue<DocsJob>;
47};
48
49/** The embedder and store this deployment has; null without them (words only). */
50export function adapters(env: IndexEnv): { embedder: Embedder | null; store: VectorStore | null } {
51 return env.AI && env.VECTORS ? { embedder: cloudflareEmbedder(env.AI), store: cloudflareVectors(env.VECTORS) } : { embedder: null, store: null };
52}
53
54/** One document to index: a page, or a project's docs file. */
55export type IndexDoc = {
56 kind: "page" | "repo_file";
57 /** The page's id, or the file's `rf_` id. */
58 doc_id: string;
59 workspace_id: string;
60 space_id: string;
61 title: string;
62 markdown: string;
63 repo_id?: string | null;
64 path?: string | null;
65};
66
67type ChunkRow = { id: string; seq: number; space_id: string; hash: string; vector_hash: string | null };
68
69const hourOf = (now: Date) => now.toISOString().slice(0, 13);
70
71function metadataOf(doc: IndexDoc): VectorMetadata {
72 return doc.kind === "page"
73 ? { workspace_id: doc.workspace_id, space_id: doc.space_id, kind: "page", page_id: doc.doc_id }
74 : { workspace_id: doc.workspace_id, space_id: doc.space_id, kind: "repo_file", repo_file_id: doc.doc_id, repo_id: doc.repo_id ?? "" };
75}
76
77/**
78 * Passages the workspace may still embed this hour; none once a workspace
79 * without the plan has used its free month (src/embed-billing.ts).
80 */
81async function allowance(db: D1Database, workspaceId: string, now: Date): Promise<number> {
82 if (await overFreeBudget(db, workspaceId, now)) {
83 console.log("embedding budget of a free workspace used for the month", workspaceId);
84 return 0;
85 }
86 const used = await db.prepare("SELECT chunks FROM doc_embed_usage WHERE workspace_id = ? AND hour = ?").bind(workspaceId, hourOf(now)).first<{ chunks: number }>();
87 return Math.max(0, EMBED_PER_HOUR - (used?.chunks ?? 0));
88}
89
90async function meter(db: D1Database, workspaceId: string, chunks: number, chars: number, now: Date): Promise<void> {
91 if (!chunks) return;
92 await db
93 .prepare(
94 "INSERT INTO doc_embed_usage (workspace_id, hour, chunks, tokens) VALUES (?1, ?2, ?3, ?4) ON CONFLICT (workspace_id, hour) DO UPDATE SET chunks = chunks + ?3, tokens = tokens + ?4",
95 )
96 .bind(workspaceId, hourOf(now), chunks, Math.ceil(chars / 4))
97 .run();
98}
99
100/**
101 * Brings one document's passages up to date: rows and full text in D1,
102 * vectors for those that changed (within the hour's cap). `capped` when
103 * some were left for later.
104 */
105export async function indexDoc(
106 db: D1Database,
107 embedder: Embedder | null,
108 store: VectorStore | null,
109 doc: IndexDoc,
110 now = new Date(),
111): Promise<{ chunks: number; embedded: number; capped: boolean }> {
112 const at = now.toISOString();
113 const chunks = chunkMarkdown(doc.markdown, doc.title).map((c) => {
114 const embed = embedText(doc.title, c);
115 return { ...c, id: chunkId(doc.doc_id, c.seq), embed, hash: textHash(embed) };
116 });
117 const column = doc.kind === "page" ? "page_id" : "repo_file_id";
118 const old = (await db.prepare(`SELECT id, seq, space_id, hash, vector_hash FROM doc_chunks WHERE ${column} = ?`).bind(doc.doc_id).all<ChunkRow>()).results;
119 const byId = new Map(old.map((r) => [r.id, r]));
120 const statements: D1PreparedStatement[] = [];
121 for (const c of chunks) {
122 const was = byId.get(c.id);
123 if (was && was.hash === c.hash && was.space_id === doc.space_id) continue;
124 statements.push(
125 db
126 .prepare(
127 `INSERT INTO doc_chunks (id, workspace_id, space_id, page_id, repo_file_id, repo_id, path, seq, heading, text, hash, vector_hash, updated_at)
128 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)
129 ON CONFLICT (id) DO UPDATE SET space_id = excluded.space_id, repo_id = excluded.repo_id, path = excluded.path, heading = excluded.heading, text = excluded.text, hash = excluded.hash, updated_at = excluded.updated_at`,
130 )
131 .bind(
132 c.id,
133 doc.workspace_id,
134 doc.space_id,
135 doc.kind === "page" ? doc.doc_id : null,
136 doc.kind === "repo_file" ? doc.doc_id : null,
137 doc.repo_id ?? null,
138 doc.path ?? null,
139 c.seq,
140 c.heading,
141 c.text,
142 c.hash,
143 at,
144 ),
145 db.prepare("DELETE FROM doc_chunks_fts WHERE chunk_id = ?").bind(c.id),
146 // Only beside a chunk row the database kept: an erased document's insert was ignored (migration 0010).
147 db.prepare("INSERT INTO doc_chunks_fts (chunk_id, space_id, doc_id, heading, text) SELECT ?1, ?2, ?3, ?4, ?5 WHERE EXISTS (SELECT 1 FROM doc_chunks WHERE id = ?1)").bind(c.id, doc.space_id, doc.doc_id, c.heading ?? "", c.text),
148 );
149 }
150 const keep = new Set(chunks.map((c) => c.id));
151 const gone = old.filter((r) => !keep.has(r.id));
152 for (const r of gone) {
153 statements.push(db.prepare("DELETE FROM doc_chunks WHERE id = ?").bind(r.id), db.prepare("DELETE FROM doc_chunks_fts WHERE chunk_id = ?").bind(r.id));
154 }
155 for (let i = 0; i < statements.length; i += 60) await db.batch(statements.slice(i, i + 60));
156
157 if (!store || !embedder) return { chunks: chunks.length, embedded: 0, capped: false };
158 if (await erasedSource(db, doc.kind === "page" ? "page" : null, doc.doc_id, doc.workspace_id)) return { chunks: 0, embedded: 0, capped: false };
159 const result = await embedChanged(db, embedder, store, doc, chunks, old, byId, now);
160 await dropIfErased(db, store, doc.kind === "page" ? "page" : null, doc.doc_id, doc.workspace_id, chunks.map((c) => c.id));
161 // Last, so a passage that moved from a place now gone could take its vector first.
162 try {
163 if (gone.some((r) => r.vector_hash)) await store.delete(gone.filter((r) => r.vector_hash).map((r) => r.id));
164 } catch (error) {
165 console.error("docs could not drop old passages from the index", doc.doc_id, String(error));
166 }
167 return result;
168}
169
170type IndexedChunk = { id: string; seq: number; heading: string | null; text: string; embed: string; hash: string };
171
172/** Vectors for the passages that need one: reused when the index already holds their text, embedded otherwise (within the cap). */
173async function embedChanged(
174 db: D1Database,
175 embedder: Embedder,
176 store: VectorStore,
177 doc: IndexDoc,
178 chunks: IndexedChunk[],
179 old: ChunkRow[],
180 byId: Map<string, ChunkRow>,
181 now: Date,
182): Promise<{ chunks: number; embedded: number; capped: boolean }> {
183 // What needs a vector: a passage whose text the index doesn't hold under its id, or one whose space changed.
184 const stale = chunks.filter((c) => {
185 const was = byId.get(c.id);
186 return !was || was.vector_hash !== c.hash || was.space_id !== doc.space_id;
187 });
188 if (!stale.length) return { chunks: chunks.length, embedded: 0, capped: false };
189 // A passage that only moved (or changed space) has its vector already, under another id or this one.
190 const held = new Map<string, string>();
191 for (const r of old) if (r.vector_hash) held.set(r.vector_hash, r.id);
192 const reuse = stale.filter((c) => held.has(c.hash));
193 const fresh = stale.filter((c) => !held.has(c.hash));
194 const budget = fresh.length ? await allowance(db, doc.workspace_id, now) : 0;
195 const embedNow = fresh.slice(0, budget);
196 const capped = embedNow.length < fresh.length;
197 if (capped) console.log("docs embedding cap reached", doc.workspace_id, `${fresh.length - embedNow.length} passages wait for the next hour`);
198 const metadata = metadataOf(doc);
199 const done: { id: string; hash: string }[] = [];
200 try {
201 const vectors: { id: string; values: number[]; metadata: VectorMetadata }[] = [];
202 if (reuse.length) {
203 const values = new Map((await store.get([...new Set(reuse.map((c) => held.get(c.hash)!))])).map((v) => [v.id, v.values]));
204 for (const c of reuse) {
205 const v = values.get(held.get(c.hash)!);
206 if (v) vectors.push({ id: c.id, values: v, metadata });
207 }
208 }
209 if (embedNow.length) {
210 const embedded = await embedder.embed(embedNow.map((c) => c.embed));
211 embedNow.forEach((c, i) => vectors.push({ id: c.id, values: embedded[i]!, metadata }));
212 await meter(
213 db,
214 doc.workspace_id,
215 embedNow.length,
216 embedNow.reduce((n, c) => n + c.embed.length, 0),
217 now,
218 );
219 }
220 if (vectors.length) await store.upsert(vectors);
221 const hashOf = new Map(chunks.map((c) => [c.id, c.hash]));
222 for (const v of vectors) done.push({ id: v.id, hash: hashOf.get(v.id)! });
223 } catch (error) {
224 console.error("docs could not embed passages; the next save tries again", doc.doc_id, String(error));
225 }
226 if (done.length) {
227 const updates = done.map((d) => db.prepare("UPDATE doc_chunks SET vector_hash = ? WHERE id = ?").bind(d.hash, d.id));
228 for (let i = 0; i < updates.length; i += 60) await db.batch(updates.slice(i, i + 60));
229 }
230 return { chunks: chunks.length, embedded: embedNow.length, capped };
231}
232
233/** Takes passages out of D1 and the index: by page ids, file ids, or a whole space. */
234export async function forgetDocs(env: IndexEnv, which: { page_ids?: string[]; repo_file_ids?: string[]; space_id?: string }): Promise<void> {
235 const db = env.DB;
236 const { store } = adapters(env);
237 try {
238 const found: string[] = [];
239 const select = async (column: string, values: string[]) => {
240 for (let i = 0; i < values.length; i += 90) {
241 const part = values.slice(i, i + 90);
242 const rows = await db.prepare(`SELECT id FROM doc_chunks WHERE ${column} IN (${part.map(() => "?").join(",")})`).bind(...part).all<{ id: string }>();
243 found.push(...rows.results.map((r) => r.id));
244 }
245 };
246 if (which.page_ids?.length) await select("page_id", which.page_ids);
247 if (which.repo_file_ids?.length) await select("repo_file_id", which.repo_file_ids);
248 if (which.space_id) await select("space_id", [which.space_id]);
249 if (!found.length) return;
250 if (store) {
251 try {
252 await store.delete(found);
253 } catch (error) {
254 console.error("docs could not drop passages from the index", found.length, String(error));
255 }
256 }
257 const statements = found.flatMap((id) => [db.prepare("DELETE FROM doc_chunks WHERE id = ?").bind(id), db.prepare("DELETE FROM doc_chunks_fts WHERE chunk_id = ?").bind(id)]);
258 for (let i = 0; i < statements.length; i += 80) await db.batch(statements.slice(i, i + 80));
259 } catch (error) {
260 console.error("docs could not forget passages", String(error));
261 }
262}
263
264type PageForIndex = { id: string; workspace_id: string; space_id: string; title: string; markdown: string; archived_at: string | null };
265
266/** Indexes one page as it is saved now; forgets it when it is gone or in the trash. Never throws. */
267export async function indexPage(env: IndexEnv, pageId: string, now = new Date()): Promise<{ capped: boolean }> {
268 try {
269 const page = await env.DB.prepare("SELECT id, workspace_id, space_id, title, markdown, archived_at FROM pages WHERE id = ?").bind(pageId).first<PageForIndex>();
270 if (!page || page.archived_at) {
271 await forgetDocs(env, { page_ids: [pageId] });
272 return { capped: false };
273 }
274 const { embedder, store } = adapters(env);
275 const result = await indexDoc(env.DB, embedder, store, { kind: "page", doc_id: page.id, workspace_id: page.workspace_id, space_id: page.space_id, title: page.title, markdown: page.markdown }, now);
276 if (result.capped) await startBackfill(env, page.workspace_id, { delaySeconds: CAP_DELAY_SECONDS });
277 return { capped: result.capped };
278 } catch (error) {
279 console.error("docs could not index a page", pageId, String(error));
280 return { capped: false };
281 }
282}
283
284type RepoFileForIndex = { space_id: string; path: string; title: string; markdown: string; workspace_id: string; repo_id: string };
285
286function repoDoc(row: RepoFileForIndex): IndexDoc {
287 return { kind: "repo_file", doc_id: repoFileId(row.space_id, row.path), workspace_id: row.workspace_id, space_id: row.space_id, title: row.title, markdown: row.markdown, repo_id: row.repo_id, path: row.path };
288}
289
290/**
291 * Indexes a project's docs files after they were read: `changed` paths
292 * again, `gone` paths forgotten. Never throws.
293 */
294export async function indexRepoFiles(env: IndexEnv, spaceId: string, changed: string[], gone: string[], now = new Date()): Promise<void> {
295 try {
296 if (gone.length) await forgetDocs(env, { repo_file_ids: gone.map((p) => repoFileId(spaceId, p)) });
297 if (!changed.length) return;
298 const { embedder, store } = adapters(env);
299 let capped = false;
300 let workspaceId: string | null = null;
301 for (let i = 0; i < changed.length; i += 50) {
302 const part = changed.slice(i, i + 50);
303 const rows = (
304 await env.DB.prepare(
305 `SELECT f.space_id, f.path, f.title, f.markdown, s.workspace_id, s.repo_id FROM repo_files f JOIN repo_spaces s ON s.id = f.space_id WHERE f.space_id = ? AND f.path IN (${part.map(() => "?").join(",")})`,
306 )
307 .bind(spaceId, ...part)
308 .all<RepoFileForIndex>()
309 ).results;
310 for (const row of rows) {
311 workspaceId = row.workspace_id;
312 // Past the cap, the rest are only kept for words until the catch-up run.
313 const result = await indexDoc(env.DB, capped ? null : embedder, capped ? null : store, repoDoc(row), now);
314 capped ||= result.capped;
315 }
316 }
317 if (capped && workspaceId) await startBackfill(env, workspaceId, { delaySeconds: CAP_DELAY_SECONDS });
318 } catch (error) {
319 console.error("docs could not index a project's docs", spaceId, String(error));
320 }
321}
322
323/**
324 * Starts (or, with `force`, restarts) indexing a workspace's existing
325 * pages and projects' docs on the queue. Does nothing while a run is
326 * going, unless it has been going so long it must have died. Without a
327 * queue, runs one batch now. True when a run was started.
328 */
329export async function startBackfill(env: IndexEnv, workspaceId: string, options: { force?: boolean; delaySeconds?: number } = {}): Promise<boolean> {
330 const at = new Date();
331 const staleBefore = new Date(at.getTime() - BACKFILL_STALE_MS).toISOString();
332 const started = await env.DB.prepare(
333 `INSERT INTO doc_index_runs (workspace_id, started_at, finished_at, cursor, pages, files) VALUES (?1, ?2, NULL, NULL, 0, 0)
334 ON CONFLICT (workspace_id) DO UPDATE SET started_at = ?2, finished_at = NULL, cursor = NULL, pages = 0, files = 0
335 WHERE ?3 OR doc_index_runs.finished_at IS NOT NULL OR doc_index_runs.started_at < ?4
336 RETURNING workspace_id`,
337 )
338 .bind(workspaceId, at.toISOString(), options.force ? 1 : 0, staleBefore)
339 .first<{ workspace_id: string }>();
340 if (!started) return false;
341 await enqueue(env, workspaceId, options.delaySeconds ?? 0);
342 return true;
343}
344
345async function enqueue(env: IndexEnv, workspaceId: string, delaySeconds: number): Promise<void> {
346 if (env.JOBS) {
347 await env.JOBS.send({ type: "docs.index", workspace_id: workspaceId }, delaySeconds ? { delaySeconds } : undefined);
348 return;
349 }
350 // No queue (self-hosted without one): one batch now; saves and later recalls carry on from there.
351 if (!delaySeconds) await runBackfill(env, workspaceId, false);
352}
353
354/**
355 * One batch of a workspace's backfill: pages by id, then projects' docs
356 * files by space and path, from the run's cursor. Hands on to the next
357 * batch (or waits out the hour's cap) through the queue.
358 */
359export async function runBackfill(env: IndexEnv, workspaceId: string, chain = true): Promise<{ done: boolean }> {
360 const run = await env.DB.prepare("SELECT cursor, finished_at, pages, files FROM doc_index_runs WHERE workspace_id = ?").bind(workspaceId).first<{ cursor: string | null; finished_at: string | null; pages: number; files: number }>();
361 if (!run || run.finished_at) return { done: true };
362 const { embedder, store } = adapters(env);
363 const cursor = run.cursor ?? "p:";
364 let next: string | null = cursor;
365 let pages = 0;
366 let files = 0;
367 let capped = false;
368 if (cursor.startsWith("o:")) {
369 // Folios, after pages and before projects' docs.
370 const rows = (
371 await env.DB.prepare(`SELECT ${FOLIO_COLUMNS} FROM folios WHERE workspace_id = ? AND trashed_at IS NULL AND id > ? ORDER BY id LIMIT ?`)
372 .bind(workspaceId, cursor.slice(2), BACKFILL_BATCH)
373 .all<FolioRow>()
374 ).results;
375 for (const row of rows) {
376 const result = await indexFolio(env, row.id);
377 pages++;
378 if (result.capped) {
379 capped = true;
380 break;
381 }
382 next = `o:${row.id}`;
383 }
384 if (!capped && rows.length < BACKFILL_BATCH) next = "f:";
385 } else if (cursor.startsWith("p:")) {
386 const rows = (
387 await env.DB.prepare("SELECT id, workspace_id, space_id, title, markdown, archived_at FROM pages WHERE workspace_id = ? AND archived_at IS NULL AND id > ? ORDER BY id LIMIT ?")
388 .bind(workspaceId, cursor.slice(2), BACKFILL_BATCH)
389 .all<PageForIndex>()
390 ).results;
391 for (const page of rows) {
392 const result = await indexDoc(env.DB, embedder, store, { kind: "page", doc_id: page.id, workspace_id: page.workspace_id, space_id: page.space_id, title: page.title, markdown: page.markdown });
393 pages++;
394 if (result.capped) {
395 // This page again next time, after the hour: the cursor stays before it.
396 capped = true;
397 break;
398 }
399 next = `p:${page.id}`;
400 }
401 if (!capped && rows.length < BACKFILL_BATCH) next = "o:";
402 } else {
403 const [space, path] = splitFileCursor(cursor.slice(2));
404 const rows = (
405 await env.DB.prepare(
406 `SELECT f.space_id, f.path, f.title, f.markdown, s.workspace_id, s.repo_id FROM repo_files f JOIN repo_spaces s ON s.id = f.space_id
407 WHERE s.workspace_id = ? AND (f.space_id > ? OR (f.space_id = ? AND f.path > ?)) ORDER BY f.space_id, f.path LIMIT ?`,
408 )
409 .bind(workspaceId, space, space, path, BACKFILL_BATCH)
410 .all<RepoFileForIndex>()
411 ).results;
412 for (const row of rows) {
413 const result = await indexDoc(env.DB, embedder, store, repoDoc(row));
414 files++;
415 if (result.capped) {
416 capped = true;
417 break;
418 }
419 next = `f:${row.space_id}\n${row.path}`;
420 }
421 if (!capped && rows.length < BACKFILL_BATCH) next = null;
422 }
423 await env.DB.prepare("UPDATE doc_index_runs SET cursor = ?, pages = pages + ?, files = files + ?, finished_at = ? WHERE workspace_id = ?")
424 .bind(next, pages, files, next === null ? new Date().toISOString() : null, workspaceId)
425 .run();
426 if (next === null) return { done: true };
427 if (chain && env.JOBS) await enqueue(env, workspaceId, capped ? CAP_DELAY_SECONDS : 0);
428 return { done: false };
429}
430
431function splitFileCursor(cursor: string): [string, string] {
432 const at = cursor.indexOf("\n");
433 // Empty: from the start.
434 return at < 0 ? ["", ""] : [cursor.slice(0, at), cursor.slice(at + 1)];
435}
436
437// ── Folios (Artifacts mode) ─────────────────────────────────────────────
438
439let saidNoFolioIndex = false;
440
441/**
442 * The embedder and folio index this deployment has. Without Workers AI or
443 * the `g1t-folios` index (FOLIO_VECTORS, which is made by hand before a
444 * deploy that uses it), folios keep their passages in D1 and recall and
445 * search match words; that is said once per isolate.
446 */
447export function folioAdapters(env: IndexEnv): { embedder: Embedder | null; store: VectorStore | null } {
448 if (env.AI && env.FOLIO_VECTORS) return { embedder: cloudflareEmbedder(env.AI), store: cloudflareVectors(env.FOLIO_VECTORS) };
449 if (!saidNoFolioIndex) {
450 saidNoFolioIndex = true;
451 console.log(`folios: no ${env.AI ? "FOLIO_VECTORS (Vectorize g1t-folios)" : "AI"} binding, so artifacts are searched and recalled by their words only`);
452 }
453 return { embedder: null, store: null };
454}
455
456type FolioChunkRow = { id: string; seq: number; scope: string; hash: string; vector_hash: string | null };
457
458/**
459 * Brings one folio's passages up to date, as `indexDoc` does for pages:
460 * rows and full text in D1, under the folio's index scope, and vectors
461 * for passages whose text or scope changed (a passage whose text the
462 * index already holds keeps its vector; only its metadata is written
463 * again). Forgets it when it is gone or in the trash. Never throws.
464 */
465export async function indexFolio(env: IndexEnv, folioId: string, now = new Date()): Promise<{ capped: boolean }> {
466 try {
467 const db = env.DB;
468 const row = await db.prepare(`SELECT ${FOLIO_COLUMNS.replace("'' AS text", "text")} FROM folios WHERE id = ?`).bind(folioId).first<FolioRow>();
469 if (!row || row.trashed_at) {
470 await forgetFolios(env, [folioId]);
471 return { capped: false };
472 }
473 const scope = (await scopesOf(db, [row])).get(row.id) ?? `folio:${row.acl_root}`;
474 const model = kindModel(row.kind);
475 const at = now.toISOString();
476 const chunks = (model ? model.chunks(row.text, row.title) : chunkMarkdown(row.text, row.title)).map((c) => {
477 const embed = embedText(row.title, c);
478 return { ...c, id: chunkId(row.id, c.seq), embed, hash: textHash(embed) };
479 });
480 const old = (await db.prepare("SELECT id, seq, scope, hash, vector_hash FROM folio_chunks WHERE folio_id = ?").bind(row.id).all<FolioChunkRow>()).results;
481 const byId = new Map(old.map((r) => [r.id, r]));
482 const statements: D1PreparedStatement[] = [];
483 for (const c of chunks) {
484 const was = byId.get(c.id);
485 if (was && was.hash === c.hash && was.scope === scope) continue;
486 statements.push(
487 db
488 .prepare(
489 `INSERT INTO folio_chunks (id, workspace_id, folio_id, kind, scope, seq, heading, text, hash, vector_hash, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)
490 ON CONFLICT (id) DO UPDATE SET kind = excluded.kind, scope = excluded.scope, heading = excluded.heading, text = excluded.text, hash = excluded.hash, updated_at = excluded.updated_at`,
491 )
492 .bind(c.id, row.workspace_id, row.id, row.kind, scope, c.seq, c.heading, c.text, c.hash, at),
493 db.prepare("DELETE FROM folio_chunks_fts WHERE chunk_id = ?").bind(c.id),
494 db.prepare("INSERT INTO folio_chunks_fts (chunk_id, scope, folio_id, heading, text) SELECT ?1, ?2, ?3, ?4, ?5 WHERE EXISTS (SELECT 1 FROM folio_chunks WHERE id = ?1)").bind(c.id, scope, row.id, c.heading ?? "", c.text),
495 );
496 }
497 const keep = new Set(chunks.map((c) => c.id));
498 const gone = old.filter((r) => !keep.has(r.id));
499 for (const r of gone) statements.push(db.prepare("DELETE FROM folio_chunks WHERE id = ?").bind(r.id), db.prepare("DELETE FROM folio_chunks_fts WHERE chunk_id = ?").bind(r.id));
500 for (let i = 0; i < statements.length; i += 60) await db.batch(statements.slice(i, i + 60));
501
502 const { embedder, store } = folioAdapters(env);
503 if (!embedder || !store) return { capped: false };
504 if (await erasedSource(db, "folio", row.id, row.workspace_id)) return { capped: false };
505 // What needs a vector written: text the index doesn't hold under this id, or a new scope.
506 const stale = chunks.filter((c) => {
507 const was = byId.get(c.id);
508 return !was || was.vector_hash !== c.hash || was.scope !== scope;
509 });
510 let capped = false;
511 if (stale.length) {
512 const held = new Map<string, string>();
513 for (const r of old) if (r.vector_hash) held.set(r.vector_hash, r.id);
514 const reuse = stale.filter((c) => held.has(c.hash));
515 const fresh = stale.filter((c) => !held.has(c.hash));
516 const budget = fresh.length ? await allowance(db, row.workspace_id, now) : 0;
517 const embedNow = fresh.slice(0, budget);
518 capped = embedNow.length < fresh.length;
519 if (capped) console.log("folios embedding cap reached", row.workspace_id, `${fresh.length - embedNow.length} passages wait for the next hour`);
520 const metadata: VectorMetadata = { workspace_id: row.workspace_id, scope, kind: row.kind, folio_id: row.id };
521 const done: { id: string; hash: string }[] = [];
522 try {
523 const vectors: { id: string; values: number[]; metadata: VectorMetadata }[] = [];
524 if (reuse.length) {
525 const values = new Map((await store.get([...new Set(reuse.map((c) => held.get(c.hash)!))])).map((v) => [v.id, v.values]));
526 for (const c of reuse) {
527 const v = values.get(held.get(c.hash)!);
528 if (v) vectors.push({ id: c.id, values: v, metadata });
529 }
530 }
531 if (embedNow.length) {
532 const embedded = await embedder.embed(embedNow.map((c) => c.embed));
533 embedNow.forEach((c, i) => vectors.push({ id: c.id, values: embedded[i]!, metadata }));
534 await meter(
535 db,
536 row.workspace_id,
537 embedNow.length,
538 embedNow.reduce((n, c) => n + c.embed.length, 0),
539 now,
540 );
541 }
542 if (vectors.length) await store.upsert(vectors);
543 if (vectors.length) await dropIfErased(db, store, "folio", row.id, row.workspace_id, vectors.map((v) => v.id));
544 const hashOf = new Map(chunks.map((c) => [c.id, c.hash]));
545 for (const v of vectors) done.push({ id: v.id, hash: hashOf.get(v.id)! });
546 } catch (error) {
547 console.error("folios could not embed passages; the next save tries again", row.id, String(error));
548 }
549 if (done.length) {
550 const updates = done.map((d) => db.prepare("UPDATE folio_chunks SET vector_hash = ? WHERE id = ?").bind(d.hash, d.id));
551 for (let i = 0; i < updates.length; i += 60) await db.batch(updates.slice(i, i + 60));
552 }
553 }
554 try {
555 if (gone.some((r) => r.vector_hash)) await store.delete(gone.filter((r) => r.vector_hash).map((r) => r.id));
556 } catch (error) {
557 console.error("folios could not drop old passages from the index", row.id, String(error));
558 }
559 if (capped) await startBackfill(env, row.workspace_id, { delaySeconds: CAP_DELAY_SECONDS });
560 return { capped };
561 } catch (error) {
562 console.error("folios could not index", folioId, String(error));
563 return { capped: false };
564 }
565}
566
567/** Takes folios' passages out of D1 and the index. Never throws. */
568export async function forgetFolios(env: IndexEnv, ids: readonly string[]): Promise<void> {
569 if (!ids.length) return;
570 const db = env.DB;
571 try {
572 const found: string[] = [];
573 for (let i = 0; i < ids.length; i += 500) {
574 const rows = await db
575 .prepare("SELECT id FROM folio_chunks WHERE folio_id IN (SELECT value FROM json_each(?))")
576 .bind(JSON.stringify(ids.slice(i, i + 500)))
577 .all<{ id: string }>();
578 found.push(...rows.results.map((r) => r.id));
579 }
580 if (!found.length) return;
581 const { store } = folioAdapters(env);
582 if (store) {
583 try {
584 await store.delete(found);
585 } catch (error) {
586 console.error("folios could not drop passages from the index", found.length, String(error));
587 }
588 }
589 const statements = found.flatMap((id) => [db.prepare("DELETE FROM folio_chunks WHERE id = ?").bind(id), db.prepare("DELETE FROM folio_chunks_fts WHERE chunk_id = ?").bind(id)]);
590 for (let i = 0; i < statements.length; i += 80) await db.batch(statements.slice(i, i + 80));
591 } catch (error) {
592 console.error("folios could not forget passages", String(error));
593 }
594}
595
596/** Whether a document, or its workspace, was erased (src/erasure.ts): nothing of it is indexed again. */
597export async function erasedSource(db: D1Database, kind: "page" | "folio" | null, id: string, workspaceId: string): Promise<boolean> {
598 const row = await db
599 .prepare("SELECT 1 FROM erased_subjects WHERE id = ?1 AND kind = 'workspace' UNION ALL SELECT 1 FROM erased_documents WHERE id = ?2 AND kind = ?3 LIMIT 1")
600 .bind(workspaceId, id, kind ?? "")
601 .first();
602 return row != null;
603}
604
605/**
606 * Checked again after vectors were written: an erasure that ran while they
607 * were being embedded found no chunk rows for them, so they are taken back
608 * here. Throws when they cannot be, so the caller does not record them.
609 */
610export async function dropIfErased(db: D1Database, store: VectorStore, kind: "page" | "folio" | null, id: string, workspaceId: string, ids: string[]): Promise<void> {
611 if (!ids.length || !(await erasedSource(db, kind, id, workspaceId))) return;
612 await store.delete(ids);
613 await db.batch([
614 db.prepare(`DELETE FROM ${kind === "folio" ? "folio_chunks_fts WHERE folio_id" : "doc_chunks_fts WHERE doc_id"} = ?`).bind(id),
615 ]);
616}