coven_database/store/store_session/
blob_bindings.rs

1use super::*;
2use crate::{ExternalBlob, ExternalBlobRecords};
3
4impl StoreSession<'_> {
5    fn eager_row_blob_refs(&self) -> Result<Vec<coven_protocol::blob::RowBlobRef>, DbError> {
6        let mut references = Vec::new();
7        for table in self.synced_tables {
8            let Some(declaration) = table.blob() else {
9                continue;
10            };
11            if declaration.fill != coven_protocol::blob::CacheFill::CacheEager {
12                continue;
13            }
14            let sql = format!(
15                "SELECT id FROM {} WHERE {} IS NOT NULL ORDER BY id",
16                crate::quote_ident(table.name()),
17                crate::quote_ident(&declaration.id_column),
18            );
19            let mut statement = self.conn.prepare(&sql).map_err(DbError::from)?;
20            let row_ids = statement
21                .query_map([], |row| row.get::<_, String>(0))
22                .map_err(DbError::from)?
23                .collect::<Result<Vec<_>, _>>()
24                .map_err(DbError::from)?;
25            drop(statement);
26            for row_id in row_ids {
27                references.push(Database::row_blob_ref_on(
28                    self.conn, self.gates, table, &row_id,
29                )?);
30            }
31        }
32        Ok(references)
33    }
34
35    fn stored_blob_reference_state(
36        &self,
37        stored: &coven_protocol::blob::locator::StoredBlobRef,
38    ) -> Result<crate::StoredBlobReferenceState, DbError> {
39        Database::stored_blob_reference_state_on(self.conn, self.gates, self.synced_tables, stored)
40    }
41
42    fn row_blob_ref(
43        &self,
44        table_name: &str,
45        row_id: &str,
46    ) -> Result<coven_protocol::blob::RowBlobRef, DbError> {
47        let table = self
48            .synced_tables
49            .iter()
50            .find(|candidate| candidate.name() == table_name)
51            .ok_or_else(|| DbError::Message(format!("undeclared synced table {table_name:?}")))?;
52        if table.blob().is_none() {
53            return Err(DbError::Message(format!(
54                "synced table {:?} has no blob declaration",
55                table.name()
56            )));
57        }
58        Database::row_blob_ref_on(self.conn, self.gates, table, row_id)
59    }
60
61    fn live_row_blob_refs(
62        &self,
63        table_name: &str,
64        row_ids: &[String],
65    ) -> Result<Vec<Option<coven_protocol::blob::RowBlobRef>>, DbError> {
66        let table = self
67            .synced_tables
68            .iter()
69            .find(|candidate| candidate.name() == table_name)
70            .ok_or_else(|| DbError::Message(format!("undeclared synced table {table_name:?}")))?;
71        if table.blob().is_none() {
72            return Err(DbError::Message(format!(
73                "synced table {:?} has no blob declaration",
74                table.name()
75            )));
76        }
77        row_ids
78            .iter()
79            .map(|row_id| Database::live_row_blob_ref_on(self.conn, self.gates, table, row_id))
80            .collect()
81    }
82
83    fn row_blob_refs_for_root(
84        &self,
85        root_table: &str,
86        root_id: &str,
87    ) -> Result<Vec<coven_protocol::blob::RowBlobRef>, DbError> {
88        Database::row_blob_refs_for_root_on(
89            self.conn,
90            self.gates,
91            self.synced_tables,
92            root_table,
93            root_id,
94        )
95    }
96
97    fn external_blob_for_row(
98        &self,
99        reference: &coven_protocol::blob::RowBlobRef,
100    ) -> Result<Option<ExternalBlob>, DbError> {
101        ExternalBlobRecords::new(self.conn).load(reference)
102    }
103}
104
105impl StoreDatabase {
106    pub async fn eager_row_blob_refs(
107        &self,
108    ) -> Result<Vec<coven_protocol::blob::RowBlobRef>, DbError> {
109        self.call_store(|session| session.eager_row_blob_refs())
110            .await
111    }
112
113    pub async fn stored_blob_reference_state(
114        &self,
115        stored: coven_protocol::blob::locator::StoredBlobRef,
116    ) -> Result<crate::StoredBlobReferenceState, DbError> {
117        self.call_store(move |session| session.stored_blob_reference_state(&stored))
118            .await
119    }
120
121    #[doc(hidden)]
122    pub async fn row_blob_ref(
123        &self,
124        table: &str,
125        row_id: &str,
126    ) -> Result<coven_protocol::blob::RowBlobRef, DbError> {
127        let table = table.to_string();
128        let row_id = row_id.to_string();
129        self.call_store(move |session| session.row_blob_ref(&table, &row_id))
130            .await
131    }
132
133    /// The exact current blob-bearing row version for each of `row_ids`, in the
134    /// order given, resolved in one read on the connection. `None` where an id
135    /// names no live blob-bearing row.
136    ///
137    /// The list form of [`row_blob_ref`](Self::row_blob_ref): a host about to
138    /// show a page of rows resolves the page in one call instead of one per row.
139    pub async fn live_row_blob_refs(
140        &self,
141        table: &str,
142        row_ids: Vec<String>,
143    ) -> Result<Vec<Option<coven_protocol::blob::RowBlobRef>>, DbError> {
144        let table = table.to_string();
145        self.call_store(move |session| session.live_row_blob_refs(&table, &row_ids))
146            .await
147    }
148
149    pub async fn row_blob_refs_for_root(
150        &self,
151        root_table: &str,
152        root_id: &str,
153    ) -> Result<Vec<coven_protocol::blob::RowBlobRef>, DbError> {
154        let root_table = root_table.to_string();
155        let root_id = root_id.to_string();
156        self.call_store(move |session| session.row_blob_refs_for_root(&root_table, &root_id))
157            .await
158    }
159
160    pub async fn validate_row_blob_ref(
161        &self,
162        reference: &coven_protocol::blob::RowBlobRef,
163    ) -> Result<(), DbError> {
164        let reference = reference.clone();
165        self.call_store(move |session| {
166            let table = session
167                .synced_tables
168                .iter()
169                .find(|table| table.name() == reference.table())
170                .ok_or_else(|| {
171                    DbError::Message(format!("undeclared synced table {:?}", reference.table()))
172                })?;
173            Database::validate_row_blob_ref_on(session.conn, session.gates, table, &reference)
174        })
175        .await
176    }
177
178    pub async fn external_blob_for_row(
179        &self,
180        reference: &coven_protocol::blob::RowBlobRef,
181    ) -> Result<Option<ExternalBlob>, DbError> {
182        let reference = reference.clone();
183        self.call_store(move |session| session.external_blob_for_row(&reference))
184            .await
185    }
186
187    #[doc(hidden)]
188    pub async fn external_blob(
189        &self,
190        table: &str,
191        row_id: &str,
192    ) -> Result<Option<ExternalBlob>, DbError> {
193        let reference = self.row_blob_ref(table, row_id).await?;
194        self.external_blob_for_row(&reference).await
195    }
196}
coven_database/store/store_session/blob_bindings.rs

coven_database/store/store_session/
blob_bindings.rs
