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            })
85            .await?
86            .into_inner();
87        Ok(TxReport {
88            basis_before: response.basis_before,
89            basis_t: response.basis_t,
90            tx_instant: response.tx_instant,
91            tempids: decode_tempids(&response.tempids)?,
92            // The server syncs its hosted peer to this basis before replying,
93            // so an as-of view at `basis_t` is a stable post-commit snapshot.
94            db_after: self.db_at(View::AsOf(response.basis_t)),
95        })
96    }
97
98    async fn sync(&self) -> Result<Db, ClientError> {
99        // A peer server keeps its hosted peer synced; the current view already
100        // reflects the latest applied basis.
101        Ok(self.db_at(View::Current))
102    }
103}
104
105/// A database backend that issues peer-server RPCs.
106struct RemoteDbBackend {
107    client: ServerClient,
108    db_name: String,
109}
110
111impl RemoteDbBackend {
112    fn view_spec(&self, view: View) -> pb::DbViewSpec {
113        let view = match view {
114            View::Current => None,
115            View::AsOf(t) => Some(pb::db_view_spec::View::AsOf(t)),
116            View::Since(t) => Some(pb::db_view_spec::View::Since(t)),
117            View::History => Some(pb::db_view_spec::View::History(true)),
118        };
119        pb::DbViewSpec {
120            db: self.db_name.clone(),
121            view,
122        }
123    }
124}
125
126#[async_trait]
127impl DbBackend for RemoteDbBackend {
128    fn db_name(&self) -> &str {
129        &self.db_name
130    }
131
132    async fn query(
133        &self,
134        view: View,
135        query: Edn,
136        args: Vec<Edn>,
137        fuel: Option<u64>,
138    ) -> Result<QueryResult, ClientError> {
139        let mut client = self.client.clone();
140        let mut stream = client
141            .query(pb::QueryRequest {
142                dbs: vec![self.view_spec(view)],
143                query: codec::encode_edn(&query),
144                args: codec::encode_edn(&Edn::Vector(args)),
145                fuel: fuel.unwrap_or(0),
146            })
147            .await?
148            .into_inner();
149        let mut chunks = Vec::new();
150        while let Some(chunk) = stream.message().await? {
151            chunks.push(chunk);
152        }
153        let shape = chunks
154            .first()
155            .map_or(ResultShape::Relation, |chunk| shape_of(chunk.shape()));
156        let value = assemble_query_result(&chunks).map_err(ClientError::Decode)?;
157        Ok(QueryResult::new(shape, value))
158    }
159
160    async fn pull(&self, view: View, pattern: Edn, eid: Edn) -> Result<Edn, ClientError> {
161        let mut client = self.client.clone();
162        let response = client
163            .pull(pb::PullRequest {
164                db: Some(self.view_spec(view)),
165                pattern: codec::encode_edn(&pattern),
166                eid: codec::encode_edn(&eid),
167            })
168            .await?
169            .into_inner();
170        Ok(codec::decode_edn(&response.result)?)
171    }
172
173    async fn datoms(
174        &self,
175        view: View,
176        index: Index,
177        components: Vec<Edn>,
178        limit: usize,
179    ) -> Result<Vec<DatomRow>, ClientError> {
180        let mut client = self.client.clone();
181        let mut stream = client
182            .datoms(pb::DatomsRequest {
183                db: Some(self.view_spec(view)),
184                index: index.as_str().to_owned(),
185                components: codec::encode_edn(&Edn::Vector(components)),
186                limit: u64::try_from(limit).unwrap_or(0),
187            })
188            .await?
189            .into_inner();
190        let mut interner = KeywordInterner::default();
191        let mut rows = Vec::new();
192        while let Some(chunk) = stream.message().await? {
193            for datom in codec::decode_datoms(&chunk.datoms, &mut interner)? {
194                rows.push(DatomRow {
195                    e: datom.e.raw(),
196                    a: datom.a.raw(),
197                    v: value_to_edn(&interner, &datom.v),
198                    tx: datom.tx.raw(),
199                    added: datom.added,
200                });
201            }
202        }
203        Ok(rows)
204    }
205
206    async fn stats(&self, view: View) -> Result<DbStats, ClientError> {
207        let mut client = self.client.clone();
208        let response = client
209            .db_stats(pb::DbStatsRequest {
210                db: Some(self.view_spec(view)),
211            })
212            .await?
213            .into_inner();
214        Ok(DbStats {
215            basis_t: response.basis_t,
216            datoms: response.datom_count,
217            entities: response.entity_count,
218            attributes: response.attribute_count,
219        })
220    }
221}
222
223fn shape_of(shape: pb::ResultShape) -> ResultShape {
224    match shape {
225        pb::ResultShape::Collection => ResultShape::Collection,
226        pb::ResultShape::Tuple => ResultShape::Tuple,
227        pb::ResultShape::Scalar => ResultShape::Scalar,
228        pb::ResultShape::Relation | pb::ResultShape::Unspecified => ResultShape::Relation,
229    }
230}
231
232/// Renders a decoded value to boundary EDN using a client-side interner for
233/// keyword names. Refs surface as longs, matching the query boundary.
234fn value_to_edn(interner: &KeywordInterner, value: &Value) -> Edn {
235    match value {
236        Value::Bool(v) => Edn::Bool(*v),
237        Value::Long(v) => Edn::Long(*v),
238        Value::Double(TotalF64(v)) => Edn::Double(TotalF64(*v)),
239        Value::Str(v) => Edn::Str(v.to_string()),
240        Value::Instant(ms) => Edn::Tagged("inst".into(), Box::new(Edn::Long(*ms))),
241        Value::Uuid(v) => Edn::Tagged("uuid".into(), Box::new(Edn::Str(format!("{v:032x}")))),
242        Value::Bytes(bytes) => Edn::Tagged(
243            "bytes".into(),
244            Box::new(Edn::Str(bytes.iter().fold(String::new(), |mut acc, b| {
245                use std::fmt::Write as _;
246                let _ = write!(acc, "{b:02x}");
247                acc
248            }))),
249        ),
250        Value::Keyword(id) => interner
251            .resolve(*id)
252            .map_or(Edn::Nil, |keyword| Edn::Keyword(keyword.clone())),
253        Value::Ref(e) => Edn::Long(i64::try_from(e.raw()).unwrap_or(i64::MAX)),
254    }
255}
256
257/// Decodes a tempid map (string name -> allocated entity id long).
258fn decode_tempids(bytes: &[u8]) -> Result<BTreeMap<String, EntityId>, ClientError> {
259    let Edn::Map(pairs) = codec::decode_edn(bytes)? else {
260        return Err(ClientError::Protocol("tempids must be a map".into()));
261    };
262    let mut tempids = BTreeMap::new();
263    for (key, value) in pairs {
264        let (Edn::Str(name), Edn::Long(raw)) = (key, value) else {
265            return Err(ClientError::Protocol("bad tempid entry".into()));
266        };
267        let raw = u64::try_from(raw).map_err(|_| ClientError::Protocol("bad entity id".into()))?;
268        tempids.insert(name, EntityId::from_raw(raw));
269    }
270    Ok(tempids)
271}
corium_client/remote.rs

corium_client/
remote.rs
