1use std::{
2 collections::{BTreeMap, BTreeSet},
3 path::Path,
4 sync::{Arc, Mutex},
5 time::Duration,
6};
7
8mod code;
9
10use rusqlite::{Connection, OptionalExtension, params};
11
12mod canvas;
13mod canvas_code;
14mod code_graph;
15mod file_index;
16mod helpers;
17mod indexing;
18mod operations;
19mod retrieval;
20mod retry;
21mod schema_columns;
22mod schema_migration;
23mod store_impls;
24
25use crate::{
26 domain::{CommitReceipt, GraphMutationBatch, GraphVersion, SourceScope},
27 storage::{GraphInspection, StorageError, StorageFuture},
28};
29
30#[cfg(test)]
31use crate::{
32 domain::IndexKind,
33 storage::{
34 CodeGraphStore, GraphSearchRequest, GraphStore, IndexStore, MutationLogStore, RetrievalHit,
35 },
36};
37use helpers::{count_rows, source_hash_for_evidence, stable_id, storage_version_range};
38
39const SQLITE_BUSY_TIMEOUT: Duration = Duration::from_secs(30);
40
41#[derive(Debug, Clone)]
43pub struct SqliteGraphStore {
44 connection: Arc<Mutex<Connection>>,
45}
46
47impl SqliteGraphStore {
48 pub fn open(path: impl AsRef<Path>) -> Result<Self, StorageError> {
50 let path = path.as_ref().to_path_buf();
51 if let Some(parent) = path.parent() {
52 std::fs::create_dir_all(parent)?;
53 }
54
55 let connection = Connection::open(&path)?;
56 configure_connection(&connection)?;
57 schema_migration::prepare_existing_database(&connection)?;
58 initialize_schema(&connection)?;
59
60 Ok(Self {
61 connection: Arc::new(Mutex::new(connection)),
62 })
63 }
64
65 pub fn open_in_memory() -> Result<Self, StorageError> {
67 let connection = Connection::open_in_memory()?;
68 configure_connection(&connection)?;
69 initialize_schema(&connection)?;
70
71 Ok(Self {
72 connection: Arc::new(Mutex::new(connection)),
73 })
74 }
75
76 pub(super) fn run<T, F>(&self, operation: F) -> StorageFuture<'_, T>
77 where
78 T: Send + 'static,
79 F: FnOnce(&mut Connection) -> Result<T, StorageError> + Send + 'static,
80 {
81 let connection = Arc::clone(&self.connection);
82
83 Box::pin(async move {
84 tokio::task::spawn_blocking(move || {
85 let mut guard = connection.lock().map_err(|_| StorageError::LockPoisoned)?;
86
87 operation(&mut guard)
88 })
89 .await?
90 })
91 }
92}
93
94fn configure_connection(connection: &Connection) -> Result<(), StorageError> {
95 connection.busy_timeout(SQLITE_BUSY_TIMEOUT)?;
96
97 Ok(())
98}
99
100fn initialize_schema(connection: &Connection) -> Result<(), StorageError> {
101 retry::retry_sqlite_transient(|| initialize_schema_once(connection))
102}
103
104fn initialize_schema_once(connection: &Connection) -> Result<(), StorageError> {
105 connection.execute_batch(
106 "
107 PRAGMA foreign_keys = ON;
108 PRAGMA journal_mode = WAL;
109
110 CREATE TABLE IF NOT EXISTS graph_state (
111 id INTEGER PRIMARY KEY CHECK (id = 1),
112 graph_version INTEGER NOT NULL
113 );
114
115 INSERT OR IGNORE INTO graph_state (id, graph_version) VALUES (1, 0);
116
117 CREATE TABLE IF NOT EXISTS entities (
118 id TEXT PRIMARY KEY,
119 label TEXT NOT NULL,
120 created_graph_version INTEGER NOT NULL
121 );
122
123 CREATE TABLE IF NOT EXISTS evidence (
124 id TEXT PRIMARY KEY,
125 source_scope TEXT NOT NULL,
126 source_path TEXT,
127 span_start_byte INTEGER,
128 span_end_byte INTEGER,
129 span_start_line INTEGER,
130 span_end_line INTEGER,
131 content TEXT NOT NULL,
132 confidence_basis_points INTEGER NOT NULL DEFAULT 10000,
133 status TEXT NOT NULL DEFAULT 'accepted',
134 modality TEXT NOT NULL DEFAULT 'text_span',
135 source_uri TEXT,
136 source_hash TEXT,
137 media_hash TEXT,
138 extractor TEXT,
139 extractor_version TEXT,
140 observed_at TEXT,
141 parent_evidence_id TEXT,
142 layout_page_number INTEGER,
143 layout_x INTEGER,
144 layout_y INTEGER,
145 layout_width INTEGER,
146 layout_height INTEGER,
147 embedding_model TEXT,
148 embedding_dimension INTEGER,
149 extraction_status TEXT NOT NULL DEFAULT 'succeeded',
150 extraction_message TEXT,
151 created_graph_version INTEGER NOT NULL
152 );
153
154 CREATE TABLE IF NOT EXISTS evidence_entities (
155 evidence_id TEXT NOT NULL,
156 entity_id TEXT NOT NULL,
157 PRIMARY KEY (evidence_id, entity_id),
158 FOREIGN KEY (evidence_id) REFERENCES evidence(id) ON DELETE CASCADE,
159 FOREIGN KEY (entity_id) REFERENCES entities(id) ON DELETE CASCADE
160 );
161
162 CREATE TABLE IF NOT EXISTS graph_mutations (
163 graph_version INTEGER PRIMARY KEY,
164 evidence_count INTEGER NOT NULL,
165 entity_count INTEGER NOT NULL,
166 relation_count INTEGER NOT NULL DEFAULT 0,
167 claim_count INTEGER NOT NULL DEFAULT 0,
168 event_count INTEGER NOT NULL DEFAULT 0,
169 affected_scopes_json TEXT NOT NULL DEFAULT '[]',
170 affected_entity_ids_json TEXT NOT NULL DEFAULT '[]',
171 evidence_ids_json TEXT NOT NULL DEFAULT '[]',
172 source_hashes_json TEXT NOT NULL DEFAULT '[]'
173 );
174
175 CREATE TABLE IF NOT EXISTS graph_relations (
176 id TEXT PRIMARY KEY,
177 source_entity_id TEXT NOT NULL,
178 relation_type TEXT NOT NULL,
179 target_entity_id TEXT NOT NULL,
180 evidence_ids_json TEXT NOT NULL,
181 confidence_basis_points INTEGER NOT NULL,
182 status TEXT NOT NULL,
183 valid_from_graph_version INTEGER NOT NULL,
184 valid_until_graph_version INTEGER,
185 created_graph_version INTEGER NOT NULL,
186 FOREIGN KEY (source_entity_id) REFERENCES entities(id),
187 FOREIGN KEY (target_entity_id) REFERENCES entities(id)
188 );
189
190 CREATE TABLE IF NOT EXISTS graph_claims (
191 id TEXT PRIMARY KEY,
192 subject_entity_id TEXT NOT NULL,
193 predicate TEXT NOT NULL,
194 object TEXT NOT NULL,
195 evidence_ids_json TEXT NOT NULL,
196 confidence_basis_points INTEGER NOT NULL,
197 status TEXT NOT NULL,
198 valid_from_graph_version INTEGER NOT NULL,
199 valid_until_graph_version INTEGER,
200 created_graph_version INTEGER NOT NULL,
201 FOREIGN KEY (subject_entity_id) REFERENCES entities(id)
202 );
203
204 CREATE TABLE IF NOT EXISTS graph_events (
205 id TEXT PRIMARY KEY,
206 event_type TEXT NOT NULL,
207 occurred_at TEXT,
208 evidence_ids_json TEXT NOT NULL,
209 confidence_basis_points INTEGER NOT NULL,
210 status TEXT NOT NULL,
211 valid_from_graph_version INTEGER NOT NULL,
212 valid_until_graph_version INTEGER,
213 created_graph_version INTEGER NOT NULL
214 );
215
216 CREATE TABLE IF NOT EXISTS graph_event_entities (
217 event_id TEXT NOT NULL,
218 entity_id TEXT NOT NULL,
219 PRIMARY KEY (event_id, entity_id),
220 FOREIGN KEY (event_id) REFERENCES graph_events(id) ON DELETE CASCADE,
221 FOREIGN KEY (entity_id) REFERENCES entities(id)
222 );
223
224 CREATE TABLE IF NOT EXISTS graph_fact_evidence (
225 fact_kind TEXT NOT NULL,
226 fact_id TEXT NOT NULL,
227 evidence_id TEXT NOT NULL,
228 PRIMARY KEY (fact_kind, fact_id, evidence_id),
229 FOREIGN KEY (evidence_id) REFERENCES evidence(id) ON DELETE CASCADE
230 );
231
232 CREATE INDEX IF NOT EXISTS graph_fact_evidence_by_evidence
233 ON graph_fact_evidence(evidence_id, fact_kind);
234 ",
235 )?;
236 schema_columns::ensure_core_schema_columns(connection)?;
237 code::initialize_code_schema(connection)?;
238 indexing::initialize_schema(connection)?;
239 code_graph::initialize_schema(connection)?;
240 operations::initialize_schema(connection)?;
241 file_index::initialize_schema(connection)?;
242 backfill_fact_evidence_links(connection)?;
243 retrieval::initialize_schema(connection)?;
244
245 Ok(())
246}
247
248fn backfill_fact_evidence_links(connection: &Connection) -> Result<(), StorageError> {
249 backfill_fact_evidence_kind(connection, "relation", "graph_relations")?;
250 backfill_fact_evidence_kind(connection, "claim", "graph_claims")?;
251 backfill_fact_evidence_kind(connection, "event", "graph_events")?;
252
253 Ok(())
254}
255
256fn backfill_fact_evidence_kind(
257 connection: &Connection,
258 fact_kind: &'static str,
259 table: &'static str,
260) -> Result<(), StorageError> {
261 let mut statement =
262 connection.prepare(&format!("SELECT id, evidence_ids_json FROM {table}"))?;
263 let rows = statement.query_map([], |row| {
264 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
265 })?;
266 let facts = rows
267 .collect::<Result<Vec<_>, _>>()
268 .map_err(StorageError::from)?;
269 drop(statement);
270
271 for (fact_id, evidence_json) in facts {
272 let evidence_ids: Vec<String> = serde_json::from_str(&evidence_json)
273 .map_err(|error| StorageError::InvalidInput(error.to_string()))?;
274 for evidence_id in evidence_ids {
275 connection.execute(
276 "
277 INSERT OR IGNORE INTO graph_fact_evidence (fact_kind, fact_id, evidence_id)
278 SELECT ?1, ?2, e.id
279 FROM evidence e
280 WHERE e.id = ?3
281 ",
282 params![fact_kind, fact_id, evidence_id],
283 )?;
284 }
285 }
286
287 Ok(())
288}
289
290fn commit_batch(
291 connection: &mut Connection,
292 batch: GraphMutationBatch,
293) -> Result<CommitReceipt, StorageError> {
294 let transaction = connection.transaction()?;
295 let current = current_graph_version_in_transaction(&transaction)?;
296 let next = GraphVersion::new(current.get() + 1);
297 let evidence_count = batch.evidence.len();
298 let relation_count = batch.relations.len();
299 let claim_count = batch.claims.len();
300 let event_count = batch.events.len();
301 let mut affected_entity_ids = BTreeSet::new();
302 let mut affected_scopes = BTreeSet::new();
303 let mut evidence_ids = BTreeSet::new();
304 let mut source_hashes = BTreeSet::new();
305 let batch_evidence_scopes = batch
306 .evidence
307 .iter()
308 .map(|evidence| {
309 (
310 evidence.id.clone(),
311 evidence.source_scope.as_str().to_owned(),
312 )
313 })
314 .collect::<BTreeMap<_, _>>();
315
316 for evidence in batch.evidence {
317 let evidence_id = evidence.id;
318 let source_scope = evidence.source_scope;
319 let source_scope_text = source_scope.as_str().to_owned();
320 let source_path = evidence.source_path;
321 let span = evidence.span;
322 let content = evidence.content;
323 let entity_labels = evidence.entity_labels;
324 let extraction = evidence.extraction;
325 let derived_source_hash = source_hash_for_evidence(
326 &extraction,
327 &source_scope_text,
328 source_path.as_deref(),
329 &content,
330 );
331 if let Some(previous_scope) = evidence_scope(&transaction, &evidence_id)? {
332 affected_scopes.insert(previous_scope);
333 }
334 affected_scopes.insert(source_scope_text.clone());
335 evidence_ids.insert(evidence_id.clone());
336 source_hashes.insert(derived_source_hash.clone());
337 if let Some(media_hash) = &extraction.media_hash {
338 source_hashes.insert(media_hash.clone());
339 }
340 if let Some(parent_evidence_id) = extraction.parent_evidence_id.as_deref() {
341 validate_parent_evidence(
342 &transaction,
343 &batch_evidence_scopes,
344 &evidence_id,
345 &source_scope_text,
346 parent_evidence_id,
347 )?;
348 }
349 let layout_region = extraction.layout_region;
350 transaction.execute(
351 "INSERT INTO evidence (
352 id, source_scope, source_path, span_start_byte, span_end_byte,
353 span_start_line, span_end_line, content, confidence_basis_points,
354 status, modality, source_uri, source_hash, media_hash, extractor,
355 extractor_version, observed_at, parent_evidence_id, layout_page_number,
356 layout_x, layout_y, layout_width, layout_height, embedding_model,
357 embedding_dimension, extraction_status, extraction_message, created_graph_version
358 )
359 VALUES (
360 ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14,
361 ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26,
362 ?27, ?28
363 )
364 ON CONFLICT(id) DO UPDATE SET
365 source_scope = excluded.source_scope,
366 source_path = excluded.source_path,
367 span_start_byte = excluded.span_start_byte,
368 span_end_byte = excluded.span_end_byte,
369 span_start_line = excluded.span_start_line,
370 span_end_line = excluded.span_end_line,
371 content = excluded.content,
372 confidence_basis_points = excluded.confidence_basis_points,
373 status = excluded.status,
374 modality = excluded.modality,
375 source_uri = excluded.source_uri,
376 source_hash = excluded.source_hash,
377 media_hash = excluded.media_hash,
378 extractor = excluded.extractor,
379 extractor_version = excluded.extractor_version,
380 observed_at = excluded.observed_at,
381 parent_evidence_id = excluded.parent_evidence_id,
382 layout_page_number = excluded.layout_page_number,
383 layout_x = excluded.layout_x,
384 layout_y = excluded.layout_y,
385 layout_width = excluded.layout_width,
386 layout_height = excluded.layout_height,
387 embedding_model = excluded.embedding_model,
388 embedding_dimension = excluded.embedding_dimension,
389 extraction_status = excluded.extraction_status,
390 extraction_message = excluded.extraction_message,
391 created_graph_version = excluded.created_graph_version",
392 params![
393 &evidence_id,
394 &source_scope_text,
395 source_path.as_deref(),
396 span.map(|value| value.start_byte),
397 span.map(|value| value.end_byte),
398 span.map(|value| value.start_line),
399 span.map(|value| value.end_line),
400 &content,
401 evidence.confidence.basis_points,
402 evidence.status.as_str(),
403 extraction.modality.as_str(),
404 extraction.source_uri.as_deref(),
405 &derived_source_hash,
406 extraction.media_hash.as_deref(),
407 extraction.extractor.as_deref(),
408 extraction.extractor_version.as_deref(),
409 extraction.observed_at.as_deref(),
410 extraction.parent_evidence_id.as_deref(),
411 layout_region.map(|region| region.page_number),
412 layout_region.map(|region| region.x),
413 layout_region.map(|region| region.y),
414 layout_region.map(|region| region.width),
415 layout_region.map(|region| region.height),
416 extraction.embedding_model.as_deref(),
417 extraction.embedding_dimension.map(i64::from),
418 extraction.diagnostic.status.as_str(),
419 extraction.diagnostic.message.as_deref(),
420 next.get()
421 ],
422 )?;
423
424 transaction.execute(
425 "DELETE FROM evidence_entities WHERE evidence_id = ?1",
426 params![&evidence_id],
427 )?;
428
429 for label in &entity_labels {
430 let entity_id = upsert_entity(&transaction, label, next)?;
431 transaction.execute(
432 "INSERT OR IGNORE INTO evidence_entities (evidence_id, entity_id)
433 VALUES (?1, ?2)",
434 params![evidence_id, entity_id],
435 )?;
436 affected_entity_ids.insert(entity_id);
437 }
438 retrieval::replace_evidence_document(
439 &transaction,
440 retrieval::EvidenceDocumentInput {
441 evidence_id: &evidence_id,
442 source_scope: &source_scope_text,
443 source_path: source_path.as_deref(),
444 entity_labels: &entity_labels,
445 content: &content,
446 status: evidence.status,
447 extraction: &extraction,
448 source_hash: &derived_source_hash,
449 graph_version: next.get(),
450 },
451 )?;
452 }
453
454 for relation in batch.relations {
455 validate_evidence_references(&transaction, &relation.source_scope, &relation.evidence_ids)?;
456 evidence_ids.extend(relation.evidence_ids.iter().cloned());
457 let source_entity_id = upsert_entity(&transaction, &relation.source_entity_label, next)?;
458 let target_entity_id = upsert_entity(&transaction, &relation.target_entity_label, next)?;
459 let version_range = storage_version_range(relation.version_range, next);
460 affected_entity_ids.insert(source_entity_id.clone());
461 affected_entity_ids.insert(target_entity_id.clone());
462 transaction.execute(
463 "
464 INSERT INTO graph_relations (
465 id, source_entity_id, relation_type, target_entity_id,
466 evidence_ids_json, confidence_basis_points, status,
467 valid_from_graph_version, valid_until_graph_version, created_graph_version
468 )
469 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
470 ON CONFLICT(id) DO UPDATE SET
471 source_entity_id = excluded.source_entity_id,
472 relation_type = excluded.relation_type,
473 target_entity_id = excluded.target_entity_id,
474 evidence_ids_json = excluded.evidence_ids_json,
475 confidence_basis_points = excluded.confidence_basis_points,
476 status = excluded.status,
477 valid_from_graph_version = excluded.valid_from_graph_version,
478 valid_until_graph_version = excluded.valid_until_graph_version,
479 created_graph_version = excluded.created_graph_version
480 ",
481 params![
482 relation.id.as_str(),
483 source_entity_id,
484 relation.relation_type,
485 target_entity_id,
486 evidence_ids_json(&relation.evidence_ids)?,
487 relation.confidence.basis_points,
488 relation.status.as_str(),
489 version_range.valid_from.get(),
490 version_range.valid_until.map(GraphVersion::get),
491 next.get(),
492 ],
493 )?;
494 replace_fact_evidence_links(
495 &transaction,
496 "relation",
497 &relation.id,
498 &relation.evidence_ids,
499 )?;
500 }
501
502 for claim in batch.claims {
503 validate_evidence_references(&transaction, &claim.source_scope, &claim.evidence_ids)?;
504 evidence_ids.extend(claim.evidence_ids.iter().cloned());
505 let subject_entity_id = upsert_entity(&transaction, &claim.subject_entity_label, next)?;
506 let version_range = storage_version_range(claim.version_range, next);
507 affected_entity_ids.insert(subject_entity_id.clone());
508 transaction.execute(
509 "
510 INSERT INTO graph_claims (
511 id, subject_entity_id, predicate, object, evidence_ids_json,
512 confidence_basis_points, status, valid_from_graph_version,
513 valid_until_graph_version, created_graph_version
514 )
515 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
516 ON CONFLICT(id) DO UPDATE SET
517 subject_entity_id = excluded.subject_entity_id,
518 predicate = excluded.predicate,
519 object = excluded.object,
520 evidence_ids_json = excluded.evidence_ids_json,
521 confidence_basis_points = excluded.confidence_basis_points,
522 status = excluded.status,
523 valid_from_graph_version = excluded.valid_from_graph_version,
524 valid_until_graph_version = excluded.valid_until_graph_version,
525 created_graph_version = excluded.created_graph_version
526 ",
527 params![
528 claim.id.as_str(),
529 subject_entity_id,
530 claim.predicate,
531 claim.object,
532 evidence_ids_json(&claim.evidence_ids)?,
533 claim.confidence.basis_points,
534 claim.status.as_str(),
535 version_range.valid_from.get(),
536 version_range.valid_until.map(GraphVersion::get),
537 next.get(),
538 ],
539 )?;
540 replace_fact_evidence_links(&transaction, "claim", &claim.id, &claim.evidence_ids)?;
541 }
542
543 for event in batch.events {
544 validate_evidence_references(&transaction, &event.source_scope, &event.evidence_ids)?;
545 evidence_ids.extend(event.evidence_ids.iter().cloned());
546 let version_range = storage_version_range(event.version_range, next);
547 transaction.execute(
548 "
549 INSERT INTO graph_events (
550 id, event_type, occurred_at, evidence_ids_json,
551 confidence_basis_points, status, valid_from_graph_version,
552 valid_until_graph_version, created_graph_version
553 )
554 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
555 ON CONFLICT(id) DO UPDATE SET
556 event_type = excluded.event_type,
557 occurred_at = excluded.occurred_at,
558 evidence_ids_json = excluded.evidence_ids_json,
559 confidence_basis_points = excluded.confidence_basis_points,
560 status = excluded.status,
561 valid_from_graph_version = excluded.valid_from_graph_version,
562 valid_until_graph_version = excluded.valid_until_graph_version,
563 created_graph_version = excluded.created_graph_version
564 ",
565 params![
566 event.id.as_str(),
567 event.event_type,
568 event.occurred_at,
569 evidence_ids_json(&event.evidence_ids)?,
570 event.confidence.basis_points,
571 event.status.as_str(),
572 version_range.valid_from.get(),
573 version_range.valid_until.map(GraphVersion::get),
574 next.get(),
575 ],
576 )?;
577 replace_fact_evidence_links(&transaction, "event", &event.id, &event.evidence_ids)?;
578 transaction.execute(
579 "DELETE FROM graph_event_entities WHERE event_id = ?1",
580 params![event.id],
581 )?;
582 for label in event.entity_labels {
583 let entity_id = upsert_entity(&transaction, &label, next)?;
584 affected_entity_ids.insert(entity_id.clone());
585 transaction.execute(
586 "INSERT OR IGNORE INTO graph_event_entities (event_id, entity_id)
587 VALUES (?1, ?2)",
588 params![event.id, entity_id],
589 )?;
590 }
591 }
592
593 transaction.execute(
594 "
595 DELETE FROM entities
596 WHERE id NOT IN (SELECT entity_id FROM evidence_entities)
597 AND id NOT IN (SELECT source_entity_id FROM graph_relations)
598 AND id NOT IN (SELECT target_entity_id FROM graph_relations)
599 AND id NOT IN (SELECT subject_entity_id FROM graph_claims)
600 AND id NOT IN (SELECT entity_id FROM graph_event_entities)
601 ",
602 [],
603 )?;
604
605 let entity_count = affected_entity_ids.len();
606 add_scopes_for_evidence_ids(&transaction, &evidence_ids, &mut affected_scopes)?;
607 if affected_scopes.is_empty()
608 && (evidence_count > 0 || relation_count > 0 || claim_count > 0 || event_count > 0)
609 {
610 affected_scopes.insert(indexing::DEFAULT_SCOPE.to_owned());
611 }
612 let affected_scopes = affected_scopes.into_iter().collect::<Vec<_>>();
613 let affected_entity_ids = affected_entity_ids.into_iter().collect::<Vec<_>>();
614 let affected_scopes_json = indexing::json_array(affected_scopes.clone())?;
615 let affected_entity_ids_json = indexing::json_array(affected_entity_ids.clone())?;
616 let evidence_ids_json = indexing::json_array(evidence_ids)?;
617 let source_hashes_json = indexing::json_array(source_hashes)?;
618 transaction.execute(
619 "INSERT INTO graph_mutations (
620 graph_version, evidence_count, entity_count, relation_count, claim_count, event_count,
621 affected_scopes_json, affected_entity_ids_json, evidence_ids_json, source_hashes_json
622 )
623 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
624 params![
625 next.get(),
626 evidence_count,
627 entity_count,
628 relation_count,
629 claim_count,
630 event_count,
631 affected_scopes_json,
632 affected_entity_ids_json,
633 evidence_ids_json,
634 source_hashes_json
635 ],
636 )?;
637 indexing::mark_mutation_cursors_stale(&transaction, &affected_scopes)?;
638 transaction.execute(
639 "UPDATE graph_state SET graph_version = ?1 WHERE id = 1",
640 params![next.get()],
641 )?;
642 transaction.execute("UPDATE index_status SET state = 'stale'", [])?;
643 transaction.commit()?;
644
645 Ok(CommitReceipt {
646 graph_version: next,
647 evidence_count,
648 entity_count,
649 relation_count,
650 claim_count,
651 event_count,
652 })
653}
654
655fn inspect_graph(connection: &mut Connection) -> Result<GraphInspection, StorageError> {
656 Ok(GraphInspection {
657 graph_version: current_graph_version(connection)?,
658 entity_count: count_rows(connection, "entities")?,
659 evidence_count: count_rows(connection, "evidence")?,
660 relation_count: count_rows(connection, "graph_relations")?,
661 claim_count: count_rows(connection, "graph_claims")?,
662 event_count: count_rows(connection, "graph_events")?,
663 mutation_count: count_rows(connection, "graph_mutations")?,
664 code_file_count: count_rows(connection, "code_files")?,
665 code_symbol_count: count_rows(connection, "code_symbols")?,
666 code_reference_count: count_rows(connection, "code_references")?,
667 code_chunk_count: count_rows(connection, "code_chunks")?,
668 code_parse_status_counts: code_graph::parse_status_counts(connection)?,
669 })
670}
671
672fn current_graph_version(connection: &mut Connection) -> Result<GraphVersion, StorageError> {
673 current_graph_version_in_transaction(connection)
674}
675
676fn current_graph_version_in_transaction(
677 connection: &Connection,
678) -> Result<GraphVersion, StorageError> {
679 let value = connection.query_row(
680 "SELECT graph_version FROM graph_state WHERE id = 1",
681 [],
682 |row| row.get::<_, u64>(0),
683 )?;
684
685 Ok(GraphVersion::new(value))
686}
687
688fn upsert_entity(
689 transaction: &rusqlite::Transaction<'_>,
690 label: &str,
691 graph_version: GraphVersion,
692) -> Result<String, StorageError> {
693 let entity_id = stable_id("entity", label);
694 transaction.execute(
695 "INSERT OR IGNORE INTO entities (id, label, created_graph_version)
696 VALUES (?1, ?2, ?3)",
697 params![entity_id, label, graph_version.get()],
698 )?;
699
700 Ok(entity_id)
701}
702
703fn add_scopes_for_evidence_ids(
704 connection: &Connection,
705 evidence_ids: &BTreeSet<String>,
706 affected_scopes: &mut BTreeSet<String>,
707) -> Result<(), StorageError> {
708 for evidence_id in evidence_ids {
709 if let Some(scope) = evidence_scope(connection, evidence_id)? {
710 affected_scopes.insert(scope);
711 }
712 }
713
714 Ok(())
715}
716
717fn evidence_scope(
718 connection: &Connection,
719 evidence_id: &str,
720) -> Result<Option<String>, StorageError> {
721 connection
722 .query_row(
723 "SELECT source_scope FROM evidence WHERE id = ?1",
724 params![evidence_id],
725 |row| row.get::<_, String>(0),
726 )
727 .optional()
728 .map_err(StorageError::from)
729}
730
731fn validate_parent_evidence(
732 connection: &Connection,
733 batch_evidence_scopes: &BTreeMap<String, String>,
734 evidence_id: &str,
735 source_scope: &str,
736 parent_evidence_id: &str,
737) -> Result<(), StorageError> {
738 if parent_evidence_id == evidence_id {
739 return Err(StorageError::InvalidInput(
740 "parent evidence id must reference a different evidence record".to_owned(),
741 ));
742 }
743 let parent_scope = if let Some(scope) = batch_evidence_scopes.get(parent_evidence_id) {
744 Some(scope.clone())
745 } else {
746 evidence_scope(connection, parent_evidence_id).map_err(|error| {
747 StorageError::InvalidInput(format!(
748 "parent evidence id '{parent_evidence_id}' could not be validated: {error}"
749 ))
750 })?
751 };
752
753 match parent_scope {
754 Some(parent_scope) if parent_scope == source_scope => Ok(()),
755 Some(parent_scope) => Err(StorageError::InvalidInput(format!(
756 "parent evidence id '{parent_evidence_id}' belongs to source scope \
757 '{parent_scope}' instead of '{source_scope}'"
758 ))),
759 None => Err(StorageError::InvalidInput(format!(
760 "parent evidence id '{parent_evidence_id}' does not exist in source scope \
761 '{source_scope}'"
762 ))),
763 }
764}
765
766fn evidence_ids_json(evidence_ids: &[String]) -> Result<String, StorageError> {
767 serde_json::to_string(evidence_ids)
768 .map_err(|error| StorageError::InvalidInput(error.to_string()))
769}
770
771fn validate_evidence_references(
772 transaction: &rusqlite::Transaction<'_>,
773 source_scope: &SourceScope,
774 evidence_ids: &[String],
775) -> Result<(), StorageError> {
776 for evidence_id in evidence_ids {
777 let actual_scope = transaction
778 .query_row(
779 "SELECT source_scope FROM evidence WHERE id = ?1",
780 params![evidence_id],
781 |row| row.get::<_, String>(0),
782 )
783 .optional()?;
784 let Some(actual_scope) = actual_scope else {
785 return Err(StorageError::InvalidInput(format!(
786 "structured fact references unknown evidence id '{evidence_id}'"
787 )));
788 };
789 if actual_scope != source_scope.as_str() {
790 return Err(StorageError::InvalidInput(format!(
791 "structured fact references evidence id '{evidence_id}' from source scope \
792 '{actual_scope}' instead of '{}'",
793 source_scope.as_str()
794 )));
795 }
796 }
797
798 Ok(())
799}
800
801fn replace_fact_evidence_links(
802 transaction: &rusqlite::Transaction<'_>,
803 fact_kind: &'static str,
804 fact_id: &str,
805 evidence_ids: &[String],
806) -> Result<(), StorageError> {
807 transaction.execute(
808 "DELETE FROM graph_fact_evidence WHERE fact_kind = ?1 AND fact_id = ?2",
809 params![fact_kind, fact_id],
810 )?;
811 for evidence_id in evidence_ids {
812 transaction.execute(
813 "INSERT OR IGNORE INTO graph_fact_evidence (fact_kind, fact_id, evidence_id)
814 VALUES (?1, ?2, ?3)",
815 params![fact_kind, fact_id, evidence_id],
816 )?;
817 }
818
819 Ok(())
820}
821
822#[cfg(test)]
823mod metadata_tests;
824
825#[cfg(test)]
826mod graph_tests;
827
828#[cfg(test)]
829mod index_refresh_queue_tests;
830
831#[cfg(test)]
832mod index_schema_migration_tests;
833
834#[cfg(test)]
835mod graphrag_phase4_tests;
836
837#[cfg(test)]
838mod index_refresh_tests;
839
840#[cfg(test)]
841mod operations_tests;