sbol_db_postgres/repo/
cluster.rs

1//! The sequence cluster store over Postgres.
2//!
3//! Assignments live in `sbol_sequence_cluster`. A rebuild replaces the whole
4//! table in one transaction (`TRUNCATE` then a single `UNNEST` insert), so
5//! readers only ever see a complete clustering.
6
7use async_trait::async_trait;
8use sbol_db_core::DomainError;
9use sbol_db_storage::{ClusterId, ClusterStore};
10use sqlx::Row;
11
12use crate::repo::db_err;
13use crate::PgPool;
14
15#[derive(Clone)]
16pub struct PgClusterStore {
17    pool: PgPool,
18}
19
20impl PgClusterStore {
21    pub fn new(pool: PgPool) -> Self {
22        Self { pool }
23    }
24}
25
26#[async_trait]
27impl ClusterStore for PgClusterStore {
28    async fn cluster_id_of(&self, iri: &str) -> Result<Option<ClusterId>, DomainError> {
29        let row =
30            sqlx::query("SELECT cluster_id FROM sbol_sequence_cluster WHERE sequence_iri = $1")
31                .bind(iri)
32                .fetch_optional(&self.pool)
33                .await
34                .map_err(db_err)?;
35        match row {
36            Some(r) => Ok(Some(ClusterId(
37                r.try_get::<i64, _>("cluster_id").map_err(db_err)?,
38            ))),
39            None => Ok(None),
40        }
41    }
42
43    async fn cluster_mates(&self, iri: &str) -> Result<Vec<String>, DomainError> {
44        let rows = sqlx::query(
45            r#"
46            SELECT mate.sequence_iri AS iri
47            FROM sbol_sequence_cluster self_row
48            JOIN sbol_sequence_cluster mate ON mate.cluster_id = self_row.cluster_id
49            WHERE self_row.sequence_iri = $1
50              AND mate.sequence_iri <> $1
51            "#,
52        )
53        .bind(iri)
54        .fetch_all(&self.pool)
55        .await
56        .map_err(db_err)?;
57        let mut out = Vec::with_capacity(rows.len());
58        for r in rows {
59            out.push(r.try_get::<String, _>("iri").map_err(db_err)?);
60        }
61        Ok(out)
62    }
63
64    async fn replace_clusters(&self, pairs: Vec<(String, ClusterId)>) -> Result<(), DomainError> {
65        let mut tx = self.pool.begin().await.map_err(db_err)?;
66        sqlx::query("TRUNCATE sbol_sequence_cluster")
67            .execute(&mut *tx)
68            .await
69            .map_err(db_err)?;
70        if !pairs.is_empty() {
71            let iris: Vec<String> = pairs.iter().map(|(iri, _)| iri.clone()).collect();
72            let cluster_ids: Vec<i64> = pairs.iter().map(|(_, c)| c.0).collect();
73            sqlx::query(
74                r#"
75                INSERT INTO sbol_sequence_cluster (sequence_iri, cluster_id)
76                SELECT * FROM UNNEST($1::text[], $2::bigint[])
77                ON CONFLICT (sequence_iri) DO UPDATE SET cluster_id = EXCLUDED.cluster_id
78                "#,
79            )
80            .bind(&iris)
81            .bind(&cluster_ids)
82            .execute(&mut *tx)
83            .await
84            .map_err(db_err)?;
85        }
86        tx.commit().await.map_err(db_err)?;
87        Ok(())
88    }
89
90    async fn all_assignments(&self) -> Result<Vec<(String, ClusterId)>, DomainError> {
91        let rows = sqlx::query("SELECT sequence_iri, cluster_id FROM sbol_sequence_cluster")
92            .fetch_all(&self.pool)
93            .await
94            .map_err(db_err)?;
95        let mut out = Vec::with_capacity(rows.len());
96        for r in rows {
97            let iri = r.try_get::<String, _>("sequence_iri").map_err(db_err)?;
98            let cluster = ClusterId(r.try_get::<i64, _>("cluster_id").map_err(db_err)?);
99            out.push((iri, cluster));
100        }
101        Ok(out)
102    }
103}
sbol_db_postgres/repo/cluster.rs

sbol_db_postgres/repo/
cluster.rs
