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