sbol_db_postgres/
store_impls.rs

1//! Postgres implementations of the `sbol-db-storage` contract.
2//!
3//! The async-trait impls delegate to the inherent methods on
4//! [`SbolObjectService`] and [`JobRepository`]; inherent-method resolution
5//! takes priority over the trait method of the same name, so these delegations
6//! do not recurse. [`PostgresTripleSource`] and [`PostgresTripleWriter`] adapt
7//! the async [`TripleRepository`] to the synchronous [`TripleSource`] and the
8//! transactional [`TripleWriter`] contracts.
9
10use std::sync::Arc;
11use std::time::Duration;
12
13use async_trait::async_trait;
14use sbol_db_core::{
15    DomainError, GraphId, GraphRecord, ImportReport, JobId, NeighborhoodQuery, NeighborhoodResult,
16    ObjectId, SbolObjectRecord, SerializationFormat, Triple,
17};
18use sbol_db_storage::{
19    distinct_graph_iris, distinct_object_iris, AccelSolutions, AcceleratedQuery, AclStore,
20    BatchSequenceMatch, CatalogSequenceRecord, ClassCount, CorpusCounts, CorpusStats,
21    CorpusStatsStore, CursorPage, EnqueueOutcome, GraphFilter, GraphOverview, GraphStore,
22    GraphTriplesPage, GraphWriteMode, ImportInput, JobAttempt, JobLogRecord, JobQueue, JobStatus,
23    LabStore, ListGraphsFilter, ListJobsFilter, ListObjectsFilter, NamedGraphCatalogStore,
24    NamedGraphQuery, NamedGraphRecord, NeighborhoodStore, NewJob, ObjectStore, OldestQueuedAge,
25    OntologyLoadReport, OntologyRecord, OntologyStore, OntologyTermRecord, PatternObject,
26    PatternSubject, QueueDepthRow, ResourceCatalogStore, ResourceOccurrence, ResourceQuery,
27    ResourceRecord, SbolJob, SbolStore, SequenceCatalogStore, SequenceMatch, SequenceQuery,
28    SequenceSearchOptions, SequenceSearchStore, TextSearchQuery, TextSearchStore, TripleChange,
29    TriplePageQuery, TripleScanPage, TripleSource, TripleWriter, UpdateOutcome, SBH_CAN_VIEW,
30    SBH_OWNED_BY,
31};
32use serde_json::Value;
33use tokio::runtime::Handle;
34
35use crate::repo::db_err;
36use crate::{AccelRepository, JobRepository, PgPool, SbolObjectService, TripleRepository};
37
38/// Verbatim source tag for triples written by SPARQL Update.
39const UPDATE_SOURCE: &str = "sparql-update";
40
41/// Synchronous [`TripleSource`] over the async [`TripleRepository`].
42///
43/// Each method runs the async scan to completion via
44/// `Handle::current().block_on(...)`. This is only valid when called from
45/// inside `tokio::task::spawn_blocking`, which is exactly how the SPARQL engine
46/// drives the evaluator's synchronous `QueryableDataset`.
47#[derive(Clone)]
48pub struct PostgresTripleSource {
49    triples: Arc<TripleRepository>,
50    accel: AccelRepository,
51}
52
53impl PostgresTripleSource {
54    pub fn new(triples: Arc<TripleRepository>, accel: AccelRepository) -> Self {
55        Self { triples, accel }
56    }
57}
58
59impl TripleSource for PostgresTripleSource {
60    fn scan_all_page(
61        &self,
62        after: Option<&str>,
63        limit: usize,
64    ) -> Result<TripleScanPage, DomainError> {
65        let after_id = after
66            .map(str::parse::<i64>)
67            .transpose()
68            .map_err(|_| DomainError::InvalidInput("invalid Postgres triple cursor".to_owned()))?
69            .unwrap_or(0);
70        let rows = Handle::current().block_on(
71            self.triples
72                .scan_all_page(after_id, i64::try_from(limit).unwrap_or(i64::MAX)),
73        )?;
74        let next_cursor = (rows.len() == limit)
75            .then(|| rows.last().map(|(id, _)| id.to_string()))
76            .flatten();
77        Ok(TripleScanPage {
78            items: rows.into_iter().map(|(_, triple)| triple).collect(),
79            next_cursor,
80        })
81    }
82
83    fn scan_pattern(
84        &self,
85        subject: Option<&PatternSubject>,
86        predicate: Option<&str>,
87        object: Option<&PatternObject>,
88        graph: Option<&GraphFilter>,
89        limit: i64,
90    ) -> Result<Vec<Triple>, DomainError> {
91        Handle::current().block_on(
92            self.triples
93                .scan_pattern(subject, predicate, object, graph, limit),
94        )
95    }
96
97    fn distinct_named_graphs(&self) -> Result<Vec<String>, DomainError> {
98        Handle::current().block_on(self.triples.distinct_named_graphs())
99    }
100
101    fn triples_for_graph(
102        &self,
103        graph: Option<&str>,
104        limit: i64,
105    ) -> Result<Vec<Triple>, DomainError> {
106        Handle::current().block_on(self.triples.triples_for_graph(graph, limit))
107    }
108
109    fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError> {
110        Handle::current().block_on(self.triples.triples_for_subject(subject_iri))
111    }
112
113    fn run_accelerated(
114        &self,
115        query: &AcceleratedQuery,
116    ) -> Result<Option<AccelSolutions>, DomainError> {
117        Handle::current().block_on(self.accel.run(query)).map(Some)
118    }
119}
120
121/// Transactional [`TripleWriter`] for SPARQL Update.
122///
123/// `apply_update` runs the whole batch in one Postgres transaction: each
124/// `Change` deletes, registers any named graph its inserts target (the graph
125/// row owns its triples via FK), then inserts; each `Clear` drops a graph's
126/// contents. Either the whole update commits or none of it does.
127#[derive(Clone)]
128pub struct PostgresTripleWriter {
129    triples: Arc<TripleRepository>,
130    accel: AccelRepository,
131    pool: PgPool,
132}
133
134impl PostgresTripleWriter {
135    pub fn new(triples: Arc<TripleRepository>, accel: AccelRepository, pool: PgPool) -> Self {
136        Self {
137            triples,
138            accel,
139            pool,
140        }
141    }
142}
143
144/// Every named graph an update touches (insert, delete, or clear), whose
145/// accelerator indexes must be marked stale. The default (graphless) partition
146/// has no named graph and is never accelerated, so it is skipped.
147fn touched_named_graphs(changes: &[TripleChange]) -> std::collections::HashSet<String> {
148    let mut graphs = std::collections::HashSet::new();
149    for change in changes {
150        match change {
151            TripleChange::Change { deletes, inserts } => {
152                for triple in deletes.iter().chain(inserts) {
153                    if let Some(graph) = &triple.graph_iri {
154                        graphs.insert(graph.as_str().to_owned());
155                    }
156                }
157            }
158            TripleChange::Clear(graph) => {
159                if let Some(graph) = graph {
160                    graphs.insert(graph.as_str().to_owned());
161                }
162            }
163        }
164    }
165    graphs
166}
167
168#[async_trait]
169impl TripleWriter for PostgresTripleWriter {
170    async fn apply_update(&self, changes: Vec<TripleChange>) -> Result<UpdateOutcome, DomainError> {
171        let mut outcome = UpdateOutcome::default();
172        let mut tx = self.pool.begin().await.map_err(db_err)?;
173        for change in &changes {
174            match change {
175                TripleChange::Change { deletes, inserts } => {
176                    outcome.deleted += self.triples.delete_triples(&mut tx, deletes).await?;
177                    let mut ensured = std::collections::HashSet::new();
178                    for triple in inserts {
179                        if let Some(graph) = &triple.graph_iri {
180                            if ensured.insert(graph.as_str().to_owned()) {
181                                self.triples
182                                    .ensure_graph(&mut tx, graph.as_str(), "verbatim")
183                                    .await?;
184                            }
185                        }
186                    }
187                    outcome.inserted += self
188                        .triples
189                        .insert_triples(&mut tx, inserts, UPDATE_SOURCE)
190                        .await?;
191                }
192                TripleChange::Clear(graph) => {
193                    outcome.deleted += self
194                        .triples
195                        .clear_graph(&mut tx, graph.as_ref().map(|i| i.as_str()))
196                        .await?;
197                }
198            }
199        }
200        for graph in touched_named_graphs(&changes) {
201            self.accel.refresh_graph(&mut tx, &graph).await?;
202        }
203        tx.commit().await.map_err(db_err)?;
204        Ok(outcome)
205    }
206}
207
208impl SbolObjectService {
209    /// A [`TripleSource`] view of this service's triple store, for the SPARQL
210    /// read engine.
211    pub fn triple_source(&self) -> Arc<dyn TripleSource> {
212        Arc::new(PostgresTripleSource::new(
213            Arc::new(self.triples().clone()),
214            self.accel().clone(),
215        ))
216    }
217
218    /// A [`TripleWriter`] view of this service's triple store, for SPARQL Update.
219    pub fn triple_writer(&self) -> Arc<dyn TripleWriter> {
220        Arc::new(PostgresTripleWriter::new(
221            Arc::new(self.triples().clone()),
222            self.accel().clone(),
223            self.pool().clone(),
224        ))
225    }
226}
227
228#[async_trait]
229impl ObjectStore for SbolObjectService {
230    async fn get_object_by_iri(&self, iri: &str) -> Result<Option<SbolObjectRecord>, DomainError> {
231        self.objects().get_by_iri(iri).await
232    }
233
234    async fn get_objects_by_iris(
235        &self,
236        iris: &[&str],
237    ) -> Result<Vec<SbolObjectRecord>, DomainError> {
238        self.objects().get_by_iris(iris).await
239    }
240
241    async fn list_objects(
242        &self,
243        filter: &ListObjectsFilter,
244    ) -> Result<Vec<SbolObjectRecord>, DomainError> {
245        self.objects().list(filter).await
246    }
247
248    async fn get_object_iri_by_id(&self, id: ObjectId) -> Result<Option<String>, DomainError> {
249        self.objects().get_iri_by_id(id).await
250    }
251}
252
253#[async_trait]
254impl GraphStore for SbolObjectService {
255    async fn get_graph(&self, id: GraphId) -> Result<Option<GraphRecord>, DomainError> {
256        self.graphs().get(id).await
257    }
258
259    async fn list_graphs(
260        &self,
261        filter: &ListGraphsFilter,
262    ) -> Result<Vec<GraphRecord>, DomainError> {
263        self.graphs().list(filter).await
264    }
265
266    async fn delete_graph(&self, id: GraphId) -> Result<bool, DomainError> {
267        self.graphs().delete(id).await
268    }
269
270    async fn graph_exists_by_hash(&self, hash: &[u8]) -> Result<bool, DomainError> {
271        self.graphs().exists_by_hash(hash).await
272    }
273
274    async fn graph_id_by_document_iri(
275        &self,
276        document_iri: &str,
277    ) -> Result<Option<GraphId>, DomainError> {
278        self.graphs().id_by_document_iri(document_iri).await
279    }
280}
281
282#[async_trait]
283impl TextSearchStore for SbolObjectService {
284    async fn search_objects(
285        &self,
286        query: &TextSearchQuery,
287    ) -> Result<(Vec<SbolObjectRecord>, i64), DomainError> {
288        self.objects().search(query).await
289    }
290}
291
292#[async_trait]
293impl OntologyStore for SbolObjectService {
294    async fn load_ontology_from_url(
295        &self,
296        prefix: &str,
297        name: &str,
298        source_url: &str,
299    ) -> Result<OntologyLoadReport, DomainError> {
300        self.ontology()
301            .load_from_url(prefix, name, source_url)
302            .await
303    }
304
305    async fn load_ontology_from_text(
306        &self,
307        prefix: &str,
308        name: &str,
309        source_url: Option<&str>,
310        text: &str,
311    ) -> Result<OntologyLoadReport, DomainError> {
312        self.ontology()
313            .load_from_text(prefix, name, source_url, text)
314            .await
315    }
316
317    async fn list_ontologies(&self) -> Result<Vec<OntologyRecord>, DomainError> {
318        self.ontology().list_ontologies().await
319    }
320
321    async fn canonicalize(&self, iri: &str) -> Result<Option<String>, DomainError> {
322        self.ontology().canonicalize(iri).await
323    }
324
325    async fn descendants(&self, iri: &str) -> Result<Vec<(String, i16)>, DomainError> {
326        self.ontology().descendants(iri).await
327    }
328
329    async fn list_ontology_terms(
330        &self,
331        prefix: &str,
332        limit: i64,
333        offset: i64,
334        search: Option<&str>,
335    ) -> Result<(Vec<OntologyTermRecord>, i64), DomainError> {
336        self.ontology()
337            .list_terms(prefix, limit, offset, search)
338            .await
339    }
340
341    async fn get_ontology_term(
342        &self,
343        iri: &str,
344    ) -> Result<Option<OntologyTermRecord>, DomainError> {
345        self.ontology().get_term(iri).await
346    }
347}
348
349#[async_trait]
350impl LabStore for SbolObjectService {
351    async fn corpus_counts(&self) -> Result<CorpusCounts, DomainError> {
352        self.lab().corpus_counts().await
353    }
354
355    async fn recent_graphs(&self, limit: i64) -> Result<Vec<GraphOverview>, DomainError> {
356        self.lab().list_graph_overviews(None, limit, 0).await
357    }
358
359    async fn top_classes(&self, limit: i64) -> Result<Vec<ClassCount>, DomainError> {
360        self.lab().top_classes(limit).await
361    }
362
363    async fn count_graphs(&self, kind: Option<&str>) -> Result<i64, DomainError> {
364        self.lab().count_graphs(kind).await
365    }
366
367    async fn list_graph_overviews(
368        &self,
369        kind: Option<&str>,
370        limit: i64,
371        offset: i64,
372    ) -> Result<Vec<GraphOverview>, DomainError> {
373        self.lab().list_graph_overviews(kind, limit, offset).await
374    }
375
376    async fn get_graph_overview(&self, id: GraphId) -> Result<Option<GraphOverview>, DomainError> {
377        self.lab().get_graph_overview(id).await
378    }
379
380    async fn graph_triples(
381        &self,
382        id: GraphId,
383        limit: i64,
384        offset: i64,
385    ) -> Result<Option<GraphTriplesPage>, DomainError> {
386        self.lab().graph_triples(id, limit, offset).await
387    }
388}
389
390#[async_trait]
391impl ResourceCatalogStore for SbolObjectService {
392    async fn catalog_resource(&self, iri: &str) -> Result<Option<ResourceRecord>, DomainError> {
393        self.accel().resource(iri).await
394    }
395
396    async fn catalog_resource_occurrences(
397        &self,
398        iri: &str,
399    ) -> Result<Vec<ResourceOccurrence>, DomainError> {
400        self.accel().resource_occurrences(iri).await
401    }
402
403    async fn catalog_resources(
404        &self,
405        query: &ResourceQuery,
406    ) -> Result<CursorPage<ResourceRecord>, DomainError> {
407        self.accel().resources(query).await
408    }
409
410    async fn catalog_resources_by_iris(
411        &self,
412        iris: &[String],
413    ) -> Result<Vec<ResourceRecord>, DomainError> {
414        self.accel().resources_for_iris(iris).await
415    }
416
417    async fn catalog_top_classes(&self, limit: u32) -> Result<Vec<ClassCount>, DomainError> {
418        self.lab().top_classes(i64::from(limit.clamp(1, 500))).await
419    }
420}
421
422#[async_trait]
423impl SequenceCatalogStore for SbolObjectService {
424    async fn catalog_sequence(
425        &self,
426        iri: &str,
427    ) -> Result<Option<CatalogSequenceRecord>, DomainError> {
428        self.accel().sequence(iri).await
429    }
430
431    async fn catalog_sequences(
432        &self,
433        query: &SequenceQuery,
434    ) -> Result<CursorPage<CatalogSequenceRecord>, DomainError> {
435        self.accel().sequences(query).await
436    }
437}
438
439#[async_trait]
440impl CorpusStatsStore for SbolObjectService {
441    async fn catalog_stats(&self) -> Result<CorpusStats, DomainError> {
442        Ok(self.lab().corpus_counts().await?.into())
443    }
444}
445
446#[async_trait]
447impl NamedGraphCatalogStore for SbolObjectService {
448    async fn catalog_graph(&self, id: GraphId) -> Result<Option<NamedGraphRecord>, DomainError> {
449        Ok(self.lab().get_graph_overview(id).await?.map(Into::into))
450    }
451
452    async fn catalog_graphs(
453        &self,
454        query: &NamedGraphQuery,
455    ) -> Result<CursorPage<NamedGraphRecord>, DomainError> {
456        let limit = query.limit.clamp(1, 500) as i64;
457        let mut rows = self
458            .lab()
459            .catalog_graphs(query.after.as_deref(), query.text.as_deref(), limit + 1)
460            .await?;
461        let has_more = rows.len() > limit as usize;
462        rows.truncate(limit as usize);
463        let next_cursor = has_more.then(|| rows.last().expect("non-empty graph page").iri.clone());
464        Ok(CursorPage {
465            items: rows.into_iter().map(Into::into).collect(),
466            next_cursor,
467        })
468    }
469
470    async fn catalog_graph_triples(
471        &self,
472        id: GraphId,
473        query: &TriplePageQuery,
474    ) -> Result<Option<CursorPage<Triple>>, DomainError> {
475        let Some(graph) = self.lab().get_graph_overview(id).await? else {
476            return Ok(None);
477        };
478        let page = self
479            .graph_store_read_page(
480                &graph.iri,
481                query.after.as_deref(),
482                query.limit.clamp(1, 5_000) as usize,
483            )
484            .await?;
485        Ok(Some(CursorPage {
486            items: page.items,
487            next_cursor: page.next_cursor,
488        }))
489    }
490}
491
492#[async_trait]
493impl NeighborhoodStore for SbolObjectService {
494    async fn walk(&self, query: &NeighborhoodQuery) -> Result<NeighborhoodResult, DomainError> {
495        self.neighborhood().walk(query).await
496    }
497}
498
499#[async_trait]
500impl SequenceSearchStore for SbolObjectService {
501    async fn search(
502        &self,
503        pattern: &str,
504        options: SequenceSearchOptions,
505    ) -> Result<Vec<SequenceMatch>, DomainError> {
506        self.sequence_search().search(pattern, options).await
507    }
508
509    async fn search_many(
510        &self,
511        patterns: &[String],
512        options: SequenceSearchOptions,
513    ) -> Result<Vec<BatchSequenceMatch>, DomainError> {
514        self.sequence_search().search_many(patterns, options).await
515    }
516
517    async fn align_candidates(&self, query: &str) -> Result<Vec<(String, String)>, DomainError> {
518        self.sequence_search().align_candidates(query).await
519    }
520
521    async fn sequences_by_iris(
522        &self,
523        iris: &[String],
524    ) -> Result<Vec<(String, String)>, DomainError> {
525        self.sequence_search().sequences_by_iris(iris).await
526    }
527
528    async fn all_nucleotide_sequences(&self) -> Result<Vec<(String, String)>, DomainError> {
529        self.sequence_search().all_nucleotide_sequences().await
530    }
531}
532
533#[async_trait]
534impl AclStore for SbolObjectService {
535    async fn owned_graphs(&self, owner_iri: &str) -> Result<Vec<String>, DomainError> {
536        let object = PatternObject::Iri(owner_iri.to_owned());
537        let triples = self
538            .triples()
539            .scan_pattern(None, Some(SBH_OWNED_BY), Some(&object), None, i64::MAX)
540            .await?;
541        Ok(distinct_graph_iris(triples))
542    }
543
544    async fn viewable_objects(&self, owner_iri: &str) -> Result<Vec<String>, DomainError> {
545        let subject = PatternSubject::Iri(owner_iri.to_owned());
546        let triples = self
547            .triples()
548            .scan_pattern(Some(&subject), Some(SBH_CAN_VIEW), None, None, i64::MAX)
549            .await?;
550        Ok(distinct_object_iris(triples))
551    }
552}
553
554#[async_trait]
555impl SbolStore for SbolObjectService {
556    async fn import_document(&self, input: ImportInput) -> Result<ImportReport, DomainError> {
557        self.import_document(input).await
558    }
559
560    async fn import_documents(
561        &self,
562        inputs: Vec<ImportInput>,
563    ) -> Result<Vec<ImportReport>, DomainError> {
564        self.import_documents(inputs).await
565    }
566
567    async fn graph_store_write(
568        &self,
569        graph: &str,
570        body: &str,
571        format: SerializationFormat,
572        mode: GraphWriteMode,
573    ) -> Result<usize, DomainError> {
574        self.graph_store_write(graph, body, format, mode).await
575    }
576
577    async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError> {
578        self.graph_store_clear(graph).await
579    }
580
581    async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError> {
582        self.graph_store_read(graph).await
583    }
584
585    async fn graph_store_read_page(
586        &self,
587        graph: &str,
588        after: Option<&str>,
589        limit: usize,
590    ) -> Result<TripleScanPage, DomainError> {
591        self.graph_store_read_page(graph, after, limit).await
592    }
593
594    async fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError> {
595        self.triples().triples_for_subject(subject_iri).await
596    }
597
598    async fn ping(&self) -> Result<(), DomainError> {
599        self.ping().await
600    }
601}
602
603#[async_trait]
604impl JobQueue for JobRepository {
605    async fn enqueue(&self, input: NewJob) -> Result<EnqueueOutcome, DomainError> {
606        self.enqueue(input).await
607    }
608
609    async fn append_log(
610        &self,
611        job_id: JobId,
612        attempt_no: Option<i32>,
613        level: &str,
614        message: &str,
615        fields: Value,
616    ) -> Result<JobLogRecord, DomainError> {
617        self.append_log(job_id, attempt_no, level, message, fields)
618            .await
619    }
620
621    async fn list_logs(
622        &self,
623        id: JobId,
624        after_id: Option<i64>,
625        limit: u32,
626    ) -> Result<Vec<JobLogRecord>, DomainError> {
627        self.list_logs(id, after_id, limit).await
628    }
629
630    async fn dequeue(
631        &self,
632        queues: &[String],
633        worker_id: &str,
634        lease: Duration,
635    ) -> Result<Option<SbolJob>, DomainError> {
636        self.dequeue(queues, worker_id, lease).await
637    }
638
639    async fn renew_lease(
640        &self,
641        job_id: JobId,
642        worker_id: &str,
643        lease: Duration,
644    ) -> Result<bool, DomainError> {
645        self.renew_lease(job_id, worker_id, lease).await
646    }
647
648    async fn mark_succeeded(
649        &self,
650        job_id: JobId,
651        worker_id: &str,
652        result: Option<Value>,
653    ) -> Result<(), DomainError> {
654        self.mark_succeeded(job_id, worker_id, result).await
655    }
656
657    async fn mark_failed(
658        &self,
659        job_id: JobId,
660        worker_id: &str,
661        error: &str,
662    ) -> Result<JobStatus, DomainError> {
663        self.mark_failed(job_id, worker_id, error).await
664    }
665
666    async fn reap_expired_leases(&self) -> Result<u64, DomainError> {
667        self.reap_expired_leases().await
668    }
669
670    async fn get(&self, id: JobId) -> Result<Option<SbolJob>, DomainError> {
671        self.get(id).await
672    }
673
674    async fn list_attempts(&self, id: JobId) -> Result<Vec<JobAttempt>, DomainError> {
675        self.list_attempts(id).await
676    }
677
678    async fn list(&self, filter: &ListJobsFilter) -> Result<Vec<SbolJob>, DomainError> {
679        self.list(filter).await
680    }
681
682    async fn cancel(&self, id: JobId) -> Result<bool, DomainError> {
683        self.cancel(id).await
684    }
685
686    async fn current_status(&self, id: JobId) -> Result<Option<JobStatus>, DomainError> {
687        self.current_status(id).await
688    }
689
690    async fn queue_depth_snapshot(&self) -> Result<Vec<QueueDepthRow>, DomainError> {
691        self.queue_depth_snapshot().await
692    }
693
694    async fn oldest_queued_age(&self) -> Result<Vec<OldestQueuedAge>, DomainError> {
695        self.oldest_queued_age().await
696    }
697
698    async fn recent_failure_count(&self, within_hours: i64) -> Result<i64, DomainError> {
699        self.recent_failure_count(within_hours).await
700    }
701}
sbol_db_postgres/store_impls.rs

sbol_db_postgres/
store_impls.rs
