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