sbol_db_storage/
traits.rs

1//! The storage contract: traits a persistence backend implements.
2//!
3//! Async traits use `#[async_trait]` so they are object-safe and can be held
4//! as `Arc<dyn ...>`. [`TripleSource`] is deliberately synchronous: it backs
5//! the SPARQL evaluator's synchronous `QueryableDataset`, and a backend that
6//! needs async runs it to completion internally.
7
8use std::time::Duration;
9
10use async_trait::async_trait;
11use sbol_db_core::{
12    DomainError, GraphId, GraphRecord, ImportReport, JobId, NeighborhoodQuery, NeighborhoodResult,
13    ObjectId, SbolObjectRecord, SerializationFormat, Triple,
14};
15use serde_json::Value;
16
17use crate::{
18    AccelSolutions, AcceleratedQuery, BatchSequenceMatch, EnqueueOutcome, GraphFilter,
19    GraphWriteMode, IdGraphFilter, IdQuad, ImportInput, JobAttempt, JobLogRecord, JobStatus,
20    ListGraphsFilter, ListJobsFilter, ListObjectsFilter, NewJob, OldestQueuedAge,
21    OntologyLoadReport, OntologyRecord, OntologyTermRecord, PatternObject, PatternSubject,
22    QueueDepthRow, SbolJob, SequenceMatch, SequenceSearchOptions, TermId, TermKey, TermValue,
23    TripleChange, UpdateOutcome,
24};
25
26/// Synchronous triple-pattern reads, as required by the SPARQL evaluator.
27pub trait TripleSource: Send + Sync {
28    /// Scan triples matching a pattern. Any position may be bound or wildcarded
29    /// (`None`); `limit` caps the rows returned per call.
30    fn scan_pattern(
31        &self,
32        subject: Option<&PatternSubject>,
33        predicate: Option<&str>,
34        object: Option<&PatternObject>,
35        graph: Option<&GraphFilter>,
36        limit: i64,
37    ) -> Result<Vec<Triple>, DomainError>;
38
39    /// Distinct named graphs present in the store.
40    fn distinct_named_graphs(&self) -> Result<Vec<String>, DomainError>;
41
42    /// Every triple in one named graph (`Some`) or the default partition
43    /// (`None`), capped at `limit`.
44    fn triples_for_graph(
45        &self,
46        graph: Option<&str>,
47        limit: i64,
48    ) -> Result<Vec<Triple>, DomainError>;
49
50    /// Every triple with the given subject IRI.
51    fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError>;
52
53    /// Whether this backend can scan by term id ([`Self::id_scan`]). An id-native
54    /// backend lets the SPARQL evaluator join on term ids and materialize terms
55    /// only for output rows; the default is `false` and the evaluator uses the
56    /// term-materializing [`Self::scan_pattern`] path.
57    fn supports_id_scan(&self) -> bool {
58        false
59    }
60
61    /// Scan a pattern returning term ids rather than materialized triples. Bound
62    /// positions are given as ids (see [`Self::term_to_id`]). Only meaningful when
63    /// [`Self::supports_id_scan`] is true.
64    fn id_scan(
65        &self,
66        _subject: Option<TermId>,
67        _predicate: Option<TermId>,
68        _object: Option<TermId>,
69        _graph: &IdGraphFilter,
70        _limit: i64,
71    ) -> Result<Vec<IdQuad>, DomainError> {
72        Err(DomainError::Database(
73            "id_scan is not supported by this backend".into(),
74        ))
75    }
76
77    /// Resolve a term to its id (the same id stored in the indexes), for binding
78    /// query constants into an id-native scan.
79    fn term_to_id(&self, _key: TermKey<'_>) -> Result<TermId, DomainError> {
80        Err(DomainError::Database(
81            "term_to_id is not supported by this backend".into(),
82        ))
83    }
84
85    /// Resolve a term id back to its value, for materializing output rows and
86    /// filter operands.
87    fn id_to_term(&self, _id: TermId) -> Result<TermValue, DomainError> {
88        Err(DomainError::Database(
89            "id_to_term is not supported by this backend".into(),
90        ))
91    }
92
93    /// Answer a recognized SynBioHub query from purpose-built indexes. Returns
94    /// `Ok(None)` when this backend has no accelerator or declines the query, so
95    /// the caller falls back to generic SPARQL evaluation.
96    fn run_accelerated(
97        &self,
98        _query: &AcceleratedQuery,
99    ) -> Result<Option<AccelSolutions>, DomainError> {
100        Ok(None)
101    }
102}
103
104/// Atomic batch application of SPARQL-update changes.
105#[async_trait]
106pub trait TripleWriter: Send + Sync {
107    /// Apply every change in one atomic unit, registering any named graph an
108    /// insert targets before writing it. Returns the inserted/deleted tally.
109    async fn apply_update(&self, changes: Vec<TripleChange>) -> Result<UpdateOutcome, DomainError>;
110}
111
112/// Derived-view object reads.
113#[async_trait]
114pub trait ObjectStore: Send + Sync {
115    async fn get_object_by_iri(&self, iri: &str) -> Result<Option<SbolObjectRecord>, DomainError>;
116    async fn get_objects_by_iris(
117        &self,
118        iris: &[&str],
119    ) -> Result<Vec<SbolObjectRecord>, DomainError>;
120    async fn list_objects(
121        &self,
122        filter: &ListObjectsFilter,
123    ) -> Result<Vec<SbolObjectRecord>, DomainError>;
124    async fn get_object_iri_by_id(&self, id: ObjectId) -> Result<Option<String>, DomainError>;
125}
126
127/// Document-graph reads and deletion.
128#[async_trait]
129pub trait GraphStore: Send + Sync {
130    async fn get_graph(&self, id: GraphId) -> Result<Option<GraphRecord>, DomainError>;
131    async fn list_graphs(&self, filter: &ListGraphsFilter)
132        -> Result<Vec<GraphRecord>, DomainError>;
133    async fn delete_graph(&self, id: GraphId) -> Result<bool, DomainError>;
134    async fn graph_exists_by_hash(&self, hash: &[u8]) -> Result<bool, DomainError>;
135}
136
137/// Ontology loading and lookup.
138#[async_trait]
139pub trait OntologyStore: Send + Sync {
140    async fn load_ontology_from_url(
141        &self,
142        prefix: &str,
143        name: &str,
144        source_url: &str,
145    ) -> Result<OntologyLoadReport, DomainError>;
146    async fn load_ontology_from_text(
147        &self,
148        prefix: &str,
149        name: &str,
150        source_url: Option<&str>,
151        text: &str,
152    ) -> Result<OntologyLoadReport, DomainError>;
153    async fn list_ontologies(&self) -> Result<Vec<OntologyRecord>, DomainError>;
154    async fn canonicalize(&self, iri: &str) -> Result<Option<String>, DomainError>;
155    async fn descendants(&self, iri: &str) -> Result<Vec<(String, i16)>, DomainError>;
156    async fn list_ontology_terms(
157        &self,
158        prefix: &str,
159        limit: i64,
160        offset: i64,
161        search: Option<&str>,
162    ) -> Result<(Vec<OntologyTermRecord>, i64), DomainError>;
163    async fn get_ontology_term(&self, iri: &str)
164        -> Result<Option<OntologyTermRecord>, DomainError>;
165}
166
167/// Graph-neighborhood traversal.
168#[async_trait]
169pub trait NeighborhoodStore: Send + Sync {
170    async fn walk(&self, query: &NeighborhoodQuery) -> Result<NeighborhoodResult, DomainError>;
171}
172
173/// Nucleotide sequence search over the derived view.
174#[async_trait]
175pub trait SequenceSearchStore: Send + Sync {
176    async fn search(
177        &self,
178        pattern: &str,
179        options: SequenceSearchOptions,
180    ) -> Result<Vec<SequenceMatch>, DomainError>;
181    async fn search_many(
182        &self,
183        patterns: &[String],
184        options: SequenceSearchOptions,
185    ) -> Result<Vec<BatchSequenceMatch>, DomainError>;
186}
187
188/// The full SBOL-aware store: ingest plus every derived-view read surface.
189#[async_trait]
190pub trait SbolStore:
191    ObjectStore + GraphStore + OntologyStore + NeighborhoodStore + SequenceSearchStore
192{
193    async fn import_document(&self, input: ImportInput) -> Result<ImportReport, DomainError>;
194    async fn import_documents(
195        &self,
196        inputs: Vec<ImportInput>,
197    ) -> Result<Vec<ImportReport>, DomainError>;
198    async fn graph_store_write(
199        &self,
200        graph: &str,
201        body: &str,
202        format: SerializationFormat,
203        mode: GraphWriteMode,
204    ) -> Result<usize, DomainError>;
205    async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError>;
206    async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError>;
207    /// Every triple with the given subject IRI, for single-object RDF export.
208    async fn triples_for_subject(&self, subject_iri: &str) -> Result<Vec<Triple>, DomainError>;
209    async fn ping(&self) -> Result<(), DomainError>;
210}
211
212/// The job queue: enqueue, lease-based dequeue, lifecycle transitions, and the
213/// operator/observability read surface.
214#[async_trait]
215pub trait JobQueue: Send + Sync {
216    async fn enqueue(&self, input: NewJob) -> Result<EnqueueOutcome, DomainError>;
217    async fn append_log(
218        &self,
219        job_id: JobId,
220        attempt_no: Option<i32>,
221        level: &str,
222        message: &str,
223        fields: Value,
224    ) -> Result<JobLogRecord, DomainError>;
225    async fn list_logs(
226        &self,
227        id: JobId,
228        after_id: Option<i64>,
229        limit: u32,
230    ) -> Result<Vec<JobLogRecord>, DomainError>;
231    async fn dequeue(
232        &self,
233        queues: &[String],
234        worker_id: &str,
235        lease: Duration,
236    ) -> Result<Option<SbolJob>, DomainError>;
237    async fn renew_lease(
238        &self,
239        job_id: JobId,
240        worker_id: &str,
241        lease: Duration,
242    ) -> Result<bool, DomainError>;
243    async fn mark_succeeded(
244        &self,
245        job_id: JobId,
246        worker_id: &str,
247        result: Option<Value>,
248    ) -> Result<(), DomainError>;
249    async fn mark_failed(
250        &self,
251        job_id: JobId,
252        worker_id: &str,
253        error: &str,
254    ) -> Result<JobStatus, DomainError>;
255    async fn reap_expired_leases(&self) -> Result<u64, DomainError>;
256    async fn get(&self, id: JobId) -> Result<Option<SbolJob>, DomainError>;
257    async fn list_attempts(&self, id: JobId) -> Result<Vec<JobAttempt>, DomainError>;
258    async fn list(&self, filter: &ListJobsFilter) -> Result<Vec<SbolJob>, DomainError>;
259    async fn cancel(&self, id: JobId) -> Result<bool, DomainError>;
260    async fn current_status(&self, id: JobId) -> Result<Option<JobStatus>, DomainError>;
261    async fn queue_depth_snapshot(&self) -> Result<Vec<QueueDepthRow>, DomainError>;
262    async fn oldest_queued_age(&self) -> Result<Vec<OldestQueuedAge>, DomainError>;
263
264    /// Count jobs that reached a terminal failure (`failed` or `dead`) in the
265    /// last `within_hours`. Backed by [`Self::list`] so every backend reports
266    /// the same number without backend-specific SQL; the dashboard's
267    /// failures tile reads it.
268    async fn recent_failure_count(&self, within_hours: i64) -> Result<i64, DomainError> {
269        let since = chrono::Utc::now() - chrono::Duration::hours(within_hours.max(0));
270        let mut total = 0i64;
271        for status in [JobStatus::Failed, JobStatus::Dead] {
272            let filter = ListJobsFilter {
273                kind: None,
274                status: Some(status),
275                queue: None,
276                correlation_id: None,
277                since: Some(since),
278                limit: 10_000,
279            };
280            total += self.list(&filter).await?.len() as i64;
281        }
282        Ok(total)
283    }
284}
sbol_db_storage/traits.rs

sbol_db_storage/
traits.rs
