sbol_db_postgres/
service.rs

1use sbol_db_core::{DomainError, ImportReport, IriString, SerializationFormat, Triple};
2use sbol_db_derive::{build_import_plan, compose_merged_input, to_rdf_format};
3use sbol_db_rdf::{rdf_graph_to_triples, GRAPH_IRI_PREFIX};
4
5use crate::repo::{
6    AccelRepository, GraphRepository, LabRepository, NeighborhoodRepository, OntologyRepository,
7    ProjectionEvent, ProjectionEventRepository, SbolObjectRepository, SequenceSearchRepository,
8    TripleRepository, TypedProjectionRepository, ValidationRepository,
9};
10use crate::PgPool;
11
12use sbol_db_storage::{GraphWriteMode, ImportInput, ImportOverwrite, TripleScanPage};
13
14pub struct SbolObjectService {
15    pool: PgPool,
16    graphs: GraphRepository,
17    objects: SbolObjectRepository,
18    triples: TripleRepository,
19    accel: AccelRepository,
20    validation: ValidationRepository,
21    projection: ProjectionEventRepository,
22    typed: TypedProjectionRepository,
23    neighborhood: NeighborhoodRepository,
24    sequence_search: SequenceSearchRepository,
25    ontology: OntologyRepository,
26    lab: LabRepository,
27}
28
29/// Per-call cap on rows returned by a Graph Store `GET`. Far above any real
30/// single-graph payload SynBioHub would round-trip; a guard against a
31/// pathological whole-graph serialize.
32const GRAPH_READ_LIMIT: i64 = 5_000_000;
33
34impl SbolObjectService {
35    pub fn new(pool: PgPool) -> Self {
36        Self {
37            graphs: GraphRepository::new(pool.clone()),
38            objects: SbolObjectRepository::new(pool.clone()),
39            triples: TripleRepository::new(pool.clone()),
40            accel: AccelRepository::new(pool.clone(), TripleRepository::new(pool.clone())),
41            validation: ValidationRepository::new(pool.clone()),
42            projection: ProjectionEventRepository::new(pool.clone()),
43            typed: TypedProjectionRepository::new(pool.clone()),
44            neighborhood: NeighborhoodRepository::new(pool.clone()),
45            sequence_search: SequenceSearchRepository::new(pool.clone()),
46            ontology: OntologyRepository::new(pool.clone()),
47            lab: LabRepository::new(pool.clone()),
48            pool,
49        }
50    }
51
52    pub fn lab(&self) -> &LabRepository {
53        &self.lab
54    }
55
56    pub fn pool(&self) -> &PgPool {
57        &self.pool
58    }
59
60    /// Cheap liveness probe against the connection pool. Used by the
61    /// HTTP `/readyz` endpoint; intentionally trivial so it can't itself
62    /// stall on application state.
63    pub async fn ping(&self) -> Result<(), DomainError> {
64        sqlx::query("SELECT 1")
65            .execute(&self.pool)
66            .await
67            .map(|_| ())
68            .map_err(|e| DomainError::Database(e.to_string()))
69    }
70
71    pub fn graphs(&self) -> &GraphRepository {
72        &self.graphs
73    }
74
75    pub fn objects(&self) -> &SbolObjectRepository {
76        &self.objects
77    }
78
79    pub fn triples(&self) -> &TripleRepository {
80        &self.triples
81    }
82
83    pub fn accel(&self) -> &AccelRepository {
84        &self.accel
85    }
86
87    pub fn neighborhood(&self) -> &NeighborhoodRepository {
88        &self.neighborhood
89    }
90
91    pub fn sequence_search(&self) -> &SequenceSearchRepository {
92        &self.sequence_search
93    }
94
95    pub fn ontology(&self) -> &OntologyRepository {
96        &self.ontology
97    }
98
99    /// Atomically import a batch of documents inside one Postgres transaction.
100    /// Either every document commits or none do — there is no half-imported
101    /// state. The implementation is sequential per-document inside the shared
102    /// transaction; the caller controls batch composition. Per-document
103    /// validation runs and projection events are still recorded individually
104    /// (so the batch shows up as N separate document_imported events), but
105    /// they share the outer transaction's atomicity.
106    ///
107    /// Callers wanting partial-success semantics for corpus-scale onboarding
108    /// should fan out to [`import_document`] themselves; the CLI directory
109    /// import is the reference for that pattern.
110    pub async fn import_documents(
111        &self,
112        inputs: Vec<ImportInput>,
113    ) -> Result<Vec<ImportReport>, DomainError> {
114        if inputs.is_empty() {
115            return Ok(Vec::new());
116        }
117        let mut tx = self.pool.begin().await.map_err(db_err)?;
118        let mut reports = Vec::with_capacity(inputs.len());
119        for input in inputs {
120            reports.push(self.import_into_conn(&mut tx, input).await?);
121        }
122        tx.commit().await.map_err(db_err)?;
123        Ok(reports)
124    }
125
126    pub async fn import_document(&self, input: ImportInput) -> Result<ImportReport, DomainError> {
127        let input = self.resolve_merge(input).await?;
128        let mut tx = self.pool.begin().await.map_err(db_err)?;
129        let report = self.import_into_conn(&mut tx, input).await?;
130        tx.commit().await.map_err(db_err)?;
131        Ok(report)
132    }
133
134    /// For a merge, read the existing graph's triples and fold them into the
135    /// incoming document, returning a `Replace` input that carries the union.
136    /// Other overwrite modes pass through unchanged. Reading the old triples
137    /// outside the transaction is safe: the in-transaction replace re-resolves
138    /// and deletes the same graph by `document_iri`.
139    async fn resolve_merge(&self, input: ImportInput) -> Result<ImportInput, DomainError> {
140        if input.overwrite != ImportOverwrite::Merge {
141            return Ok(input);
142        }
143        let Some(document_iri) = input.document_iri.clone() else {
144            return Ok(input);
145        };
146        match self
147            .graphs
148            .id_by_document_iri(document_iri.as_str())
149            .await?
150        {
151            Some(old_id) => {
152                let old_graph_iri = format!("{GRAPH_IRI_PREFIX}{}", old_id.as_uuid());
153                let old_triples = self
154                    .triples
155                    .triples_for_graph(Some(&old_graph_iri), GRAPH_READ_LIMIT)
156                    .await?;
157                compose_merged_input(&old_triples, &input)
158            }
159            None => Ok(ImportInput {
160                overwrite: ImportOverwrite::Replace,
161                ..input
162            }),
163        }
164    }
165
166    /// Graph Store HTTP Protocol write: one ingest mode feeding the shared,
167    /// graph-owned store. Parses `body` as RDF and writes its triples
168    /// **verbatim** into `graph` (registered via [`TripleRepository::ensure_graph`]),
169    /// with no SBOL interpretation at write time. `Merge` appends; `Replace`
170    /// first clears the graph; both run in one transaction. Returns the inserted
171    /// triple count.
172    ///
173    /// This is the same storage substrate [`Self::import_document`] writes to —
174    /// the difference is only how the triples arrive (a posted RDF graph here, a
175    /// parsed/validated SBOL document there). The SBOL3 typed view
176    /// ([`Self::apply_sbol_view`]) is a derivation over a graph's triples; for
177    /// graphs populated this way it is produced by the asynchronous
178    /// reprojection path rather than inline (see the derived-view work).
179    pub async fn graph_store_write(
180        &self,
181        graph: &str,
182        body: &str,
183        format: SerializationFormat,
184        mode: GraphWriteMode,
185    ) -> Result<usize, DomainError> {
186        let rdf_format = to_rdf_format(format)?;
187        let parsed = sbol_rdf::Graph::parse(body, rdf_format)
188            .map_err(|e| DomainError::Parse(e.to_string()))?;
189        let triples = rdf_graph_to_triples(&parsed, &IriString::unchecked(graph));
190
191        let mut tx = self.pool.begin().await.map_err(db_err)?;
192        self.triples
193            .ensure_graph(&mut tx, graph, "verbatim")
194            .await?;
195        if mode == GraphWriteMode::Replace {
196            self.triples.clear_graph(&mut tx, Some(graph)).await?;
197        }
198        let inserted = self
199            .triples
200            .insert_triples(&mut tx, &triples, "graph-store")
201            .await?;
202        self.accel.refresh_graph(&mut tx, graph).await?;
203        tx.commit().await.map_err(db_err)?;
204        Ok(inserted)
205    }
206
207    /// Graph Store CRUD `DELETE`: drop a named graph entirely (its triples and
208    /// `sbol_graphs` registry row). Returns the number of triples removed.
209    pub async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError> {
210        let mut tx = self.pool.begin().await.map_err(db_err)?;
211        let deleted = self.triples.clear_graph(&mut tx, Some(graph)).await?;
212        self.triples.delete_graph(&mut tx, graph).await?;
213        self.accel.refresh_graph(&mut tx, graph).await?;
214        tx.commit().await.map_err(db_err)?;
215        Ok(deleted)
216    }
217
218    /// Graph Store CRUD `GET`: read back a named graph's triples.
219    pub async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError> {
220        self.triples
221            .triples_for_graph(Some(graph), GRAPH_READ_LIMIT)
222            .await
223    }
224
225    pub async fn graph_store_read_page(
226        &self,
227        graph: &str,
228        after: Option<&str>,
229        limit: usize,
230    ) -> Result<TripleScanPage, DomainError> {
231        let after_id = after
232            .map(str::parse::<i64>)
233            .transpose()
234            .map_err(|_| DomainError::InvalidInput("invalid Postgres graph cursor".to_owned()))?
235            .unwrap_or(0);
236        let rows = self
237            .triples
238            .scan_graph_page(graph, after_id, i64::try_from(limit).unwrap_or(i64::MAX))
239            .await?;
240        let next_cursor = (rows.len() == limit)
241            .then(|| rows.last().map(|(id, _)| id.to_string()))
242            .flatten();
243        Ok(TripleScanPage {
244            items: rows.into_iter().map(|(_, triple)| triple).collect(),
245            next_cursor,
246        })
247    }
248
249    async fn import_into_conn(
250        &self,
251        conn: &mut sqlx::PgConnection,
252        input: ImportInput,
253    ) -> Result<ImportReport, DomainError> {
254        // Replace (and the merge path, which resolves to Replace) drops the
255        // prior graph carrying this document IRI in the same transaction, so
256        // the re-import does not collide with the `document_iri` unique
257        // constraint and no half-replaced state is observable.
258        if input.overwrite == ImportOverwrite::Replace {
259            if let Some(document_iri) = &input.document_iri {
260                if let Some(old_id) = self
261                    .graphs
262                    .id_by_document_iri_in(&mut *conn, document_iri.as_str())
263                    .await?
264                {
265                    self.graphs.delete_in(&mut *conn, old_id).await?;
266                }
267            }
268        }
269
270        let plan = build_import_plan(&input)?;
271
272        // A document is an `sbol3`-kind graph that owns its triples. Register
273        // the graph row (under the id minted in the plan), then write the
274        // document's triples and derived SBOL view into it.
275        self.graphs
276            .insert(&mut *conn, plan.graph_id, plan.new_graph)
277            .await?;
278
279        let triple_count = self
280            .triples
281            .insert_triples(&mut *conn, &plan.triples, "sbol")
282            .await?;
283        self.accel
284            .refresh_graph(&mut *conn, plan.graph_iri.as_str())
285            .await?;
286
287        let object_count = plan.summaries.len();
288        for summary in &plan.summaries {
289            self.objects
290                .upsert(&mut *conn, summary, Some(plan.graph_id))
291                .await?;
292        }
293        let typed_counts = self.typed.upsert_all(&mut *conn, &plan.projections).await?;
294        let recorded = self
295            .validation
296            .record_run(
297                &mut *conn,
298                &plan.target_iri,
299                Some(plan.graph_id),
300                "sbol-rs",
301                Some(sbol::v3::SPEC_VERSION),
302                "sbol3-3.1.0",
303                &plan.validation,
304            )
305            .await?;
306
307        self.projection
308            .append(
309                &mut *conn,
310                ProjectionEvent {
311                    event_type: "document_imported".to_owned(),
312                    subject_iri: Some(IriString::unchecked(plan.target_iri.clone())),
313                    graph_iri: Some(plan.graph_iri.clone()),
314                    payload: serde_json::json!({
315                        "graph_id": plan.graph_id.as_uuid(),
316                        "object_count": object_count,
317                        "triple_count": triple_count,
318                        "typed_counts": typed_counts,
319                    }),
320                },
321            )
322            .await?;
323
324        Ok(ImportReport {
325            graph_id: plan.graph_id,
326            object_count,
327            triple_count,
328            validation_status: recorded.status,
329            validation_issue_count: recorded.issue_count,
330        })
331    }
332}
333
334fn db_err<E: std::fmt::Display>(e: E) -> DomainError {
335    DomainError::Database(e.to_string())
336}
sbol_db_postgres/service.rs

sbol_db_postgres/
service.rs
