corium_client/
local.rs

1//! The local peer: the fluent API over the in-process [`corium_peer`]
2//! library. Queries execute against immutable database values read directly
3//! from storage, with no round trip to the transactor.
4
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use corium_core::{EntityId, IndexOrder};
9use corium_db::{Db as DbValue, key_prefix};
10use corium_peer::{ConnectConfig, Connection};
11use corium_query::edn::Edn;
12use corium_query::{ExecOptions, QInput, ast, boundary, exec};
13
14use crate::result::{QueryResult, ResultShape};
15use crate::{ClientError, DatomRow, Db, DbBackend, DbStats, Index, Peer, TxData, TxReport, View};
16
17/// A fluent client backed by the in-process peer library.
18pub struct LocalPeer {
19    connection: Arc<Connection>,
20}
21
22impl LocalPeer {
23    /// Connects a peer to a transactor and wraps it in the fluent API.
24    ///
25    /// # Errors
26    /// Returns [`ClientError`] when no endpoint accepts the subscription.
27    pub async fn connect(config: ConnectConfig) -> Result<Self, ClientError> {
28        Ok(Self {
29            connection: Arc::new(Connection::connect(config).await?),
30        })
31    }
32
33    /// Wraps an already-established peer connection.
34    #[must_use]
35    pub fn from_connection(connection: Arc<Connection>) -> Self {
36        Self { connection }
37    }
38
39    /// The underlying peer connection, for peer-library operations the fluent
40    /// API does not surface (tx-report streaming, index policy, and so on).
41    #[must_use]
42    pub fn connection(&self) -> &Arc<Connection> {
43        &self.connection
44    }
45
46    fn snapshot_db(&self, snapshot: DbValue) -> Db {
47        Db::new(
48            Arc::new(LocalDbBackend {
49                snapshot,
50                db_name: self.connection.db_name().to_owned(),
51            }),
52            View::Current,
53        )
54    }
55}
56
57#[async_trait]
58impl Peer for LocalPeer {
59    fn db_name(&self) -> &str {
60        self.connection.db_name()
61    }
62
63    async fn db(&self) -> Result<Db, ClientError> {
64        Ok(self.snapshot_db(self.connection.db()))
65    }
66
67    async fn transact(&self, tx: TxData) -> Result<TxReport, ClientError> {
68        let result = self.connection.transact(tx.into_forms()).await?;
69        Ok(TxReport {
70            basis_before: result.basis_before,
71            basis_t: result.basis_t,
72            tx_instant: result.tx_instant,
73            tempids: result.tempids,
74            db_after: self.snapshot_db(result.db_after),
75        })
76    }
77
78    async fn sync(&self) -> Result<Db, ClientError> {
79        Ok(self.snapshot_db(self.connection.sync().await?))
80    }
81}
82
83/// A database backend over a captured immutable snapshot.
84struct LocalDbBackend {
85    snapshot: DbValue,
86    db_name: String,
87}
88
89impl LocalDbBackend {
90    fn resolve(&self, view: View) -> DbValue {
91        match view {
92            View::Current => self.snapshot.clone(),
93            View::AsOf(t) => self.snapshot.as_of(t),
94            View::Since(t) => self.snapshot.since(t),
95            View::History => self.snapshot.history(),
96        }
97    }
98}
99
100#[async_trait]
101impl DbBackend for LocalDbBackend {
102    fn db_name(&self) -> &str {
103        &self.db_name
104    }
105
106    async fn query(
107        &self,
108        view: View,
109        query: Edn,
110        args: Vec<Edn>,
111        fuel: Option<u64>,
112    ) -> Result<QueryResult, ClientError> {
113        let db = self.resolve(view);
114        let parsed = ast::parse_query(&query)?;
115        // Bind the receiver as the single default `$`, and each remaining
116        // non-database `:in` spec to the next positional argument.
117        let mut inputs: Vec<QInput<'_>> = Vec::with_capacity(parsed.inputs.len());
118        let mut bound_db = false;
119        let mut next_arg = args.iter();
120        for spec in &parsed.inputs {
121            match spec {
122                ast::InSpec::Db(_) if !bound_db => {
123                    inputs.push(QInput::Db(&db));
124                    bound_db = true;
125                }
126                ast::InSpec::Db(_) => {
127                    return Err(ClientError::Protocol(
128                        "the local fluent API binds a single database source".into(),
129                    ));
130                }
131                _ => {
132                    let arg = next_arg.next().ok_or_else(|| {
133                        ClientError::Protocol("query needs more arguments".into())
134                    })?;
135                    inputs.push(QInput::Edn(arg.clone()));
136                }
137            }
138        }
139        if parsed.inputs.is_empty() {
140            // Default `:in [$]`.
141            inputs.push(QInput::Db(&db));
142        }
143        let (value, _report) = corium_query::run(
144            &parsed,
145            &inputs,
146            ExecOptions {
147                fuel,
148                ..ExecOptions::default()
149            },
150        )?;
151        Ok(QueryResult::new(shape_of(&parsed.find), value))
152    }
153
154    async fn pull(&self, view: View, pattern: Edn, eid: Edn) -> Result<Edn, ClientError> {
155        let db = self.resolve(view);
156        let entity = resolve_eid(&db, &eid)?;
157        Ok(corium_query::pull(&db, &pattern, entity)?)
158    }
159
160    async fn datoms(
161        &self,
162        view: View,
163        index: Index,
164        components: Vec<Edn>,
165        limit: usize,
166    ) -> Result<Vec<DatomRow>, ClientError> {
167        let db = self.resolve(view);
168        let order = index_order(index);
169        let (e, a, v) = resolve_components(&db, order, &components)?;
170        let prefix = key_prefix(order, e, a, v.as_ref());
171        let limit = if limit == 0 { usize::MAX } else { limit };
172        Ok(db
173            .datoms_prefix(order, &prefix)
174            .take(limit)
175            .map(|datom| DatomRow {
176                e: datom.e.raw(),
177                a: datom.a.raw(),
178                v: boundary::value_to_edn(&db, &datom.v),
179                tx: datom.tx.raw(),
180                added: datom.added,
181            })
182            .collect())
183    }
184
185    async fn stats(&self, view: View) -> Result<DbStats, ClientError> {
186        let db = self.resolve(view);
187        let stats = db.stats();
188        Ok(DbStats {
189            basis_t: db.basis_t(),
190            datoms: stats.datoms as u64,
191            entities: stats.entities as u64,
192            attributes: stats.attributes as u64,
193        })
194    }
195}
196
197fn shape_of(find: &ast::FindSpec) -> ResultShape {
198    match find {
199        ast::FindSpec::Rel(_) => ResultShape::Relation,
200        ast::FindSpec::Coll(_) => ResultShape::Collection,
201        ast::FindSpec::Tuple(_) => ResultShape::Tuple,
202        ast::FindSpec::Scalar(_) => ResultShape::Scalar,
203    }
204}
205
206fn index_order(index: Index) -> IndexOrder {
207    match index {
208        Index::Eavt => IndexOrder::Eavt,
209        Index::Aevt => IndexOrder::Aevt,
210        Index::Avet => IndexOrder::Avet,
211        Index::Vaet => IndexOrder::Vaet,
212    }
213}
214
215/// Resolves an entity-position boundary form to an entity id, accepting raw
216/// longs, `#eid` tags, idents, and lookup refs (matching the peer server).
217fn resolve_eid(db: &DbValue, form: &Edn) -> Result<EntityId, ClientError> {
218    match form {
219        Edn::Long(n) => u64::try_from(*n)
220            .map(EntityId::from_raw)
221            .map_err(|_| ClientError::Decode("negative entity id".into())),
222        Edn::Tagged(tag, value) if tag == "eid" => match value.as_ref() {
223            Edn::Long(n) => u64::try_from(*n)
224                .map(EntityId::from_raw)
225                .map_err(|_| ClientError::Decode("negative entity id".into())),
226            _ => Err(ClientError::Decode("#eid requires a long".into())),
227        },
228        Edn::Keyword(keyword) => db
229            .idents()
230            .entid(keyword)
231            .ok_or_else(|| ClientError::Decode(format!("unknown ident {keyword}"))),
232        Edn::Vector(items) => {
233            let [attr_form, value_form] = items.as_slice() else {
234                return Err(ClientError::Decode(
235                    "lookup ref requires [attr value]".into(),
236                ));
237            };
238            let attr = attr_form
239                .as_keyword()
240                .and_then(|keyword| db.idents().entid(keyword))
241                .ok_or_else(|| ClientError::Decode("unknown lookup attribute".into()))?;
242            let value = boundary::edn_to_value(Some(db), value_form)
243                .ok_or_else(|| ClientError::Decode("bad lookup value".into()))?;
244            let value = db.schema().get(attr).map_or(value.clone(), |meta| {
245                exec::coerce_for_type(value, meta.value_type)
246            });
247            db.lookup(attr, &value)
248                .ok_or_else(|| ClientError::Decode(format!("lookup ref {form} did not resolve")))
249        }
250        other => Err(ClientError::Decode(format!("bad entity position {other}"))),
251    }
252}
253
254enum Slot {
255    E,
256    A,
257    V,
258}
259
260/// The entity/attribute/value prefix positions of an index scan.
261type PrefixParts = (
262    Option<EntityId>,
263    Option<EntityId>,
264    Option<corium_core::Value>,
265);
266
267/// Resolves the leading index components into the entity/attribute/value
268/// prefix positions for `order`.
269fn resolve_components(
270    db: &DbValue,
271    order: IndexOrder,
272    components: &[Edn],
273) -> Result<PrefixParts, ClientError> {
274    if components.len() > 3 {
275        return Err(ClientError::Decode("at most three components".into()));
276    }
277    let positions = match order {
278        IndexOrder::Eavt => [Slot::E, Slot::A, Slot::V],
279        IndexOrder::Aevt => [Slot::A, Slot::E, Slot::V],
280        IndexOrder::Avet => [Slot::A, Slot::V, Slot::E],
281        IndexOrder::Vaet => [Slot::V, Slot::A, Slot::E],
282    };
283    let mut e = None;
284    let mut a = None;
285    let mut v = None;
286    for (slot, form) in positions.iter().zip(components) {
287        match slot {
288            Slot::E => e = Some(resolve_eid(db, form)?),
289            Slot::A => {
290                a = Some(match form {
291                    Edn::Keyword(keyword) => db.idents().entid(keyword).ok_or_else(|| {
292                        ClientError::Decode(format!("unknown attribute {keyword}"))
293                    })?,
294                    other => resolve_eid(db, other)?,
295                });
296            }
297            Slot::V => {
298                let value = boundary::edn_to_value(Some(db), form)
299                    .ok_or_else(|| ClientError::Decode(format!("bad value {form}")))?;
300                let value = a
301                    .and_then(|a| db.schema().get(a))
302                    .map_or(value.clone(), |meta| {
303                        exec::coerce_for_type(value, meta.value_type)
304                    });
305                v = Some(value);
306            }
307        }
308    }
309    Ok((e, a, v))
310}
corium_client/local.rs

corium_client/
local.rs
