sbol_db_sqlite/
store.rs

1//! The SQLite-backed SBOL store: ingest plus the derived-view read surface,
2//! the SPARQL read/write adapters, and the storage-trait implementations
3//! (objects, graphs, ontology, neighborhood, sequence search, and the lab
4//! dashboard). The job queue lives in [`crate::repo::job`].
5
6use std::collections::HashSet;
7use std::sync::Arc;
8
9use async_trait::async_trait;
10use sbol_db_core::{
11    DomainError, GraphId, GraphRecord, ImportReport, IriString, NeighborhoodQuery,
12    NeighborhoodResult, ObjectId, SbolObjectRecord, SerializationFormat, Triple,
13};
14use sbol_db_derive::{build_import_plan, compose_merged_input, to_rdf_format};
15use sbol_db_rdf::{rdf_graph_to_triples, GRAPH_IRI_PREFIX};
16use sbol_db_storage::{
17    distinct_graph_iris, distinct_object_iris, AccelSolutions, AcceleratedQuery, AclStore,
18    BatchSequenceMatch, CatalogSequenceRecord, ClassCount, CorpusCounts, CorpusStats,
19    CorpusStatsStore, CursorPage, GraphFilter, GraphOverview, GraphStore, GraphTriplesPage,
20    GraphWriteMode, ImportInput, ImportOverwrite, LabStore, ListGraphsFilter, ListObjectsFilter,
21    NamedGraphCatalogStore, NamedGraphQuery, NamedGraphRecord, NeighborhoodStore, ObjectStore,
22    OntologyLoadReport, OntologyRecord, OntologyStore, OntologyTermRecord, PatternObject,
23    PatternSubject, ResourceCatalogStore, ResourceOccurrence, ResourceQuery, ResourceRecord,
24    SbolStore, SequenceCatalogStore, SequenceMatch, SequenceQuery, SequenceSearchOptions,
25    SequenceSearchStore, TextSearchQuery, TextSearchStore, TripleChange, TriplePageQuery,
26    TripleScanPage, TripleSource, TripleWriter, UpdateOutcome, SBH_CAN_VIEW, SBH_OWNED_BY,
27};
28use tokio::runtime::Handle;
29
30use crate::pool::db_err;
31use crate::repo::{
32    AccelRepository, GraphRepository, LabRepository, OntologyRepository, SbolObjectRepository,
33    SequenceSearchRepository, TripleRepository,
34};
35use crate::SqlitePool;
36
37/// Per-call cap on a Graph Store `GET`, matching the Postgres backend.
38const GRAPH_READ_LIMIT: i64 = 5_000_000;
39const UPDATE_SOURCE: &str = "sparql-update";
40
41/// The SQLite SBOL store. Cloneable; all clones share one connection pool.
42#[derive(Clone)]
43pub struct SqliteStore {
44    pool: SqlitePool,
45    graphs: GraphRepository,
46    objects: SbolObjectRepository,
47    triples: TripleRepository,
48    accel: AccelRepository,
49    ontology: OntologyRepository,
50    sequences: SequenceSearchRepository,
51    lab: LabRepository,
52}
53
54impl SqliteStore {
55    pub fn new(pool: SqlitePool) -> Self {
56        Self {
57            graphs: GraphRepository::new(pool.clone()),
58            objects: SbolObjectRepository::new(pool.clone()),
59            triples: TripleRepository::new(pool.clone()),
60            accel: AccelRepository::new(pool.clone(), TripleRepository::new(pool.clone())),
61            ontology: OntologyRepository::new(pool.clone()),
62            sequences: SequenceSearchRepository::new(pool.clone()),
63            lab: LabRepository::new(pool.clone()),
64            pool,
65        }
66    }
67
68    pub fn pool(&self) -> &SqlitePool {
69        &self.pool
70    }
71
72    pub async fn ping(&self) -> Result<(), DomainError> {
73        sqlx::query("SELECT 1")
74            .execute(&self.pool)
75            .await
76            .map(|_| ())
77            .map_err(db_err)
78    }
79
80    pub fn triple_source(&self) -> Arc<dyn TripleSource> {
81        Arc::new(SqliteTripleSource {
82            triples: Arc::new(self.triples.clone()),
83            accel: self.accel.clone(),
84        })
85    }
86
87    pub fn triple_writer(&self) -> Arc<dyn TripleWriter> {
88        Arc::new(SqliteTripleWriter {
89            triples: Arc::new(self.triples.clone()),
90            accel: self.accel.clone(),
91            pool: self.pool.clone(),
92        })
93    }
94
95    pub async fn import_document(&self, input: ImportInput) -> Result<ImportReport, DomainError> {
96        let input = self.resolve_merge(input).await?;
97        let mut tx = self.pool.begin().await.map_err(db_err)?;
98        let report = self.import_into_conn(&mut tx, input).await?;
99        tx.commit().await.map_err(db_err)?;
100        Ok(report)
101    }
102
103    /// Fold an existing graph's triples into a merge import, yielding a
104    /// `Replace` input carrying the union. Non-merge inputs pass through.
105    async fn resolve_merge(&self, input: ImportInput) -> Result<ImportInput, DomainError> {
106        if input.overwrite != ImportOverwrite::Merge {
107            return Ok(input);
108        }
109        let Some(document_iri) = input.document_iri.clone() else {
110            return Ok(input);
111        };
112        match self
113            .graphs
114            .id_by_document_iri(document_iri.as_str())
115            .await?
116        {
117            Some(old_id) => {
118                let old_graph_iri = format!("{GRAPH_IRI_PREFIX}{}", old_id.as_uuid());
119                let old_triples = self
120                    .triples
121                    .triples_for_graph(Some(&old_graph_iri), GRAPH_READ_LIMIT)
122                    .await?;
123                compose_merged_input(&old_triples, &input)
124            }
125            None => Ok(ImportInput {
126                overwrite: ImportOverwrite::Replace,
127                ..input
128            }),
129        }
130    }
131
132    pub async fn import_documents(
133        &self,
134        inputs: Vec<ImportInput>,
135    ) -> Result<Vec<ImportReport>, DomainError> {
136        if inputs.is_empty() {
137            return Ok(Vec::new());
138        }
139        let mut tx = self.pool.begin().await.map_err(db_err)?;
140        let mut reports = Vec::with_capacity(inputs.len());
141        for input in inputs {
142            reports.push(self.import_into_conn(&mut tx, input).await?);
143        }
144        tx.commit().await.map_err(db_err)?;
145        Ok(reports)
146    }
147
148    async fn import_into_conn(
149        &self,
150        conn: &mut sqlx::SqliteConnection,
151        input: ImportInput,
152    ) -> Result<ImportReport, DomainError> {
153        if input.overwrite == ImportOverwrite::Replace {
154            if let Some(document_iri) = &input.document_iri {
155                if let Some(old_id) = self
156                    .graphs
157                    .id_by_document_iri_in(&mut *conn, document_iri.as_str())
158                    .await?
159                {
160                    self.graphs.delete_in(&mut *conn, old_id).await?;
161                }
162            }
163        }
164
165        let plan = build_import_plan(&input)?;
166
167        self.graphs
168            .insert(&mut *conn, plan.graph_id, plan.new_graph)
169            .await?;
170        let triple_count = self
171            .triples
172            .insert_triples(&mut *conn, &plan.triples, "sbol")
173            .await?;
174        self.accel
175            .refresh_graph(&mut *conn, plan.graph_iri.as_str())
176            .await?;
177        let object_count = plan.summaries.len();
178        for summary in &plan.summaries {
179            self.objects
180                .upsert(&mut *conn, summary, Some(plan.graph_id))
181                .await?;
182        }
183
184        // Persist sequence projections so sequence search has data + a k-mer
185        // index. (Other typed projections are not yet materialized in SQLite.)
186        for sequence in &plan.projections.sequences {
187            self.sequences.upsert_sequence(&mut *conn, sequence).await?;
188        }
189
190        Ok(ImportReport {
191            graph_id: plan.graph_id,
192            object_count,
193            triple_count,
194            validation_status: plan.validation_status,
195            validation_issue_count: plan.validation_issue_count,
196        })
197    }
198
199    pub async fn graph_store_write(
200        &self,
201        graph: &str,
202        body: &str,
203        format: SerializationFormat,
204        mode: GraphWriteMode,
205    ) -> Result<usize, DomainError> {
206        let rdf_format = to_rdf_format(format)?;
207        let parsed = sbol_rdf::Graph::parse(body, rdf_format)
208            .map_err(|e| DomainError::Parse(e.to_string()))?;
209        let triples = rdf_graph_to_triples(&parsed, &IriString::unchecked(graph));
210
211        let mut tx = self.pool.begin().await.map_err(db_err)?;
212        self.triples
213            .ensure_graph(&mut tx, graph, "verbatim")
214            .await?;
215        if mode == GraphWriteMode::Replace {
216            self.triples.clear_graph(&mut tx, Some(graph)).await?;
217        }
218        let inserted = self
219            .triples
220            .insert_triples(&mut tx, &triples, "graph-store")
221            .await?;
222        self.accel.refresh_graph(&mut tx, graph).await?;
223        tx.commit().await.map_err(db_err)?;
224        Ok(inserted)
225    }
226
227    pub async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError> {
228        let mut tx = self.pool.begin().await.map_err(db_err)?;
229        let deleted = self.triples.clear_graph(&mut tx, Some(graph)).await?;
230        self.triples.delete_graph(&mut tx, graph).await?;
231        self.accel.refresh_graph(&mut tx, graph).await?;
232        tx.commit().await.map_err(db_err)?;
233        Ok(deleted)
234    }
235
236    pub async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError> {
237        self.triples
238            .triples_for_graph(Some(graph), GRAPH_READ_LIMIT)
239            .await
240    }
241
242    pub async fn graph_store_read_page(
243        &self,
244        graph: &str,
245        after: Option<&str>,
246        limit: usize,
247    ) -> Result<TripleScanPage, DomainError> {
248        let after_id = after
249            .map(str::parse::<i64>)
250            .transpose()
251            .map_err(|_| DomainError::InvalidInput("invalid SQLite graph cursor".to_owned()))?
252            .unwrap_or(0);
253        let rows = self
254            .triples
255            .scan_graph_page(graph, after_id, i64::try_from(limit).unwrap_or(i64::MAX))
256            .await?;
257        let next_cursor = (rows.len() == limit)
258            .then(|| rows.last().map(|(id, _)| id.to_string()))
259            .flatten();
260        Ok(TripleScanPage {
261            items: rows.into_iter().map(|(_, triple)| triple).collect(),
262            next_cursor,
263        })
264    }
265}
266
267/// Synchronous [`TripleSource`] over the async triplestore, mirroring the
268/// Postgres adapter: each call blocks the current async task to completion, so
269/// it is only valid inside `tokio::task::spawn_blocking` (how the SPARQL
270/// evaluator drives it).
271#[derive(Clone)]
272struct SqliteTripleSource {
273    triples: Arc<TripleRepository>,
274    accel: AccelRepository,
275}
276
277impl TripleSource for SqliteTripleSource {
278    fn scan_all_page(
279        &self,
280        after: Option<&str>,
281        limit: usize,
282    ) -> Result<TripleScanPage, DomainError> {
283        let after_id = after
284            .map(str::parse::<i64>)
285            .transpose()
286            .map_err(|_| DomainError::InvalidInput("invalid SQLite triple cursor".to_owned()))?
287            .unwrap_or(0);
288        let rows = Handle::current().block_on(
289            self.triples
290                .scan_all_page(after_id, i64::try_from(limit).unwrap_or(i64::MAX)),
291        )?;
292        let next_cursor = (rows.len() == limit)
293            .then(|| rows.last().map(|(id, _)| id.to_string()))
294            .flatten();
295        Ok(TripleScanPage {
296            items: rows.into_iter().map(|(_, triple)| triple).collect(),
297            next_cursor,
298        })
299    }
300
301    fn scan_pattern(
302        &self,
303        subject: Option<&PatternSubject>,
304        predicate: Option<&str>,
305        object: Option<&PatternObject>,
306        graph: Option<&GraphFilter>,
307        limit: i64,
308    ) -> Result<Vec<Triple>, DomainError> {
309        Handle::current().block_on(
310            self.triples
311                .scan_pattern(subject, predicate, object, graph, limit),
312        )
313    }
314
315    fn distinct_named_graphs(&self) -> Result<Vec<String>, DomainError> {
316        Handle::current().block_on(self.triples.distinct_named_graphs())
317    }
318
319    fn triples_for_graph(
320        &self,
321        graph: Option<&str>,
322        limit: i64,
323    ) -> Result<Vec<Triple>, DomainError> {
324        Handle::current().block_on(self.triples.triples_for_graph(graph, limit))
325    }
326
327    fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError> {
328        Handle::current().block_on(self.triples.triples_for_subject(subject_iri))
329    }
330
331    fn run_accelerated(
332        &self,
333        query: &AcceleratedQuery,
334    ) -> Result<Option<AccelSolutions>, DomainError> {
335        Handle::current().block_on(self.accel.run(query)).map(Some)
336    }
337}
338
339/// Transactional [`TripleWriter`] for SPARQL Update: the whole batch commits or
340/// none of it does.
341#[derive(Clone)]
342struct SqliteTripleWriter {
343    triples: Arc<TripleRepository>,
344    accel: AccelRepository,
345    pool: SqlitePool,
346}
347
348/// Every named graph an update touches (insert, delete, or clear), whose
349/// accelerator indexes must be marked stale. The default (graphless) partition
350/// has no named graph and is never accelerated, so it is skipped.
351fn touched_named_graphs(changes: &[TripleChange]) -> HashSet<String> {
352    let mut graphs = HashSet::new();
353    for change in changes {
354        match change {
355            TripleChange::Change { deletes, inserts } => {
356                for triple in deletes.iter().chain(inserts) {
357                    if let Some(graph) = &triple.graph_iri {
358                        graphs.insert(graph.as_str().to_owned());
359                    }
360                }
361            }
362            TripleChange::Clear(graph) => {
363                if let Some(graph) = graph {
364                    graphs.insert(graph.as_str().to_owned());
365                }
366            }
367        }
368    }
369    graphs
370}
371
372#[async_trait]
373impl TripleWriter for SqliteTripleWriter {
374    async fn apply_update(&self, changes: Vec<TripleChange>) -> Result<UpdateOutcome, DomainError> {
375        let mut outcome = UpdateOutcome::default();
376        let mut tx = self.pool.begin().await.map_err(db_err)?;
377        for change in &changes {
378            match change {
379                TripleChange::Change { deletes, inserts } => {
380                    outcome.deleted += self.triples.delete_triples(&mut tx, deletes).await?;
381                    let mut ensured = HashSet::new();
382                    for triple in inserts {
383                        if let Some(graph) = &triple.graph_iri {
384                            if ensured.insert(graph.as_str().to_owned()) {
385                                self.triples
386                                    .ensure_graph(&mut tx, graph.as_str(), "verbatim")
387                                    .await?;
388                            }
389                        }
390                    }
391                    outcome.inserted += self
392                        .triples
393                        .insert_triples(&mut tx, inserts, UPDATE_SOURCE)
394                        .await?;
395                }
396                TripleChange::Clear(graph) => {
397                    outcome.deleted += self
398                        .triples
399                        .clear_graph(&mut tx, graph.as_ref().map(|i| i.as_str()))
400                        .await?;
401                }
402            }
403        }
404        for graph in touched_named_graphs(&changes) {
405            self.accel.refresh_graph(&mut tx, &graph).await?;
406        }
407        tx.commit().await.map_err(db_err)?;
408        Ok(outcome)
409    }
410}
411
412#[async_trait]
413impl ObjectStore for SqliteStore {
414    async fn get_object_by_iri(&self, iri: &str) -> Result<Option<SbolObjectRecord>, DomainError> {
415        self.objects.get_by_iri(iri).await
416    }
417
418    async fn get_objects_by_iris(
419        &self,
420        iris: &[&str],
421    ) -> Result<Vec<SbolObjectRecord>, DomainError> {
422        self.objects.get_by_iris(iris).await
423    }
424
425    async fn list_objects(
426        &self,
427        filter: &ListObjectsFilter,
428    ) -> Result<Vec<SbolObjectRecord>, DomainError> {
429        self.objects.list(filter).await
430    }
431
432    async fn get_object_iri_by_id(&self, id: ObjectId) -> Result<Option<String>, DomainError> {
433        self.objects.get_iri_by_id(id).await
434    }
435}
436
437#[async_trait]
438impl GraphStore for SqliteStore {
439    async fn get_graph(&self, id: GraphId) -> Result<Option<GraphRecord>, DomainError> {
440        self.graphs.get(id).await
441    }
442
443    async fn list_graphs(
444        &self,
445        filter: &ListGraphsFilter,
446    ) -> Result<Vec<GraphRecord>, DomainError> {
447        self.graphs.list(filter).await
448    }
449
450    async fn delete_graph(&self, id: GraphId) -> Result<bool, DomainError> {
451        self.graphs.delete(id).await
452    }
453
454    async fn graph_exists_by_hash(&self, hash: &[u8]) -> Result<bool, DomainError> {
455        self.graphs.exists_by_hash(hash).await
456    }
457
458    async fn graph_id_by_document_iri(
459        &self,
460        document_iri: &str,
461    ) -> Result<Option<GraphId>, DomainError> {
462        self.graphs.id_by_document_iri(document_iri).await
463    }
464}
465
466#[async_trait]
467impl TextSearchStore for SqliteStore {
468    async fn search_objects(
469        &self,
470        query: &TextSearchQuery,
471    ) -> Result<(Vec<SbolObjectRecord>, i64), DomainError> {
472        self.objects.search(query).await
473    }
474}
475
476#[async_trait]
477impl OntologyStore for SqliteStore {
478    async fn load_ontology_from_url(
479        &self,
480        prefix: &str,
481        name: &str,
482        source_url: &str,
483    ) -> Result<OntologyLoadReport, DomainError> {
484        let client = reqwest::Client::builder()
485            .user_agent("sbol-db/0.1 (+https://github.com/marpaia/sbol-db)")
486            .build()
487            .map_err(|e| DomainError::InvalidInput(format!("reqwest client: {e}")))?;
488        let body = client
489            .get(source_url)
490            .send()
491            .await
492            .map_err(|e| DomainError::InvalidInput(format!("fetch {source_url}: {e}")))?
493            .error_for_status()
494            .map_err(|e| DomainError::InvalidInput(format!("HTTP {source_url}: {e}")))?
495            .text()
496            .await
497            .map_err(|e| DomainError::InvalidInput(format!("decode {source_url}: {e}")))?;
498        self.ontology
499            .load_from_text(prefix, name, Some(source_url), &body)
500            .await
501    }
502
503    async fn load_ontology_from_text(
504        &self,
505        prefix: &str,
506        name: &str,
507        source_url: Option<&str>,
508        text: &str,
509    ) -> Result<OntologyLoadReport, DomainError> {
510        self.ontology
511            .load_from_text(prefix, name, source_url, text)
512            .await
513    }
514
515    async fn list_ontologies(&self) -> Result<Vec<OntologyRecord>, DomainError> {
516        self.ontology.list_ontologies().await
517    }
518
519    async fn canonicalize(&self, iri: &str) -> Result<Option<String>, DomainError> {
520        self.ontology.canonicalize(iri).await
521    }
522
523    async fn descendants(&self, iri: &str) -> Result<Vec<(String, i16)>, DomainError> {
524        self.ontology.descendants(iri).await
525    }
526
527    async fn list_ontology_terms(
528        &self,
529        prefix: &str,
530        limit: i64,
531        offset: i64,
532        search: Option<&str>,
533    ) -> Result<(Vec<OntologyTermRecord>, i64), DomainError> {
534        self.ontology
535            .list_terms(prefix, limit, offset, search)
536            .await
537    }
538
539    async fn get_ontology_term(
540        &self,
541        iri: &str,
542    ) -> Result<Option<OntologyTermRecord>, DomainError> {
543        self.ontology.get_term(iri).await
544    }
545}
546
547#[async_trait]
548impl NeighborhoodStore for SqliteStore {
549    async fn walk(&self, query: &NeighborhoodQuery) -> Result<NeighborhoodResult, DomainError> {
550        crate::repo::neighborhood::walk(&self.triples, &self.objects, query).await
551    }
552}
553
554#[async_trait]
555impl SequenceSearchStore for SqliteStore {
556    async fn search(
557        &self,
558        pattern: &str,
559        options: SequenceSearchOptions,
560    ) -> Result<Vec<SequenceMatch>, DomainError> {
561        self.sequences.search(pattern, options).await
562    }
563
564    async fn search_many(
565        &self,
566        patterns: &[String],
567        options: SequenceSearchOptions,
568    ) -> Result<Vec<BatchSequenceMatch>, DomainError> {
569        self.sequences.search_many(patterns, options).await
570    }
571
572    async fn align_candidates(&self, query: &str) -> Result<Vec<(String, String)>, DomainError> {
573        self.sequences.align_candidates(query).await
574    }
575
576    async fn sequences_by_iris(
577        &self,
578        iris: &[String],
579    ) -> Result<Vec<(String, String)>, DomainError> {
580        self.sequences.sequences_by_iris(iris).await
581    }
582
583    async fn all_nucleotide_sequences(&self) -> Result<Vec<(String, String)>, DomainError> {
584        self.sequences.all_nucleotide_sequences().await
585    }
586}
587
588#[async_trait]
589impl LabStore for SqliteStore {
590    async fn corpus_counts(&self) -> Result<CorpusCounts, DomainError> {
591        self.lab.corpus_counts().await
592    }
593
594    async fn recent_graphs(&self, limit: i64) -> Result<Vec<GraphOverview>, DomainError> {
595        self.lab.list_graph_overviews(None, limit, 0).await
596    }
597
598    async fn top_classes(&self, limit: i64) -> Result<Vec<ClassCount>, DomainError> {
599        self.lab.top_classes(limit).await
600    }
601
602    async fn count_graphs(&self, kind: Option<&str>) -> Result<i64, DomainError> {
603        self.lab.count_graphs(kind).await
604    }
605
606    async fn list_graph_overviews(
607        &self,
608        kind: Option<&str>,
609        limit: i64,
610        offset: i64,
611    ) -> Result<Vec<GraphOverview>, DomainError> {
612        self.lab.list_graph_overviews(kind, limit, offset).await
613    }
614
615    async fn get_graph_overview(&self, id: GraphId) -> Result<Option<GraphOverview>, DomainError> {
616        self.lab.get_graph_overview(id).await
617    }
618
619    async fn graph_triples(
620        &self,
621        id: GraphId,
622        limit: i64,
623        offset: i64,
624    ) -> Result<Option<GraphTriplesPage>, DomainError> {
625        self.lab.graph_triples(id, limit, offset).await
626    }
627}
628
629#[async_trait]
630impl ResourceCatalogStore for SqliteStore {
631    async fn catalog_resource(&self, iri: &str) -> Result<Option<ResourceRecord>, DomainError> {
632        self.accel.resource(iri).await
633    }
634
635    async fn catalog_resource_occurrences(
636        &self,
637        iri: &str,
638    ) -> Result<Vec<ResourceOccurrence>, DomainError> {
639        self.accel.resource_occurrences(iri).await
640    }
641
642    async fn catalog_resources(
643        &self,
644        query: &ResourceQuery,
645    ) -> Result<CursorPage<ResourceRecord>, DomainError> {
646        self.accel.resources(query).await
647    }
648
649    async fn catalog_resources_by_iris(
650        &self,
651        iris: &[String],
652    ) -> Result<Vec<ResourceRecord>, DomainError> {
653        self.accel.resources_for_iris(iris).await
654    }
655
656    async fn catalog_top_classes(&self, limit: u32) -> Result<Vec<ClassCount>, DomainError> {
657        self.lab.top_classes(i64::from(limit.clamp(1, 500))).await
658    }
659}
660
661#[async_trait]
662impl SequenceCatalogStore for SqliteStore {
663    async fn catalog_sequence(
664        &self,
665        iri: &str,
666    ) -> Result<Option<CatalogSequenceRecord>, DomainError> {
667        self.accel.sequence(iri).await
668    }
669
670    async fn catalog_sequences(
671        &self,
672        query: &SequenceQuery,
673    ) -> Result<CursorPage<CatalogSequenceRecord>, DomainError> {
674        self.accel.sequences(query).await
675    }
676}
677
678#[async_trait]
679impl CorpusStatsStore for SqliteStore {
680    async fn catalog_stats(&self) -> Result<CorpusStats, DomainError> {
681        Ok(self.lab.corpus_counts().await?.into())
682    }
683}
684
685#[async_trait]
686impl NamedGraphCatalogStore for SqliteStore {
687    async fn catalog_graph(&self, id: GraphId) -> Result<Option<NamedGraphRecord>, DomainError> {
688        Ok(self.lab.get_graph_overview(id).await?.map(Into::into))
689    }
690
691    async fn catalog_graphs(
692        &self,
693        query: &NamedGraphQuery,
694    ) -> Result<CursorPage<NamedGraphRecord>, DomainError> {
695        let limit = query.limit.clamp(1, 500) as i64;
696        let mut rows = self
697            .lab
698            .catalog_graphs(query.after.as_deref(), query.text.as_deref(), limit + 1)
699            .await?;
700        let has_more = rows.len() > limit as usize;
701        rows.truncate(limit as usize);
702        let next_cursor = has_more.then(|| rows.last().expect("non-empty graph page").iri.clone());
703        Ok(CursorPage {
704            items: rows.into_iter().map(Into::into).collect(),
705            next_cursor,
706        })
707    }
708
709    async fn catalog_graph_triples(
710        &self,
711        id: GraphId,
712        query: &TriplePageQuery,
713    ) -> Result<Option<CursorPage<Triple>>, DomainError> {
714        let Some(graph) = self.lab.get_graph_overview(id).await? else {
715            return Ok(None);
716        };
717        let page = SqliteStore::graph_store_read_page(
718            self,
719            &graph.iri,
720            query.after.as_deref(),
721            query.limit.clamp(1, 5_000) as usize,
722        )
723        .await?;
724        Ok(Some(CursorPage {
725            items: page.items,
726            next_cursor: page.next_cursor,
727        }))
728    }
729}
730
731#[async_trait]
732impl AclStore for SqliteStore {
733    async fn owned_graphs(&self, owner_iri: &str) -> Result<Vec<String>, DomainError> {
734        let object = PatternObject::Iri(owner_iri.to_owned());
735        let triples = self
736            .triples
737            .scan_pattern(None, Some(SBH_OWNED_BY), Some(&object), None, i64::MAX)
738            .await?;
739        Ok(distinct_graph_iris(triples))
740    }
741
742    async fn viewable_objects(&self, owner_iri: &str) -> Result<Vec<String>, DomainError> {
743        let subject = PatternSubject::Iri(owner_iri.to_owned());
744        let triples = self
745            .triples
746            .scan_pattern(Some(&subject), Some(SBH_CAN_VIEW), None, None, i64::MAX)
747            .await?;
748        Ok(distinct_object_iris(triples))
749    }
750}
751
752#[async_trait]
753impl SbolStore for SqliteStore {
754    async fn import_document(&self, input: ImportInput) -> Result<ImportReport, DomainError> {
755        self.import_document(input).await
756    }
757
758    async fn import_documents(
759        &self,
760        inputs: Vec<ImportInput>,
761    ) -> Result<Vec<ImportReport>, DomainError> {
762        self.import_documents(inputs).await
763    }
764
765    async fn graph_store_write(
766        &self,
767        graph: &str,
768        body: &str,
769        format: SerializationFormat,
770        mode: GraphWriteMode,
771    ) -> Result<usize, DomainError> {
772        self.graph_store_write(graph, body, format, mode).await
773    }
774
775    async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError> {
776        self.graph_store_clear(graph).await
777    }
778
779    async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError> {
780        self.graph_store_read(graph).await
781    }
782
783    async fn graph_store_read_page(
784        &self,
785        graph: &str,
786        after: Option<&str>,
787        limit: usize,
788    ) -> Result<TripleScanPage, DomainError> {
789        self.graph_store_read_page(graph, after, limit).await
790    }
791
792    async fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError> {
793        self.triples.triples_for_subject(subject_iri).await
794    }
795
796    async fn ping(&self) -> Result<(), DomainError> {
797        self.ping().await
798    }
799}
sbol_db_sqlite/store.rs

sbol_db_sqlite/
store.rs
