sbol_db_postgres/
service.rs

1use sbol_db_core::{DomainError, ImportReport, IriString, SerializationFormat, Triple};
2use sbol_db_derive::{build_import_plan, to_rdf_format};
3use sbol_db_rdf::rdf_graph_to_triples;
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};
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 mut tx = self.pool.begin().await.map_err(db_err)?;
128        let report = self.import_into_conn(&mut tx, input).await?;
129        tx.commit().await.map_err(db_err)?;
130        Ok(report)
131    }
132
133    /// Graph Store HTTP Protocol write: one ingest mode feeding the shared,
134    /// graph-owned store. Parses `body` as RDF and writes its triples
135    /// **verbatim** into `graph` (registered via [`TripleRepository::ensure_graph`]),
136    /// with no SBOL interpretation at write time. `Merge` appends; `Replace`
137    /// first clears the graph; both run in one transaction. Returns the inserted
138    /// triple count.
139    ///
140    /// This is the same storage substrate [`Self::import_document`] writes to —
141    /// the difference is only how the triples arrive (a posted RDF graph here, a
142    /// parsed/validated SBOL document there). The SBOL3 typed view
143    /// ([`Self::apply_sbol_view`]) is a derivation over a graph's triples; for
144    /// graphs populated this way it is produced by the asynchronous
145    /// reprojection path rather than inline (see the derived-view work).
146    pub async fn graph_store_write(
147        &self,
148        graph: &str,
149        body: &str,
150        format: SerializationFormat,
151        mode: GraphWriteMode,
152    ) -> Result<usize, DomainError> {
153        let rdf_format = to_rdf_format(format)?;
154        let parsed = sbol_rdf::Graph::parse(body, rdf_format)
155            .map_err(|e| DomainError::Parse(e.to_string()))?;
156        let triples = rdf_graph_to_triples(&parsed, &IriString::unchecked(graph));
157
158        let mut tx = self.pool.begin().await.map_err(db_err)?;
159        self.triples
160            .ensure_graph(&mut tx, graph, "verbatim")
161            .await?;
162        if mode == GraphWriteMode::Replace {
163            self.triples.clear_graph(&mut tx, Some(graph)).await?;
164        }
165        let inserted = self
166            .triples
167            .insert_triples(&mut tx, &triples, "graph-store")
168            .await?;
169        self.accel.refresh_graph(&mut tx, graph).await?;
170        tx.commit().await.map_err(db_err)?;
171        Ok(inserted)
172    }
173
174    /// Graph Store CRUD `DELETE`: drop a named graph entirely (its triples and
175    /// `sbol_graphs` registry row). Returns the number of triples removed.
176    pub async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError> {
177        let mut tx = self.pool.begin().await.map_err(db_err)?;
178        let deleted = self.triples.clear_graph(&mut tx, Some(graph)).await?;
179        self.triples.delete_graph(&mut tx, graph).await?;
180        self.accel.refresh_graph(&mut tx, graph).await?;
181        tx.commit().await.map_err(db_err)?;
182        Ok(deleted)
183    }
184
185    /// Graph Store CRUD `GET`: read back a named graph's triples.
186    pub async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError> {
187        self.triples
188            .triples_for_graph(Some(graph), GRAPH_READ_LIMIT)
189            .await
190    }
191
192    async fn import_into_conn(
193        &self,
194        conn: &mut sqlx::PgConnection,
195        input: ImportInput,
196    ) -> Result<ImportReport, DomainError> {
197        let plan = build_import_plan(&input)?;
198
199        // A document is an `sbol3`-kind graph that owns its triples. Register
200        // the graph row (under the id minted in the plan), then write the
201        // document's triples and derived SBOL view into it.
202        self.graphs
203            .insert(&mut *conn, plan.graph_id, plan.new_graph)
204            .await?;
205
206        let triple_count = self
207            .triples
208            .insert_triples(&mut *conn, &plan.triples, "sbol")
209            .await?;
210        self.accel
211            .refresh_graph(&mut *conn, plan.graph_iri.as_str())
212            .await?;
213
214        let object_count = plan.summaries.len();
215        for summary in &plan.summaries {
216            self.objects
217                .upsert(&mut *conn, summary, Some(plan.graph_id))
218                .await?;
219        }
220        let typed_counts = self.typed.upsert_all(&mut *conn, &plan.projections).await?;
221        let recorded = self
222            .validation
223            .record_run(
224                &mut *conn,
225                &plan.target_iri,
226                Some(plan.graph_id),
227                "sbol-rs",
228                Some(sbol::SPEC_VERSION),
229                "sbol3-3.1.0",
230                &plan.validation,
231            )
232            .await?;
233
234        self.projection
235            .append(
236                &mut *conn,
237                ProjectionEvent {
238                    event_type: "document_imported".to_owned(),
239                    subject_iri: Some(IriString::unchecked(plan.target_iri.clone())),
240                    graph_iri: Some(plan.graph_iri.clone()),
241                    payload: serde_json::json!({
242                        "graph_id": plan.graph_id.as_uuid(),
243                        "object_count": object_count,
244                        "triple_count": triple_count,
245                        "typed_counts": typed_counts,
246                    }),
247                },
248            )
249            .await?;
250
251        Ok(ImportReport {
252            graph_id: plan.graph_id,
253            object_count,
254            triple_count,
255            validation_status: recorded.status,
256            validation_issue_count: recorded.issue_count,
257        })
258    }
259}
260
261fn db_err<E: std::fmt::Display>(e: E) -> DomainError {
262    DomainError::Database(e.to_string())
263}
sbol_db_postgres/service.rs

sbol_db_postgres/
service.rs
