corium_client/
remote.rs

1//! The remote peer: the fluent API over a peer server, reached over gRPC.
2//!
3//! Presents the same surface as [`crate::LocalPeer`], but every read and
4//! write is an RPC to a hosted peer. Views are named in the request and
5//! resolved server-side; results stream back in chunks and are reassembled
6//! here.
7
8use std::collections::BTreeMap;
9use std::sync::Arc;
10use std::time::Duration;
11
12use async_trait::async_trait;
13use corium_core::{EntityId, KeywordInterner, TotalF64, Value};
14use corium_peer::server::assemble_query_result;
15use corium_protocol::auth::TokenInterceptor;
16use corium_protocol::codec;
17use corium_protocol::pb;
18use corium_protocol::pb::peer_server_client::PeerServerClient;
19use corium_query::edn::Edn;
20use tonic::service::interceptor::InterceptedService;
21use tonic::transport::{Channel, ClientTlsConfig, Endpoint};
22
23use crate::result::{QueryResult, ResultShape};
24use crate::{ClientError, DatomRow, Db, DbBackend, DbStats, Index, Peer, TxData, TxReport, View};
25
26type ServerClient = PeerServerClient<InterceptedService<Channel, TokenInterceptor>>;
27
28/// A fluent client backed by a remote peer server over gRPC.
29pub struct RemotePeer {
30    backend: Arc<RemoteDbBackend>,
31}
32
33impl RemotePeer {
34    /// Connects to a peer server hosting `db`.
35    ///
36    /// # Errors
37    /// Returns [`ClientError`] when the endpoint is unreachable.
38    pub async fn connect(
39        endpoint: impl Into<String>,
40        db: impl Into<String>,
41        token: Option<String>,
42        tls: Option<ClientTlsConfig>,
43    ) -> Result<Self, ClientError> {
44        let endpoint = endpoint.into();
45        let mut builder = Endpoint::from_shared(endpoint)
46            .map_err(|error| ClientError::Protocol(format!("bad endpoint: {error}")))?
47            .connect_timeout(Duration::from_secs(10));
48        if let Some(tls) = tls {
49            builder = builder.tls_config(tls)?;
50        }
51        let channel = builder.connect().await?;
52        let client = PeerServerClient::with_interceptor(channel, TokenInterceptor::new(token));
53        Ok(Self {
54            backend: Arc::new(RemoteDbBackend {
55                client,
56                db_name: db.into(),
57            }),
58        })
59    }
60
61    fn db_at(&self, view: View) -> Db {
62        Db::new(self.backend.clone(), view)
63    }
64}
65
66#[async_trait]
67impl Peer for RemotePeer {
68    fn db_name(&self) -> &str {
69        &self.backend.db_name
70    }
71
72    async fn db(&self) -> Result<Db, ClientError> {
73        Ok(self.db_at(View::Current))
74    }
75
76    async fn transact(&self, tx: TxData) -> Result<TxReport, ClientError> {
77        let tx_data = codec::encode_edn(&Edn::Vector(tx.into_forms()));
78        let mut client = self.backend.client.clone();
79        let response = client
80            .transact(pb::TransactRequest {
81                db: self.backend.db_name.clone(),
82                protocol_version: corium_protocol::PROTOCOL_VERSION,
83                tx_data,
84                expected_basis_t: None,
85            })
86            .await?
87            .into_inner();
88        Ok(TxReport {
89            basis_before: response.basis_before,
90            basis_t: response.basis_t,
91            tx_instant: response.tx_instant,
92            tempids: decode_tempids(&response.tempids)?,
93            // The server syncs its hosted peer to this basis before replying,
94            // so an as-of view at `basis_t` is a stable post-commit snapshot.
95            db_after: self.db_at(View::AsOf(response.basis_t)),
96        })
97    }
98
99    async fn sync(&self) -> Result<Db, ClientError> {
100        // A peer server keeps its hosted peer synced; the current view already
101        // reflects the latest applied basis.
102        Ok(self.db_at(View::Current))
103    }
104}
105
106/// A database backend that issues peer-server RPCs.
107struct RemoteDbBackend {
108    client: ServerClient,
109    db_name: String,
110}
111
112impl RemoteDbBackend {
113    fn view_spec(&self, view: View) -> pb::DbViewSpec {
114        let view = match view {
115            View::Current => None,
116            View::AsOf(t) => Some(pb::db_view_spec::View::AsOf(t)),
117            View::Since(t) => Some(pb::db_view_spec::View::Since(t)),
118            View::History => Some(pb::db_view_spec::View::History(true)),
119        };
120        pb::DbViewSpec {
121            db: self.db_name.clone(),
122            view,
123        }
124    }
125}
126
127#[async_trait]
128impl DbBackend for RemoteDbBackend {
129    fn db_name(&self) -> &str {
130        &self.db_name
131    }
132
133    async fn query(
134        &self,
135        view: View,
136        query: Edn,
137        args: Vec<Edn>,
138        fuel: Option<u64>,
139    ) -> Result<QueryResult, ClientError> {
140        let mut client = self.client.clone();
141        let mut stream = client
142            .query(pb::QueryRequest {
143                dbs: vec![self.view_spec(view)],
144                query: codec::encode_edn(&query),
145                args: codec::encode_edn(&Edn::Vector(args)),
146                fuel: fuel.unwrap_or(0),
147            })
148            .await?
149            .into_inner();
150        let mut chunks = Vec::new();
151        while let Some(chunk) = stream.message().await? {
152            chunks.push(chunk);
153        }
154        let shape = chunks
155            .first()
156            .map_or(ResultShape::Relation, |chunk| shape_of(chunk.shape()));
157        let value = assemble_query_result(&chunks).map_err(ClientError::Decode)?;
158        Ok(QueryResult::new(shape, value))
159    }
160
161    async fn pull(&self, view: View, pattern: Edn, eid: Edn) -> Result<Edn, ClientError> {
162        let mut client = self.client.clone();
163        let response = client
164            .pull(pb::PullRequest {
165                db: Some(self.view_spec(view)),
166                pattern: codec::encode_edn(&pattern),
167                eid: codec::encode_edn(&eid),
168            })
169            .await?
170            .into_inner();
171        Ok(codec::decode_edn(&response.result)?)
172    }
173
174    async fn datoms(
175        &self,
176        view: View,
177        index: Index,
178        components: Vec<Edn>,
179        limit: usize,
180    ) -> Result<Vec<DatomRow>, ClientError> {
181        let mut client = self.client.clone();
182        let mut stream = client
183            .datoms(pb::DatomsRequest {
184                db: Some(self.view_spec(view)),
185                index: index.as_str().to_owned(),
186                components: codec::encode_edn(&Edn::Vector(components)),
187                limit: u64::try_from(limit).unwrap_or(0),
188            })
189            .await?
190            .into_inner();
191        let mut interner = KeywordInterner::default();
192        let mut rows = Vec::new();
193        while let Some(chunk) = stream.message().await? {
194            for datom in codec::decode_datoms(&chunk.datoms, &mut interner)? {
195                rows.push(DatomRow {
196                    e: datom.e.raw(),
197                    a: datom.a.raw(),
198                    v: value_to_edn(&interner, &datom.v),
199                    tx: datom.tx.raw(),
200                    added: datom.added,
201                });
202            }
203        }
204        Ok(rows)
205    }
206
207    async fn stats(&self, view: View) -> Result<DbStats, ClientError> {
208        let mut client = self.client.clone();
209        let response = client
210            .db_stats(pb::DbStatsRequest {
211                db: Some(self.view_spec(view)),
212            })
213            .await?
214            .into_inner();
215        Ok(DbStats {
216            basis_t: response.basis_t,
217            datoms: response.datom_count,
218            entities: response.entity_count,
219            attributes: response.attribute_count,
220        })
221    }
222}
223
224fn shape_of(shape: pb::ResultShape) -> ResultShape {
225    match shape {
226        pb::ResultShape::Collection => ResultShape::Collection,
227        pb::ResultShape::Tuple => ResultShape::Tuple,
228        pb::ResultShape::Scalar => ResultShape::Scalar,
229        pb::ResultShape::Relation | pb::ResultShape::Unspecified => ResultShape::Relation,
230    }
231}
232
233/// Renders a decoded value to boundary EDN using a client-side interner for
234/// keyword names. Refs surface as longs, matching the query boundary.
235fn value_to_edn(interner: &KeywordInterner, value: &Value) -> Edn {
236    match value {
237        Value::Bool(v) => Edn::Bool(*v),
238        Value::Long(v) => Edn::Long(*v),
239        Value::Double(TotalF64(v)) => Edn::Double(TotalF64(*v)),
240        Value::Str(v) => Edn::Str(v.to_string()),
241        Value::Instant(ms) => Edn::Tagged("inst".into(), Box::new(Edn::Long(*ms))),
242        Value::Uuid(v) => Edn::Tagged("uuid".into(), Box::new(Edn::Str(format!("{v:032x}")))),
243        Value::Bytes(bytes) => Edn::Tagged(
244            "bytes".into(),
245            Box::new(Edn::Str(bytes.iter().fold(String::new(), |mut acc, b| {
246                use std::fmt::Write as _;
247                let _ = write!(acc, "{b:02x}");
248                acc
249            }))),
250        ),
251        Value::Keyword(id) => interner
252            .resolve(*id)
253            .map_or(Edn::Nil, |keyword| Edn::Keyword(keyword.clone())),
254        Value::Ref(e) => Edn::Long(i64::try_from(e.raw()).unwrap_or(i64::MAX)),
255    }
256}
257
258/// Decodes a tempid map (string name -> allocated entity id long).
259fn decode_tempids(bytes: &[u8]) -> Result<BTreeMap<String, EntityId>, ClientError> {
260    let Edn::Map(pairs) = codec::decode_edn(bytes)? else {
261        return Err(ClientError::Protocol("tempids must be a map".into()));
262    };
263    let mut tempids = BTreeMap::new();
264    for (key, value) in pairs {
265        let (Edn::Str(name), Edn::Long(raw)) = (key, value) else {
266            return Err(ClientError::Protocol("bad tempid entry".into()));
267        };
268        let raw = u64::try_from(raw).map_err(|_| ClientError::Protocol("bad entity id".into()))?;
269        tempids.insert(name, EntityId::from_raw(raw));
270    }
271    Ok(tempids)
272}
corium_client/remote.rs

corium_client/
remote.rs
