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