sbol-db-postgres 0.1.8

use sbol_db_core::{DomainError, ImportReport, IriString, SerializationFormat, Triple};
use sbol_db_derive::{build_import_plan, compose_merged_input, to_rdf_format};
use sbol_db_rdf::{rdf_graph_to_triples, GRAPH_IRI_PREFIX};

use crate::repo::{
    AccelRepository, GraphRepository, LabRepository, NeighborhoodRepository, OntologyRepository,
    ProjectionEvent, ProjectionEventRepository, SbolObjectRepository, SequenceSearchRepository,
    TripleRepository, TypedProjectionRepository, ValidationRepository,
};
use crate::PgPool;

use sbol_db_storage::{GraphWriteMode, ImportInput, ImportOverwrite, TripleScanPage};

pub struct SbolObjectService {
    pool: PgPool,
    graphs: GraphRepository,
    objects: SbolObjectRepository,
    triples: TripleRepository,
    accel: AccelRepository,
    validation: ValidationRepository,
    projection: ProjectionEventRepository,
    typed: TypedProjectionRepository,
    neighborhood: NeighborhoodRepository,
    sequence_search: SequenceSearchRepository,
    ontology: OntologyRepository,
    lab: LabRepository,
}

/// Per-call cap on rows returned by a Graph Store `GET`. Far above any real
/// single-graph payload SynBioHub would round-trip; a guard against a
/// pathological whole-graph serialize.
const GRAPH_READ_LIMIT: i64 = 5_000_000;

impl SbolObjectService {
    pub fn new(pool: PgPool) -> Self {
        Self {
            graphs: GraphRepository::new(pool.clone()),
            objects: SbolObjectRepository::new(pool.clone()),
            triples: TripleRepository::new(pool.clone()),
            accel: AccelRepository::new(pool.clone(), TripleRepository::new(pool.clone())),
            validation: ValidationRepository::new(pool.clone()),
            projection: ProjectionEventRepository::new(pool.clone()),
            typed: TypedProjectionRepository::new(pool.clone()),
            neighborhood: NeighborhoodRepository::new(pool.clone()),
            sequence_search: SequenceSearchRepository::new(pool.clone()),
            ontology: OntologyRepository::new(pool.clone()),
            lab: LabRepository::new(pool.clone()),
            pool,
        }
    }

    pub fn lab(&self) -> &LabRepository {
        &self.lab
    }

    pub fn pool(&self) -> &PgPool {
        &self.pool
    }

    /// Cheap liveness probe against the connection pool. Used by the
    /// HTTP `/readyz` endpoint; intentionally trivial so it can't itself
    /// stall on application state.
    pub async fn ping(&self) -> Result<(), DomainError> {
        sqlx::query("SELECT 1")
            .execute(&self.pool)
            .await
            .map(|_| ())
            .map_err(|e| DomainError::Database(e.to_string()))
    }

    pub fn graphs(&self) -> &GraphRepository {
        &self.graphs
    }

    pub fn objects(&self) -> &SbolObjectRepository {
        &self.objects
    }

    pub fn triples(&self) -> &TripleRepository {
        &self.triples
    }

    pub fn accel(&self) -> &AccelRepository {
        &self.accel
    }

    pub fn neighborhood(&self) -> &NeighborhoodRepository {
        &self.neighborhood
    }

    pub fn sequence_search(&self) -> &SequenceSearchRepository {
        &self.sequence_search
    }

    pub fn ontology(&self) -> &OntologyRepository {
        &self.ontology
    }

    /// Atomically import a batch of documents inside one Postgres transaction.
    /// Either every document commits or none do — there is no half-imported
    /// state. The implementation is sequential per-document inside the shared
    /// transaction; the caller controls batch composition. Per-document
    /// validation runs and projection events are still recorded individually
    /// (so the batch shows up as N separate document_imported events), but
    /// they share the outer transaction's atomicity.
    ///
    /// Callers wanting partial-success semantics for corpus-scale onboarding
    /// should fan out to [`import_document`] themselves; the CLI directory
    /// import is the reference for that pattern.
    pub async fn import_documents(
        &self,
        inputs: Vec<ImportInput>,
    ) -> Result<Vec<ImportReport>, DomainError> {
        if inputs.is_empty() {
            return Ok(Vec::new());
        }
        let mut tx = self.pool.begin().await.map_err(db_err)?;
        let mut reports = Vec::with_capacity(inputs.len());
        for input in inputs {
            reports.push(self.import_into_conn(&mut tx, input).await?);
        }
        tx.commit().await.map_err(db_err)?;
        Ok(reports)
    }

