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            View::AsOfInstant(instant) => self.snapshot.as_of_instant(instant),
97            View::SinceInstant(instant) => self.snapshot.since_instant(instant),
98        }
99    }
100}
101
102#[async_trait]
103impl DbBackend for LocalDbBackend {
104    fn db_name(&self) -> &str {
105        &self.db_name
106    }
107
108    async fn query(
109        &self,
110        view: View,
111        query: Edn,
112        args: Vec<Edn>,
113        fuel: Option<u64>,
114    ) -> Result<QueryResult, ClientError> {
115        let db = self.resolve(view);
116        let parsed = ast::parse_query(&query)?;
117        // Bind the receiver as the single default `$`, and each remaining
118        // non-database `:in` spec to the next positional argument.
119        let mut inputs: Vec<QInput<'_>> = Vec::with_capacity(parsed.inputs.len());
120        let mut bound_db = false;
121        let mut next_arg = args.iter();
122        for spec in &parsed.inputs {
123            match spec {
124                ast::InSpec::Db(_) if !bound_db => {
125                    inputs.push(QInput::Db(&db));
126                    bound_db = true;
127                }
128                ast::InSpec::Db(_) => {
129                    return Err(ClientError::Protocol(
130                        "the local fluent API binds a single database source".into(),
131                    ));
132                }
133                _ => {
134                    let arg = next_arg.next().ok_or_else(|| {
135                        ClientError::Protocol("query needs more arguments".into())
136                    })?;
137                    inputs.push(QInput::Edn(arg.clone()));
138                }
139            }
140        }
141        if parsed.inputs.is_empty() {
142            // Default `:in [$]`.
143            inputs.push(QInput::Db(&db));
144        }
145        let (value, _report) = corium_query::run(
146            &parsed,
147            &inputs,
148            ExecOptions {
149                fuel,
150                ..ExecOptions::default()
151            },
152        )?;
153        Ok(QueryResult::new(shape_of(&parsed.find), value))
154    }
155
156    async fn pull(&self, view: View, pattern: Edn, eid: Edn) -> Result<Edn, ClientError> {
157        let db = self.resolve(view);
158        let entity = resolve_eid(&db, &eid)?;
159        Ok(corium_query::pull(&db, &pattern, entity)?)
160    }
161
162    async fn datoms(
163        &self,
164        view: View,
165        index: Index,
166        components: Vec<Edn>,
167        limit: usize,
168    ) -> Result<Vec<DatomRow>, ClientError> {
169        let db = self.resolve(view);
170        let order = index_order(index);
171        let (e, a, v) = resolve_components(&db, order, &components)?;
172        let prefix = key_prefix(order, e, a, v.as_ref());
173        let limit = if limit == 0 { usize::MAX } else { limit };
174        Ok(db
175            .datoms_prefix(order, &prefix)
176            .take(limit)
177            .map(|datom| DatomRow {
178                e: datom.e.raw(),
179                a: datom.a.raw(),
180                v: boundary::value_to_edn(&db, &datom.v),
181                tx: datom.tx.raw(),
182                added: datom.added,
183            })
184            .collect())
185    }
186
187    async fn stats(&self, view: View) -> Result<DbStats, ClientError> {
188        let db = self.resolve(view);
189        let stats = db.stats();
190        Ok(DbStats {
191            basis_t: db.basis_t(),
192            datoms: stats.datoms as u64,
193            entities: stats.entities as u64,
194            attributes: stats.attributes as u64,
195        })
196    }
197}
198
199fn shape_of(find: &ast::FindSpec) -> ResultShape {
200    match find {
201        ast::FindSpec::Rel(_) => ResultShape::Relation,
202        ast::FindSpec::Coll(_) => ResultShape::Collection,
203        ast::FindSpec::Tuple(_) => ResultShape::Tuple,
204        ast::FindSpec::Scalar(_) => ResultShape::Scalar,
205    }
206}
207
208fn index_order(index: Index) -> IndexOrder {
209    match index {
210        Index::Eavt => IndexOrder::Eavt,
211        Index::Aevt => IndexOrder::Aevt,
212        Index::Avet => IndexOrder::Avet,
213        Index::Vaet => IndexOrder::Vaet,
214    }
215}
216
217/// Resolves an entity-position boundary form to an entity id, accepting raw
218/// longs, `#eid` tags, idents, and lookup refs (matching the peer server).
219fn resolve_eid(db: &DbValue, form: &Edn) -> Result<EntityId, ClientError> {
220    match form {
221        Edn::Long(n) => u64::try_from(*n)
222            .map(EntityId::from_raw)
223            .map_err(|_| ClientError::Decode("negative entity id".into())),
224        Edn::Tagged(tag, value) if tag == "eid" => match value.as_ref() {
225            Edn::Long(n) => u64::try_from(*n)
226                .map(EntityId::from_raw)
227                .map_err(|_| ClientError::Decode("negative entity id".into())),
228            _ => Err(ClientError::Decode("#eid requires a long".into())),
229        },
230        Edn::Keyword(keyword) => db
231            .idents()
232            .entid(keyword)
233            .ok_or_else(|| ClientError::Decode(format!("unknown ident {keyword}"))),
234        Edn::Vector(items) => {
235            let [attr_form, value_form] = items.as_slice() else {
236                return Err(ClientError::Decode(
237                    "lookup ref requires [attr value]".into(),
238                ));
239            };
240            let attr = attr_form
241                .as_keyword()
242                .and_then(|keyword| db.idents().entid(keyword))
243                .ok_or_else(|| ClientError::Decode("unknown lookup attribute".into()))?;
244            let value = boundary::edn_to_value(Some(db), value_form)
245                .ok_or_else(|| ClientError::Decode("bad lookup value".into()))?;
246            let value = db.schema().get(attr).map_or(value.clone(), |meta| {
247                exec::coerce_for_type(value, meta.value_type)
248            });
249            db.lookup(attr, &value)
250                .ok_or_else(|| ClientError::Decode(format!("lookup ref {form} did not resolve")))
251        }
252        other => Err(ClientError::Decode(format!("bad entity position {other}"))),
253    }
254}
255
256enum Slot {
257    E,
258    A,
259    V,
260}
261
262/// The entity/attribute/value prefix positions of an index scan.
263type PrefixParts = (
264    Option<EntityId>,
265    Option<EntityId>,
266    Option<corium_core::Value>,
267);
268
269/// Resolves the leading index components into the entity/attribute/value
270/// prefix positions for `order`.
271fn resolve_components(
272    db: &DbValue,
273    order: IndexOrder,
274    components: &[Edn],
275) -> Result<PrefixParts, ClientError> {
276    if components.len() > 3 {
277        return Err(ClientError::Decode("at most three components".into()));
278    }
279    let positions = match order {
280        IndexOrder::Eavt => [Slot::E, Slot::A, Slot::V],
281        IndexOrder::Aevt => [Slot::A, Slot::E, Slot::V],
282        IndexOrder::Avet => [Slot::A, Slot::V, Slot::E],
283        IndexOrder::Vaet => [Slot::V, Slot::A, Slot::E],
284    };
285    let mut e = None;
286    let mut a = None;
287    let mut v = None;
288    for (slot, form) in positions.iter().zip(components) {
289        match slot {
290            Slot::E => e = Some(resolve_eid(db, form)?),
291            Slot::A => {
292                a = Some(match form {
293                    Edn::Keyword(keyword) => db.idents().entid(keyword).ok_or_else(|| {
294                        ClientError::Decode(format!("unknown attribute {keyword}"))
295                    })?,
296                    other => resolve_eid(db, other)?,
297                });
298            }
299            Slot::V => {
300                let value = boundary::edn_to_value(Some(db), form)
301                    .ok_or_else(|| ClientError::Decode(format!("bad value {form}")))?;
302                let value = a
303                    .and_then(|a| db.schema().get(a))
304                    .map_or(value.clone(), |meta| {
305                        exec::coerce_for_type(value, meta.value_type)
306                    });
307                v = Some(value);
308            }
309        }
310    }
311    Ok((e, a, v))
312}
corium_client/local.rs

corium_client/
local.rs
