sbol_db_postgres/repo/
object.rs

1use sbol_db_core::{DomainError, GraphId, IriString, ObjectId, ObjectSummary, SbolObjectRecord};
2use sqlx::Row;
3use uuid::Uuid;
4
5use crate::repo::db_err;
6use crate::PgPool;
7
8use sbol_db_storage::ListObjectsFilter;
9
10#[derive(Clone)]
11pub struct SbolObjectRepository {
12    pool: PgPool,
13}
14
15pub struct UpsertResult {
16    pub object_id: ObjectId,
17    pub revision_number: i64,
18    pub previous_content_hash: Option<Vec<u8>>,
19}
20
21impl SbolObjectRepository {
22    pub fn new(pool: PgPool) -> Self {
23        Self { pool }
24    }
25
26    /// Insert or update by IRI. Returns the resulting object id, the new
27    /// revision number, and the previous content hash (if any) so callers
28    /// can detect a no-op.
29    pub async fn upsert(
30        &self,
31        conn: &mut sqlx::PgConnection,
32        summary: &ObjectSummary,
33        graph_id: Option<GraphId>,
34    ) -> Result<UpsertResult, DomainError> {
35        let row = sqlx::query(
36            r#"
37            WITH prev AS (
38                SELECT id, content_hash FROM sbol_objects WHERE iri = $1
39            ),
40            upsert AS (
41                INSERT INTO sbol_objects (
42                    iri, sbol_class, display_id, name, description,
43                    graph_id, types, roles, data, content_hash, updated_at
44                ) VALUES ($1, $2, $3, $4, $5, $6, $7::text[]::sbol_ontology_term[],
45                          $8::text[]::sbol_ontology_term[], $9, $10, now())
46                ON CONFLICT (iri) DO UPDATE SET
47                    sbol_class = EXCLUDED.sbol_class,
48                    display_id = EXCLUDED.display_id,
49                    name = EXCLUDED.name,
50                    description = EXCLUDED.description,
51                    graph_id = EXCLUDED.graph_id,
52                    types = EXCLUDED.types,
53                    roles = EXCLUDED.roles,
54                    data = EXCLUDED.data,
55                    content_hash = EXCLUDED.content_hash,
56                    updated_at = now()
57                RETURNING id
58            ),
59            next_revision AS (
60                INSERT INTO sbol_object_revisions (
61                    object_id, iri, revision_number, data, content_hash
62                )
63                SELECT
64                    (SELECT id FROM upsert),
65                    $1,
66                    COALESCE(
67                        (SELECT MAX(revision_number) + 1 FROM sbol_object_revisions
68                            WHERE object_id = (SELECT id FROM upsert)),
69                        1
70                    ),
71                    $9,
72                    $10
73                RETURNING revision_number
74            )
75            SELECT
76                (SELECT id FROM upsert) AS id,
77                (SELECT revision_number FROM next_revision) AS revision_number,
78                (SELECT content_hash FROM prev) AS previous_content_hash
79            "#,
80        )
81        .bind(summary.iri.as_str())
82        .bind(&summary.sbol_class)
83        .bind(summary.display_id.as_deref())
84        .bind(summary.name.as_deref())
85        .bind(summary.description.as_deref())
86        .bind(graph_id.map(|d| d.0))
87        .bind(&summary.types)
88        .bind(&summary.roles)
89        .bind(&summary.data)
90        .bind(&summary.content_hash)
91        .fetch_one(&mut *conn)
92        .await
93        .map_err(db_err)?;
94
95        let id: Uuid = row.try_get("id").map_err(db_err)?;
96        let revision_number: i64 = row.try_get("revision_number").map_err(db_err)?;
97        let previous_content_hash: Option<Vec<u8>> =
98            row.try_get("previous_content_hash").map_err(db_err)?;
99
100        Ok(UpsertResult {
101            object_id: ObjectId(id),
102            revision_number,
103            previous_content_hash,
104        })
105    }
106
107    pub async fn get_iri_by_id(&self, id: ObjectId) -> Result<Option<String>, DomainError> {
108        let row = sqlx::query("SELECT iri FROM sbol_objects WHERE id = $1")
109            .bind(id.0)
110            .fetch_optional(&self.pool)
111            .await
112            .map_err(db_err)?;
113        match row {
114            Some(row) => Ok(Some(row.try_get::<String, _>("iri").map_err(db_err)?)),
115            None => Ok(None),
116        }
117    }
118
119    pub async fn get_by_iri(&self, iri: &str) -> Result<Option<SbolObjectRecord>, DomainError> {
120        let row = sqlx::query(
121            r#"
122            SELECT id, iri::text AS iri, sbol_class, display_id, name, description,
123                   graph_id, types::text[] AS types, roles::text[] AS roles,
124                   data, content_hash
125            FROM sbol_objects
126            WHERE iri = $1 AND is_deleted = false
127            "#,
128        )
129        .bind(iri)
130        .fetch_optional(&self.pool)
131        .await
132        .map_err(db_err)?;
133
134        row.map(row_to_record).transpose()
135    }
136
137    /// Bulk variant of [`get_by_iri`]. Returns matching live rows in arbitrary
138    /// order; IRIs that don't resolve are simply absent from the result. A
139    /// single indexed scan via `iri = ANY($1)`.
140    pub async fn get_by_iris(&self, iris: &[&str]) -> Result<Vec<SbolObjectRecord>, DomainError> {
141        if iris.is_empty() {
142            return Ok(Vec::new());
143        }
144        let owned: Vec<String> = iris.iter().map(|s| (*s).to_owned()).collect();
145        let rows = sqlx::query(
146            r#"
147            SELECT id, iri::text AS iri, sbol_class, display_id, name, description,
148                   graph_id, types::text[] AS types, roles::text[] AS roles,
149                   data, content_hash
150            FROM sbol_objects
151            WHERE iri::text = ANY($1::text[]) AND is_deleted = false
152            "#,
153        )
154        .bind(&owned)
155        .fetch_all(&self.pool)
156        .await
157        .map_err(db_err)?;
158
159        rows.into_iter().map(row_to_record).collect()
160    }
161
162    /// Paginated listing for corpus-scale export. Keyset cursor on `iri`
163    /// (lexicographic ascending); pass the last `iri` of the prior page as
164    /// `after_iri` to fetch the next. Filters compose: when any of
165    /// `sbol_class`, `role`, or `graph_id` is `Some`, the corresponding
166    /// predicate is added to the WHERE clause.
167    pub async fn list(
168        &self,
169        filter: &ListObjectsFilter,
170    ) -> Result<Vec<SbolObjectRecord>, DomainError> {
171        let limit = filter.limit.clamp(1, 5000) as i64;
172        let rows = sqlx::query(
173            r#"
174            SELECT id, iri::text AS iri, sbol_class, display_id, name, description,
175                   graph_id, types::text[] AS types, roles::text[] AS roles,
176                   data, content_hash
177            FROM sbol_objects
178            WHERE is_deleted = false
179              AND ($1::text IS NULL OR sbol_class = $1)
180              AND ($2::text IS NULL OR $2 = ANY(roles::text[]))
181              AND ($3::uuid IS NULL OR graph_id = $3)
182              AND ($4::text IS NULL OR iri::text > $4)
183            ORDER BY iri::text ASC
184            LIMIT $5
185            "#,
186        )
187        .bind(filter.sbol_class.as_deref())
188        .bind(filter.role.as_deref())
189        .bind(filter.graph_id.map(|d| d.0))
190        .bind(filter.after_iri.as_deref())
191        .bind(limit)
192        .fetch_all(&self.pool)
193        .await
194        .map_err(db_err)?;
195
196        rows.into_iter().map(row_to_record).collect()
197    }
198}
199
200fn row_to_record(row: sqlx::postgres::PgRow) -> Result<SbolObjectRecord, DomainError> {
201    let id: Uuid = row.try_get("id").map_err(db_err)?;
202    let iri: String = row.try_get("iri").map_err(db_err)?;
203    let sbol_class: String = row.try_get("sbol_class").map_err(db_err)?;
204    let display_id: Option<String> = row.try_get("display_id").map_err(db_err)?;
205    let name: Option<String> = row.try_get("name").map_err(db_err)?;
206    let description: Option<String> = row.try_get("description").map_err(db_err)?;
207    let graph_id: Option<Uuid> = row.try_get("graph_id").map_err(db_err)?;
208    let types: Vec<String> = row.try_get("types").map_err(db_err)?;
209    let roles: Vec<String> = row.try_get("roles").map_err(db_err)?;
210    let data: serde_json::Value = row.try_get("data").map_err(db_err)?;
211    let content_hash: Option<Vec<u8>> = row.try_get("content_hash").map_err(db_err)?;
212    Ok(SbolObjectRecord {
213        id: ObjectId(id),
214        iri: IriString::unchecked(iri),
215        sbol_class,
216        display_id,
217        name,
218        description,
219        graph_id: graph_id.map(GraphId),
220        types,
221        roles,
222        data,
223        content_hash: content_hash.unwrap_or_default(),
224    })
225}
sbol_db_postgres/repo/object.rs

sbol_db_postgres/repo/
object.rs