    pub async fn import_document(&self, input: ImportInput) -> Result<ImportReport, DomainError> {
        let input = self.resolve_merge(input).await?;
        let mut tx = self.pool.begin().await.map_err(db_err)?;
        let report = self.import_into_conn(&mut tx, input).await?;
        tx.commit().await.map_err(db_err)?;
        Ok(report)
    }

    /// For a merge, read the existing graph's triples and fold them into the
    /// incoming document, returning a `Replace` input that carries the union.
    /// Other overwrite modes pass through unchanged. Reading the old triples
    /// outside the transaction is safe: the in-transaction replace re-resolves
    /// and deletes the same graph by `document_iri`.
    async fn resolve_merge(&self, input: ImportInput) -> Result<ImportInput, DomainError> {
        if input.overwrite != ImportOverwrite::Merge {
            return Ok(input);
        }
        let Some(document_iri) = input.document_iri.clone() else {
            return Ok(input);
        };
        match self
            .graphs
            .id_by_document_iri(document_iri.as_str())
            .await?
        {
            Some(old_id) => {
                let old_graph_iri = format!("{GRAPH_IRI_PREFIX}{}", old_id.as_uuid());
                let old_triples = self
                    .triples
                    .triples_for_graph(Some(&old_graph_iri), GRAPH_READ_LIMIT)
                    .await?;
                compose_merged_input(&old_triples, &input)
            }
            None => Ok(ImportInput {
                overwrite: ImportOverwrite::Replace,
                ..input
            }),
        }
    }

    /// Graph Store HTTP Protocol write: one ingest mode feeding the shared,
    /// graph-owned store. Parses `body` as RDF and writes its triples
    /// **verbatim** into `graph` (registered via [`TripleRepository::ensure_graph`]),
    /// with no SBOL interpretation at write time. `Merge` appends; `Replace`
    /// first clears the graph; both run in one transaction. Returns the inserted
    /// triple count.
    ///
    /// This is the same storage substrate [`Self::import_document`] writes to —
    /// the difference is only how the triples arrive (a posted RDF graph here, a
    /// parsed/validated SBOL document there). The SBOL3 typed view
    /// ([`Self::apply_sbol_view`]) is a derivation over a graph's triples; for
    /// graphs populated this way it is produced by the asynchronous
    /// reprojection path rather than inline (see the derived-view work).
    pub async fn graph_store_write(
        &self,
        graph: &str,
        body: &str,
        format: SerializationFormat,
        mode: GraphWriteMode,
    ) -> Result<usize, DomainError> {
        let rdf_format = to_rdf_format(format)?;
        let parsed = sbol_rdf::Graph::parse(body, rdf_format)
            .map_err(|e| DomainError::Parse(e.to_string()))?;
        let triples = rdf_graph_to_triples(&parsed, &IriString::unchecked(graph));

        let mut tx = self.pool.begin().await.map_err(db_err)?;
        self.triples
            .ensure_graph(&mut tx, graph, "verbatim")
            .await?;
        if mode == GraphWriteMode::Replace {
            self.triples.clear_graph(&mut tx, Some(graph)).await?;
        }
        let inserted = self
            .triples
            .insert_triples(&mut tx, &triples, "graph-store")
            .await?;
        self.accel.refresh_graph(&mut tx, graph).await?;
        tx.commit().await.map_err(db_err)?;
        Ok(inserted)
    }

