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    AccelSolutions, AcceleratedQuery, BatchSequenceMatch, ClassCount, CorpusCounts, EnqueueOutcome,
20    GraphFilter, GraphOverview, GraphStore, GraphTriplesPage, GraphWriteMode, ImportInput,
21    JobAttempt, JobLogRecord, JobQueue, JobStatus, LabStore, ListGraphsFilter, ListJobsFilter,
22    ListObjectsFilter, NeighborhoodStore, NewJob, ObjectStore, OldestQueuedAge, OntologyLoadReport,
23    OntologyRecord, OntologyStore, OntologyTermRecord, PatternObject, PatternSubject,
24    QueueDepthRow, SbolJob, SbolStore, SequenceMatch, SequenceSearchOptions, SequenceSearchStore,
25    TripleChange, TripleSource, TripleWriter, UpdateOutcome,
26};
27use serde_json::Value;
28use tokio::runtime::Handle;
29
30use crate::repo::db_err;
31use crate::{AccelRepository, JobRepository, PgPool, SbolObjectService, TripleRepository};
32
33/// Verbatim source tag for triples written by SPARQL Update.
34const UPDATE_SOURCE: &str = "sparql-update";
35
36/// Synchronous [`TripleSource`] over the async [`TripleRepository`].
37///
38/// Each method runs the async scan to completion via
39/// `Handle::current().block_on(...)`. This is only valid when called from
40/// inside `tokio::task::spawn_blocking`, which is exactly how the SPARQL engine
41/// drives the evaluator's synchronous `QueryableDataset`.
42#[derive(Clone)]
43pub struct PostgresTripleSource {
44    triples: Arc<TripleRepository>,
45    accel: AccelRepository,
46}
47
48impl PostgresTripleSource {
49    pub fn new(triples: Arc<TripleRepository>, accel: AccelRepository) -> Self {
50        Self { triples, accel }
51    }
52}
53
54impl TripleSource for PostgresTripleSource {
55    fn scan_pattern(
56        &self,
57        subject: Option<&PatternSubject>,
58        predicate: Option<&str>,
59        object: Option<&PatternObject>,
60        graph: Option<&GraphFilter>,
61        limit: i64,
62    ) -> Result<Vec<Triple>, DomainError> {
63        Handle::current().block_on(
64            self.triples
65                .scan_pattern(subject, predicate, object, graph, limit),
66        )
67    }
68
69    fn distinct_named_graphs(&self) -> Result<Vec<String>, DomainError> {
70        Handle::current().block_on(self.triples.distinct_named_graphs())
71    }
72
73    fn triples_for_graph(
74        &self,
75        graph: Option<&str>,
76        limit: i64,
77    ) -> Result<Vec<Triple>, DomainError> {
78        Handle::current().block_on(self.triples.triples_for_graph(graph, limit))
79    }
80
81    fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError> {
82        Handle::current().block_on(self.triples.triples_for_subject(subject_iri))
83    }
84
85    fn run_accelerated(
86        &self,
87        query: &AcceleratedQuery,
88    ) -> Result<Option<AccelSolutions>, DomainError> {
89        Handle::current().block_on(self.accel.run(query)).map(Some)
90    }
91}
92
93/// Transactional [`TripleWriter`] for SPARQL Update.
94///
95/// `apply_update` runs the whole batch in one Postgres transaction: each
96/// `Change` deletes, registers any named graph its inserts target (the graph
97/// row owns its triples via FK), then inserts; each `Clear` drops a graph's
98/// contents. Either the whole update commits or none of it does.
99#[derive(Clone)]
100pub struct PostgresTripleWriter {
101    triples: Arc<TripleRepository>,
102    accel: AccelRepository,
103    pool: PgPool,
104}
105
106impl PostgresTripleWriter {
107    pub fn new(triples: Arc<TripleRepository>, accel: AccelRepository, pool: PgPool) -> Self {
108        Self {
109            triples,
110            accel,
111            pool,
112        }
113    }
114}
115
116/// Every named graph an update touches (insert, delete, or clear), whose
117/// accelerator indexes must be marked stale. The default (graphless) partition
118/// has no named graph and is never accelerated, so it is skipped.
119fn touched_named_graphs(changes: &[TripleChange]) -> std::collections::HashSet<String> {
120    let mut graphs = std::collections::HashSet::new();
121    for change in changes {
122        match change {
123            TripleChange::Change { deletes, inserts } => {
124                for triple in deletes.iter().chain(inserts) {
125                    if let Some(graph) = &triple.graph_iri {
126                        graphs.insert(graph.as_str().to_owned());
127                    }
128                }
129            }
130            TripleChange::Clear(graph) => {
131                if let Some(graph) = graph {
132                    graphs.insert(graph.as_str().to_owned());
133                }
134            }
135        }
136    }
137    graphs
138}
139
140#[async_trait]
141impl TripleWriter for PostgresTripleWriter {
142    async fn apply_update(&self, changes: Vec<TripleChange>) -> Result<UpdateOutcome, DomainError> {
143        let mut outcome = UpdateOutcome::default();
144        let mut tx = self.pool.begin().await.map_err(db_err)?;
145        for change in &changes {
146            match change {
147                TripleChange::Change { deletes, inserts } => {
148                    outcome.deleted += self.triples.delete_triples(&mut tx, deletes).await?;
149                    let mut ensured = std::collections::HashSet::new();
150                    for triple in inserts {
151                        if let Some(graph) = &triple.graph_iri {
152                            if ensured.insert(graph.as_str().to_owned()) {
153                                self.triples
154                                    .ensure_graph(&mut tx, graph.as_str(), "verbatim")
155                                    .await?;
156                            }
157                        }
158                    }
159                    outcome.inserted += self
160                        .triples
161                        .insert_triples(&mut tx, inserts, UPDATE_SOURCE)
162                        .await?;
163                }
164                TripleChange::Clear(graph) => {
165                    outcome.deleted += self
166                        .triples
167                        .clear_graph(&mut tx, graph.as_ref().map(|i| i.as_str()))
168                        .await?;
169                }
170            }
171        }
172        for graph in touched_named_graphs(&changes) {
173            self.accel.refresh_graph(&mut tx, &graph).await?;
174        }
175        tx.commit().await.map_err(db_err)?;
176        Ok(outcome)
177    }
178}
179
180impl SbolObjectService {
181    /// A [`TripleSource`] view of this service's triple store, for the SPARQL
182    /// read engine.
183    pub fn triple_source(&self) -> Arc<dyn TripleSource> {
184        Arc::new(PostgresTripleSource::new(
185            Arc::new(self.triples().clone()),
186            self.accel().clone(),
187        ))
188    }
189
190    /// A [`TripleWriter`] view of this service's triple store, for SPARQL Update.
191    pub fn triple_writer(&self) -> Arc<dyn TripleWriter> {
192        Arc::new(PostgresTripleWriter::new(
193            Arc::new(self.triples().clone()),
194            self.accel().clone(),
195            self.pool().clone(),
196        ))
197    }
198}
199
200#[async_trait]
201impl ObjectStore for SbolObjectService {
202    async fn get_object_by_iri(&self, iri: &str) -> Result<Option<SbolObjectRecord>, DomainError> {
203        self.objects().get_by_iri(iri).await
204    }
205
206    async fn get_objects_by_iris(
207        &self,
208        iris: &[&str],
209    ) -> Result<Vec<SbolObjectRecord>, DomainError> {
210        self.objects().get_by_iris(iris).await
211    }
212
213    async fn list_objects(
214        &self,
215        filter: &ListObjectsFilter,
216    ) -> Result<Vec<SbolObjectRecord>, DomainError> {
217        self.objects().list(filter).await
218    }
219
220    async fn get_object_iri_by_id(&self, id: ObjectId) -> Result<Option<String>, DomainError> {
221        self.objects().get_iri_by_id(id).await
222    }
223}
224
225#[async_trait]
226impl GraphStore for SbolObjectService {
227    async fn get_graph(&self, id: GraphId) -> Result<Option<GraphRecord>, DomainError> {
228        self.graphs().get(id).await
229    }
230
231    async fn list_graphs(
232        &self,
233        filter: &ListGraphsFilter,
234    ) -> Result<Vec<GraphRecord>, DomainError> {
235        self.graphs().list(filter).await
236    }
237
238    async fn delete_graph(&self, id: GraphId) -> Result<bool, DomainError> {
239        self.graphs().delete(id).await
240    }
241
242    async fn graph_exists_by_hash(&self, hash: &[u8]) -> Result<bool, DomainError> {
243        self.graphs().exists_by_hash(hash).await
244    }
245}
246
247#[async_trait]
248impl OntologyStore for SbolObjectService {
249    async fn load_ontology_from_url(
250        &self,
251        prefix: &str,
252        name: &str,
253        source_url: &str,
254    ) -> Result<OntologyLoadReport, DomainError> {
255        self.ontology()
256            .load_from_url(prefix, name, source_url)
257            .await
258    }
259
260    async fn load_ontology_from_text(
261        &self,
262        prefix: &str,
263        name: &str,
264        source_url: Option<&str>,
265        text: &str,
266    ) -> Result<OntologyLoadReport, DomainError> {
267        self.ontology()
268            .load_from_text(prefix, name, source_url, text)
269            .await
270    }
271
272    async fn list_ontologies(&self) -> Result<Vec<OntologyRecord>, DomainError> {
273        self.ontology().list_ontologies().await
274    }
275
276    async fn canonicalize(&self, iri: &str) -> Result<Option<String>, DomainError> {
277        self.ontology().canonicalize(iri).await
278    }
279
280    async fn descendants(&self, iri: &str) -> Result<Vec<(String, i16)>, DomainError> {
281        self.ontology().descendants(iri).await
282    }
283
284    async fn list_ontology_terms(
285        &self,
286        prefix: &str,
287        limit: i64,
288        offset: i64,
289        search: Option<&str>,
290    ) -> Result<(Vec<OntologyTermRecord>, i64), DomainError> {
291        self.ontology()
292            .list_terms(prefix, limit, offset, search)
293            .await
294    }
295
296    async fn get_ontology_term(
297        &self,
298        iri: &str,
299    ) -> Result<Option<OntologyTermRecord>, DomainError> {
300        self.ontology().get_term(iri).await
301    }
302}
303
304#[async_trait]
305impl LabStore for SbolObjectService {
306    async fn corpus_counts(&self) -> Result<CorpusCounts, DomainError> {
307        self.lab().corpus_counts().await
308    }
309
310    async fn recent_graphs(&self, limit: i64) -> Result<Vec<GraphOverview>, DomainError> {
311        self.lab().list_graph_overviews(None, limit, 0).await
312    }
313
314    async fn top_classes(&self, limit: i64) -> Result<Vec<ClassCount>, DomainError> {
315        self.lab().top_classes(limit).await
316    }
317
318    async fn count_graphs(&self, kind: Option<&str>) -> Result<i64, DomainError> {
319        self.lab().count_graphs(kind).await
320    }
321
322    async fn list_graph_overviews(
323        &self,
324        kind: Option<&str>,
325        limit: i64,
326        offset: i64,
327    ) -> Result<Vec<GraphOverview>, DomainError> {
328        self.lab().list_graph_overviews(kind, limit, offset).await
329    }
330
331    async fn get_graph_overview(&self, id: GraphId) -> Result<Option<GraphOverview>, DomainError> {
332        self.lab().get_graph_overview(id).await
333    }
334
335    async fn graph_triples(
336        &self,
337        id: GraphId,
338        limit: i64,
339        offset: i64,
340    ) -> Result<Option<GraphTriplesPage>, DomainError> {
341        self.lab().graph_triples(id, limit, offset).await
342    }
343}
344
345#[async_trait]
346impl NeighborhoodStore for SbolObjectService {
347    async fn walk(&self, query: &NeighborhoodQuery) -> Result<NeighborhoodResult, DomainError> {
348        self.neighborhood().walk(query).await
349    }
350}
351
352#[async_trait]
353impl SequenceSearchStore for SbolObjectService {
354    async fn search(
355        &self,
356        pattern: &str,
357        options: SequenceSearchOptions,
358    ) -> Result<Vec<SequenceMatch>, DomainError> {
359        self.sequence_search().search(pattern, options).await
360    }
361
362    async fn search_many(
363        &self,
364        patterns: &[String],
365        options: SequenceSearchOptions,
366    ) -> Result<Vec<BatchSequenceMatch>, DomainError> {
367        self.sequence_search().search_many(patterns, options).await
368    }
369}
370
371#[async_trait]
372impl SbolStore for SbolObjectService {
373    async fn import_document(&self, input: ImportInput) -> Result<ImportReport, DomainError> {
374        self.import_document(input).await
375    }
376
377    async fn import_documents(
378        &self,
379        inputs: Vec<ImportInput>,
380    ) -> Result<Vec<ImportReport>, DomainError> {
381        self.import_documents(inputs).await
382    }
383
384    async fn graph_store_write(
385        &self,
386        graph: &str,
387        body: &str,
388        format: SerializationFormat,
389        mode: GraphWriteMode,
390    ) -> Result<usize, DomainError> {
391        self.graph_store_write(graph, body, format, mode).await
392    }
393
394    async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError> {
395        self.graph_store_clear(graph).await
396    }
397
398    async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError> {
399        self.graph_store_read(graph).await
400    }
401
402    async fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError> {
403        self.triples().triples_for_subject(subject_iri).await
404    }
405
406    async fn ping(&self) -> Result<(), DomainError> {
407        self.ping().await
408    }
409}
410
411#[async_trait]
412impl JobQueue for JobRepository {
413    async fn enqueue(&self, input: NewJob) -> Result<EnqueueOutcome, DomainError> {
414        self.enqueue(input).await
415    }
416
417    async fn append_log(
418        &self,
419        job_id: JobId,
420        attempt_no: Option<i32>,
421        level: &str,
422        message: &str,
423        fields: Value,
424    ) -> Result<JobLogRecord, DomainError> {
425        self.append_log(job_id, attempt_no, level, message, fields)
426            .await
427    }
428
429    async fn list_logs(
430        &self,
431        id: JobId,
432        after_id: Option<i64>,
433        limit: u32,
434    ) -> Result<Vec<JobLogRecord>, DomainError> {
435        self.list_logs(id, after_id, limit).await
436    }
437
438    async fn dequeue(
439        &self,
440        queues: &[String],
441        worker_id: &str,
442        lease: Duration,
443    ) -> Result<Option<SbolJob>, DomainError> {
444        self.dequeue(queues, worker_id, lease).await
445    }
446
447    async fn renew_lease(
448        &self,
449        job_id: JobId,
450        worker_id: &str,
451        lease: Duration,
452    ) -> Result<bool, DomainError> {
453        self.renew_lease(job_id, worker_id, lease).await
454    }
455
456    async fn mark_succeeded(
457        &self,
458        job_id: JobId,
459        worker_id: &str,
460        result: Option<Value>,
461    ) -> Result<(), DomainError> {
462        self.mark_succeeded(job_id, worker_id, result).await
463    }
464
465    async fn mark_failed(
466        &self,
467        job_id: JobId,
468        worker_id: &str,
469        error: &str,
470    ) -> Result<JobStatus, DomainError> {
471        self.mark_failed(job_id, worker_id, error).await
472    }
473
474    async fn reap_expired_leases(&self) -> Result<u64, DomainError> {
475        self.reap_expired_leases().await
476    }
477
478    async fn get(&self, id: JobId) -> Result<Option<SbolJob>, DomainError> {
479        self.get(id).await
480    }
481
482    async fn list_attempts(&self, id: JobId) -> Result<Vec<JobAttempt>, DomainError> {
483        self.list_attempts(id).await
484    }
485
486    async fn list(&self, filter: &ListJobsFilter) -> Result<Vec<SbolJob>, DomainError> {
487        self.list(filter).await
488    }
489
490    async fn cancel(&self, id: JobId) -> Result<bool, DomainError> {
491        self.cancel(id).await
492    }
493
494    async fn current_status(&self, id: JobId) -> Result<Option<JobStatus>, DomainError> {
495        self.current_status(id).await
496    }
497
498    async fn queue_depth_snapshot(&self) -> Result<Vec<QueueDepthRow>, DomainError> {
499        self.queue_depth_snapshot().await
500    }
501
502    async fn oldest_queued_age(&self) -> Result<Vec<OldestQueuedAge>, DomainError> {
503        self.oldest_queued_age().await
504    }
505
506    async fn recent_failure_count(&self, within_hours: i64) -> Result<i64, DomainError> {
507        self.recent_failure_count(within_hours).await
508    }
509}
sbol_db_postgres/store_impls.rs

sbol_db_postgres/
store_impls.rs