    /// Graph Store CRUD `DELETE`: drop a named graph entirely (its triples and
    /// `sbol_graphs` registry row). Returns the number of triples removed.
    pub async fn graph_store_clear(&self, graph: &str) -> Result<usize, DomainError> {
        let mut tx = self.pool.begin().await.map_err(db_err)?;
        let deleted = self.triples.clear_graph(&mut tx, Some(graph)).await?;
        self.triples.delete_graph(&mut tx, graph).await?;
        self.accel.refresh_graph(&mut tx, graph).await?;
        tx.commit().await.map_err(db_err)?;
        Ok(deleted)
    }

    /// Graph Store CRUD `GET`: read back a named graph's triples.
    pub async fn graph_store_read(&self, graph: &str) -> Result<Vec<Triple>, DomainError> {
        self.triples
            .triples_for_graph(Some(graph), GRAPH_READ_LIMIT)
            .await
    }

    pub async fn graph_store_read_page(
        &self,
        graph: &str,
        after: Option<&str>,
        limit: usize,
    ) -> Result<TripleScanPage, DomainError> {
        let after_id = after
            .map(str::parse::<i64>)
            .transpose()
            .map_err(|_| DomainError::InvalidInput("invalid Postgres graph cursor".to_owned()))?
            .unwrap_or(0);
        let rows = self
            .triples
            .scan_graph_page(graph, after_id, i64::try_from(limit).unwrap_or(i64::MAX))
            .await?;
        let next_cursor = (rows.len() == limit)
            .then(|| rows.last().map(|(id, _)| id.to_string()))
            .flatten();
        Ok(TripleScanPage {
            items: rows.into_iter().map(|(_, triple)| triple).collect(),
            next_cursor,
        })
    }

    async fn import_into_conn(
        &self,
        conn: &mut sqlx::PgConnection,
        input: ImportInput,
    ) -> Result<ImportReport, DomainError> {
        // Replace (and the merge path, which resolves to Replace) drops the
        // prior graph carrying this document IRI in the same transaction, so
        // the re-import does not collide with the `document_iri` unique
        // constraint and no half-replaced state is observable.
        if input.overwrite == ImportOverwrite::Replace {
            if let Some(document_iri) = &input.document_iri {
                if let Some(old_id) = self
                    .graphs
                    .id_by_document_iri_in(&mut *conn, document_iri.as_str())
                    .await?
                {
                    self.graphs.delete_in(&mut *conn, old_id).await?;
                }
            }
        }

        let plan = build_import_plan(&input)?;

        // A document is an `sbol3`-kind graph that owns its triples. Register
        // the graph row (under the id minted in the plan), then write the
        // document's triples and derived SBOL view into it.
        self.graphs
            .insert(&mut *conn, plan.graph_id, plan.new_graph)
            .await?;

        let triple_count = self
            .triples
            .insert_triples(&mut *conn, &plan.triples, "sbol")
            .await?;
        self.accel
            .refresh_graph(&mut *conn, plan.graph_iri.as_str())
            .await?;

        let object_count = plan.summaries.len();
        for summary in &plan.summaries {
            self.objects
                .upsert(&mut *conn, summary, Some(plan.graph_id))
                .await?;
        }
        let typed_counts = self.typed.upsert_all(&mut *conn, &plan.projections).await?;
        let recorded = self
            .validation
            .record_run(
                &mut *conn,
                &plan.target_iri,
                Some(plan.graph_id),
                "sbol-rs",
                Some(sbol::v3::SPEC_VERSION),
                "sbol3-3.1.0",
                &plan.validation,
            )
            .await?;

        self.projection
            .append(
                &mut *conn,
                ProjectionEvent {
                    event_type: "document_imported".to_owned(),
                    subject_iri: Some(IriString::unchecked(plan.target_iri.clone())),
                    graph_iri: Some(plan.graph_iri.clone()),
                    payload: serde_json::json!({
                        "graph_id": plan.graph_id.as_uuid(),
                        "object_count": object_count,
                        "triple_count": triple_count,
                        "typed_counts": typed_counts,
                    }),
                },
            )
            .await?;

        Ok(ImportReport {
            graph_id: plan.graph_id,
            object_count,
            triple_count,
            validation_status: recorded.status,
            validation_issue_count: recorded.issue_count,
        })
    }
}

fn db_err<E: std::fmt::Display>(e: E) -> DomainError {
    DomainError::Database(e.to_string())
}