gonzalo_store_server/
lib.rs

1//! A `Store` backed by a remote gonzalo daemon. Operators choose the
2//! transport: [`ServerStore::http`] (HTTP/JSON via reqwest) or
3//! [`ServerStore::grpc`] (gRPC via tonic). Both speak the same JSON
4//! serialization of `gonzalo-core` types. Each transport has a
5//! `*_with_token` constructor that sends `Authorization: Bearer <token>`.
6
7use async_trait::async_trait;
8use gonzalo_core::{
9    BlobStore, ContentHash, CoreError, DeleteResult, KeyPrefix, PutResult, Record, RecordKey,
10    Result, Revision, Store, store::Conflict,
11};
12use gonzalo_proto::http::{DeleteBody, DeleteOutcome, PutBody, PutOutcome};
13use gonzalo_proto::v1::{
14    DeleteBlobRequest, DeleteRequest, GetBlobRequest, GetRequest, ListBlobsRequest, ListRequest,
15    PutBlobRequest, PutRequest, gonzalo_client::GonzaloClient,
16};
17use tonic::transport::Channel;
18
19enum Backend {
20    Http {
21        base: reqwest::Url,
22        client: reqwest::Client,
23        token: Option<String>,
24    },
25    Grpc {
26        client: GonzaloClient<Channel>,
27        token: Option<String>,
28    },
29}
30
31/// A client substrate over a remote gonzalo daemon.
32pub struct ServerStore {
33    backend: Backend,
34}
35
36impl ServerStore {
37    /// Talk to the daemon's HTTP/JSON API rooted at `base_url`.
38    pub fn http(base_url: &str) -> Result<Self> {
39        Self::http_inner(base_url, None)
40    }
41
42    /// As [`ServerStore::http`], sending a bearer token on every request.
43    pub fn http_with_token(base_url: &str, token: impl Into<String>) -> Result<Self> {
44        Self::http_inner(base_url, Some(token.into()))
45    }
46
47    fn http_inner(base_url: &str, token: Option<String>) -> Result<Self> {
48        let base = reqwest::Url::parse(base_url).map_err(|e| CoreError::Backend(e.to_string()))?;
49        Ok(Self {
50            backend: Backend::Http {
51                base,
52                client: reqwest::Client::new(),
53                token,
54            },
55        })
56    }
57
58    /// Talk to the daemon's gRPC API at `endpoint` (e.g. `http://host:50051`).
59    pub async fn grpc(endpoint: String) -> Result<Self> {
60        Self::grpc_inner(endpoint, None).await
61    }
62
63    /// As [`ServerStore::grpc`], sending a bearer token on every call.
64    pub async fn grpc_with_token(endpoint: String, token: impl Into<String>) -> Result<Self> {
65        Self::grpc_inner(endpoint, Some(token.into())).await
66    }
67
68    async fn grpc_inner(endpoint: String, token: Option<String>) -> Result<Self> {
69        let client = GonzaloClient::connect(endpoint)
70            .await
71            .map_err(|e| CoreError::Backend(e.to_string()))?
72            .max_decoding_message_size(gonzalo_proto::DEFAULT_MAX_BLOB_SIZE);
73        Ok(Self {
74            backend: Backend::Grpc { client, token },
75        })
76    }
77
78    fn records_url(base: &reqwest::Url, key: &RecordKey) -> Result<reqwest::Url> {
79        let mut url = base.clone();
80        url.path_segments_mut()
81            .map_err(|_| CoreError::Backend("base URL cannot be a base".into()))?
82            .extend(["v1", "records", &key.namespace, &key.collection, &key.id]);
83        Ok(url)
84    }
85
86    /// `…/v1/blobs` (list) or `…/v1/blobs/{hash}` (one blob) when `hash` is set.
87    fn blobs_url(base: &reqwest::Url, hash: Option<&str>) -> Result<reqwest::Url> {
88        let mut url = base.clone();
89        {
90            let mut seg = url
91                .path_segments_mut()
92                .map_err(|_| CoreError::Backend("base URL cannot be a base".into()))?;
93            seg.extend(["v1", "blobs"]);
94            if let Some(h) = hash {
95                seg.push(h);
96            }
97        }
98        Ok(url)
99    }
100}
101
102/// Wrap a gRPC message in a request, attaching the bearer token if present.
103fn grpc_request<T>(msg: T, token: &Option<String>) -> Result<tonic::Request<T>> {
104    let mut req = tonic::Request::new(msg);
105    if let Some(t) = token {
106        let value = format!("Bearer {t}")
107            .parse()
108            .map_err(|_| CoreError::Backend("invalid token characters".into()))?;
109        req.metadata_mut().insert("authorization", value);
110    }
111    Ok(req)
112}
113
114fn maybe_auth(rb: reqwest::RequestBuilder, token: &Option<String>) -> reqwest::RequestBuilder {
115    match token {
116        Some(t) => rb.bearer_auth(t),
117        None => rb,
118    }
119}
120
121#[async_trait]
122impl Store for ServerStore {
123    async fn get(&self, key: &RecordKey) -> Result<Option<Record>> {
124        match &self.backend {
125            Backend::Http {
126                base,
127                client,
128                token,
129            } => {
130                let url = Self::records_url(base, key)?;
131                let resp = maybe_auth(client.get(url), token)
132                    .send()
133                    .await
134                    .map_err(be)?;
135                if resp.status() == reqwest::StatusCode::NOT_FOUND {
136                    return Ok(None);
137                }
138                let resp = resp.error_for_status().map_err(be)?;
139                Ok(Some(resp.json::<Record>().await.map_err(be)?))
140            }
141            Backend::Grpc { client, token } => {
142                let mut client = client.clone();
143                let req = grpc_request(
144                    GetRequest {
145                        namespace: key.namespace.clone(),
146                        collection: key.collection.clone(),
147                        id: key.id.clone(),
148                    },
149                    token,
150                )?;
151                let resp = client.get(req).await.map_err(status)?.into_inner();
152                if resp.found {
153                    Ok(Some(serde_json::from_slice(&resp.record_json).map_err(se)?))
154                } else {
155                    Ok(None)
156                }
157            }
158        }
159    }
160
161    async fn put(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
162        match &self.backend {
163            Backend::Http {
164                base,
165                client,
166                token,
167            } => {
168                let url = Self::records_url(base, &record.key)?;
169                let body = PutBody { record, expected };
170                let resp = maybe_auth(client.put(url).json(&body), token)
171                    .send()
172                    .await
173                    .map_err(be)?;
174                // Read the body as text first so error statuses (403/413/400,
175                // which carry plain-text bodies) surface their real status and
176                // message instead of being masked as a JSON decode error (#147).
177                let status = resp.status();
178                let text = resp.text().await.map_err(be)?;
179                classify_put_response(status, &text)
180            }
181            Backend::Grpc { client, token } => {
182                let mut client = client.clone();
183                let req = grpc_request(
184                    PutRequest {
185                        record_json: serde_json::to_vec(&record).map_err(se)?,
186                        expected_json: serde_json::to_vec(&expected).map_err(se)?,
187                    },
188                    token,
189                )?;
190                let resp = client.put(req).await.map_err(status)?.into_inner();
191                match resp.outcome.as_str() {
192                    "committed" => {
193                        let rev: Revision =
194                            serde_json::from_slice(&resp.payload_json).map_err(se)?;
195                        Ok(PutResult::Committed(rev))
196                    }
197                    "conflict" => {
198                        let c: Conflict = serde_json::from_slice(&resp.payload_json).map_err(se)?;
199                        Ok(PutResult::Conflict(Box::new(c)))
200                    }
201                    other => Err(CoreError::Backend(format!("unknown put outcome: {other}"))),
202                }
203            }
204        }
205    }
206
207    async fn list(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
208        match &self.backend {
209            Backend::Http {
210                base,
211                client,
212                token,
213            } => {
214                let mut url = base.clone();
215                url.path_segments_mut()
216                    .map_err(|_| CoreError::Backend("base URL cannot be a base".into()))?
217                    .extend(["v1", "keys"]);
218                {
219                    let mut q = url.query_pairs_mut();
220                    if let Some(ns) = &prefix.namespace {
221                        q.append_pair("namespace", ns);
222                    }
223                    if let Some(col) = &prefix.collection {
224                        q.append_pair("collection", col);
225                    }
226                }
227                let resp = maybe_auth(client.get(url), token)
228                    .send()
229                    .await
230                    .map_err(be)?
231                    .error_for_status()
232                    .map_err(be)?;
233                Ok(resp.json::<Vec<RecordKey>>().await.map_err(be)?)
234            }
235            Backend::Grpc { client, token } => {
236                let mut client = client.clone();
237                let req = grpc_request(
238                    ListRequest {
239                        namespace: prefix.namespace.clone(),
240                        collection: prefix.collection.clone(),
241                    },
242                    token,
243                )?;
244                let resp = client.list(req).await.map_err(status)?.into_inner();
245                resp.keys_json
246                    .iter()
247                    .map(|b| serde_json::from_slice::<RecordKey>(b).map_err(se))
248                    .collect()
249            }
250        }
251    }
252
253    async fn delete(&self, key: &RecordKey, expected: Option<Revision>) -> Result<DeleteResult> {
254        match &self.backend {
255            Backend::Http {
256                base,
257                client,
258                token,
259            } => {
260                let url = Self::records_url(base, key)?;
261                let body = DeleteBody { expected };
262                let resp = maybe_auth(client.delete(url).json(&body), token)
263                    .send()
264                    .await
265                    .map_err(be)?;
266                // Read the body as text first so error statuses (403/400, which
267                // carry plain-text bodies) surface their real status and message
268                // instead of being masked as a JSON decode error (mirrors `put`).
269                let status = resp.status();
270                let text = resp.text().await.map_err(be)?;
271                classify_delete_response(status, &text)
272            }
273            Backend::Grpc { client, token } => {
274                let mut client = client.clone();
275                let req = grpc_request(
276                    DeleteRequest {
277                        namespace: key.namespace.clone(),
278                        collection: key.collection.clone(),
279                        id: key.id.clone(),
280                        expected_json: serde_json::to_vec(&expected).map_err(se)?,
281                    },
282                    token,
283                )?;
284                let resp = client.delete(req).await.map_err(status)?.into_inner();
285                match resp.outcome.as_str() {
286                    "deleted" => Ok(DeleteResult::Deleted),
287                    "conflict" => {
288                        let c: Conflict = serde_json::from_slice(&resp.payload_json).map_err(se)?;
289                        Ok(DeleteResult::Conflict(Box::new(c)))
290                    }
291                    other => Err(CoreError::Backend(format!(
292                        "unknown delete outcome: {other}"
293                    ))),
294                }
295            }
296        }
297    }
298}
299
300#[async_trait]
301impl BlobStore for ServerStore {
302    async fn put_blob(&self, content: &[u8]) -> Result<ContentHash> {
303        // The blob is content-addressed: compute the hash locally to address
304        // the request, exactly as the daemon will recompute and verify it.
305        let hash = ContentHash::of(content);
306        match &self.backend {
307            Backend::Http {
308                base,
309                client,
310                token,
311            } => {
312                let url = Self::blobs_url(base, Some(&hash.0))?;
313                let resp = maybe_auth(client.put(url).body(content.to_vec()), token)
314                    .send()
315                    .await
316                    .map_err(be)?;
317                let status = resp.status();
318                let text = resp.text().await.map_err(be)?;
319                classify_blob_put_response(status, &text, hash)
320            }
321            Backend::Grpc { client, token } => {
322                let mut client = client.clone();
323                let req = grpc_request(
324                    PutBlobRequest {
325                        hash: hash.0.clone(),
326                        content: content.to_vec(),
327                    },
328                    token,
329                )?;
330                let resp = client.put_blob(req).await.map_err(status)?.into_inner();
331                Ok(ContentHash(resp.hash))
332            }
333        }
334    }
335
336    async fn get_blob(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
337        match &self.backend {
338            Backend::Http {
339                base,
340                client,
341                token,
342            } => {
343                let url = Self::blobs_url(base, Some(&hash.0))?;
344                let resp = maybe_auth(client.get(url), token)
345                    .send()
346                    .await
347                    .map_err(be)?;
348                if resp.status() == reqwest::StatusCode::NOT_FOUND {
349                    return Ok(None);
350                }
351                let resp = resp.error_for_status().map_err(be)?;
352                Ok(Some(resp.bytes().await.map_err(be)?.to_vec()))
353            }
354            Backend::Grpc { client, token } => {
355                let mut client = client.clone();
356                let req = grpc_request(
357                    GetBlobRequest {
358                        hash: hash.0.clone(),
359                    },
360                    token,
361                )?;
362                let resp = client.get_blob(req).await.map_err(status)?.into_inner();
363                Ok(resp.found.then_some(resp.content))
364            }
365        }
366    }
367
368    async fn list_blobs(&self) -> Result<Vec<ContentHash>> {
369        match &self.backend {
370            Backend::Http {
371                base,
372                client,
373                token,
374            } => {
375                let url = Self::blobs_url(base, None)?;
376                let resp = maybe_auth(client.get(url), token)
377                    .send()
378                    .await
379                    .map_err(be)?
380                    .error_for_status()
381                    .map_err(be)?;
382                Ok(resp.json::<Vec<ContentHash>>().await.map_err(be)?)
383            }
384            Backend::Grpc { client, token } => {
385                let mut client = client.clone();
386                let req = grpc_request(ListBlobsRequest {}, token)?;
387                let resp = client.list_blobs(req).await.map_err(status)?.into_inner();
388                Ok(resp.hashes.into_iter().map(ContentHash).collect())
389            }
390        }
391    }
392
393    async fn delete_blob(&self, hash: &ContentHash) -> Result<()> {
394        match &self.backend {
395            Backend::Http {
396                base,
397                client,
398                token,
399            } => {
400                let url = Self::blobs_url(base, Some(&hash.0))?;
401                let resp = maybe_auth(client.delete(url), token)
402                    .send()
403                    .await
404                    .map_err(be)?;
405                let status = resp.status();
406                if status == reqwest::StatusCode::OK {
407                    return Ok(());
408                }
409                let text = resp.text().await.map_err(be)?;
410                Err(CoreError::Backend(format!(
411                    "daemon returned {status}: {text}"
412                )))
413            }
414            Backend::Grpc { client, token } => {
415                let mut client = client.clone();
416                let req = grpc_request(
417                    DeleteBlobRequest {
418                        hash: hash.0.clone(),
419                    },
420                    token,
421                )?;
422                client.delete_blob(req).await.map_err(status)?;
423                Ok(())
424            }
425        }
426    }
427}
428
429fn delete_outcome_to_result(outcome: DeleteOutcome) -> DeleteResult {
430    match outcome {
431        DeleteOutcome::Deleted => DeleteResult::Deleted,
432        DeleteOutcome::Conflict { conflict } => DeleteResult::Conflict(conflict),
433    }
434}
435
436/// Decide a `delete`'s result from the HTTP response status and body text.
437///
438/// The daemon speaks JSON (`DeleteOutcome`) only for `200 OK` (deleted) and
439/// `409 Conflict`; every other status (`403` authorization denial, `400` bad
440/// request) carries a plain-text body, surfaced verbatim as
441/// `CoreError::Backend("daemon returned <status>: <body>")` (mirrors `put`).
442fn classify_delete_response(status: reqwest::StatusCode, body: &str) -> Result<DeleteResult> {
443    match status {
444        reqwest::StatusCode::OK | reqwest::StatusCode::CONFLICT => {
445            let outcome: DeleteOutcome = serde_json::from_str(body).map_err(se)?;
446            Ok(delete_outcome_to_result(outcome))
447        }
448        other => Err(CoreError::Backend(format!(
449            "daemon returned {other}: {body}"
450        ))),
451    }
452}
453
454fn outcome_to_result(outcome: PutOutcome) -> PutResult {
455    match outcome {
456        PutOutcome::Committed { revision } => PutResult::Committed(revision),
457        PutOutcome::Conflict { conflict } => PutResult::Conflict(conflict),
458    }
459}
460
461/// Decide a `put`'s result from the HTTP response status and body text.
462///
463/// The daemon speaks JSON (`PutOutcome`) only for `200 OK` (committed) and
464/// `409 Conflict`; every other status — notably `403` (namespace authorization
465/// denial), `413` (too large), and `400` (bad request) — carries a plain-text
466/// body. Feeding those bodies to a JSON parser masks the real failure behind a
467/// generic "error decoding response body" (#147), so any other status is
468/// surfaced verbatim as `CoreError::Backend("daemon returned <status>: <body>")`.
469fn classify_put_response(status: reqwest::StatusCode, body: &str) -> Result<PutResult> {
470    match status {
471        reqwest::StatusCode::OK | reqwest::StatusCode::CONFLICT => {
472            let outcome: PutOutcome = serde_json::from_str(body).map_err(se)?;
473            Ok(outcome_to_result(outcome))
474        }
475        other => Err(CoreError::Backend(format!(
476            "daemon returned {other}: {body}"
477        ))),
478    }
479}
480
481/// Decide a blob `put`'s result from the HTTP response status and body text.
482/// `200 OK` → the (already-known) `hash`; every other status carries a plain
483/// text body surfaced verbatim as `Backend("daemon returned <status>: <body>")`
484/// — notably `413` (too large), `403` (authz), and `400` (hash mismatch) — so
485/// the real failure is never masked (#147).
486fn classify_blob_put_response(
487    status: reqwest::StatusCode,
488    body: &str,
489    hash: ContentHash,
490) -> Result<ContentHash> {
491    match status {
492        reqwest::StatusCode::OK => Ok(hash),
493        other => Err(CoreError::Backend(format!(
494            "daemon returned {other}: {body}"
495        ))),
496    }
497}
498
499fn be<E: std::fmt::Display>(e: E) -> CoreError {
500    CoreError::Backend(e.to_string())
501}
502fn se<E: std::fmt::Display>(e: E) -> CoreError {
503    CoreError::Serde(e.to_string())
504}
505fn status(s: tonic::Status) -> CoreError {
506    CoreError::Backend(s.to_string())
507}
508
509#[cfg(test)]
510mod tests {
511    use super::*;
512    use gonzalo_core::store::Conflict;
513    use gonzalo_core::{Body, ContentHash, Identity, Meta, Record, RecordKind};
514    use reqwest::StatusCode;
515    use std::collections::BTreeMap;
516
517    fn sample_record() -> Record {
518        let body = Body::Inline(b"hello".to_vec());
519        Record {
520            key: RecordKey::new("ns", "col", "id"),
521            kind: RecordKind::Topic,
522            revision: Revision::initial(body.bytes()),
523            parent: None,
524            body,
525            meta: Meta {
526                author: Identity::new("tester"),
527                origin_system: "test".into(),
528                created: 0,
529                updated: 0,
530                labels: BTreeMap::new(),
531            },
532            links: Vec::new(),
533        }
534    }
535
536    /// `200 OK` carries a `committed` JSON body → `PutResult::Committed`.
537    #[test]
538    fn ok_body_parses_committed() {
539        let revision = Revision::initial(b"hello");
540        let json = serde_json::to_string(&PutOutcome::Committed {
541            revision: revision.clone(),
542        })
543        .unwrap();
544        let result = classify_put_response(StatusCode::OK, &json).unwrap();
545        assert!(matches!(result, PutResult::Committed(r) if r == revision));
546    }
547
548    /// `409 Conflict` carries a `conflict` JSON body → `PutResult::Conflict`.
549    #[test]
550    fn conflict_body_parses_conflict() {
551        let record = sample_record();
552        let conflict = Conflict {
553            key: record.key.clone(),
554            expected: None,
555            current: record,
556        };
557        let json = serde_json::to_string(&PutOutcome::Conflict {
558            conflict: Box::new(conflict),
559        })
560        .unwrap();
561        let result = classify_put_response(StatusCode::CONFLICT, &json).unwrap();
562        assert!(matches!(result, PutResult::Conflict(_)));
563    }
564
565    /// #147: a `403` with a plain-text body surfaces the status + body as a
566    /// `Backend` error — NOT a masked JSON decode error.
567    #[test]
568    fn forbidden_surfaces_status_and_body() {
569        let body = "principal \"alice\" lacks Write on namespace \"secrets\"";
570        let err = classify_put_response(StatusCode::FORBIDDEN, body).unwrap_err();
571        match err {
572            CoreError::Backend(msg) => {
573                assert!(msg.contains("403"), "want status 403 in {msg:?}");
574                assert!(msg.contains(body), "want daemon body in {msg:?}");
575                assert!(
576                    !msg.contains("decoding"),
577                    "must not be a decode error: {msg:?}"
578                );
579            }
580            other => panic!("expected Backend error, got {other:?}"),
581        }
582    }
583
584    /// `413 Payload Too Large` (plain-text body) also surfaces status + body.
585    #[test]
586    fn payload_too_large_surfaces_status_and_body() {
587        let body = "record exceeds max size";
588        let err = classify_put_response(StatusCode::PAYLOAD_TOO_LARGE, body).unwrap_err();
589        match err {
590            CoreError::Backend(msg) => {
591                assert!(msg.contains("413"), "want status 413 in {msg:?}");
592                assert!(msg.contains(body), "want daemon body in {msg:?}");
593            }
594            other => panic!("expected Backend error, got {other:?}"),
595        }
596    }
597
598    /// `400 Bad Request` (plain-text body) surfaces status + body, not a decode
599    /// error.
600    #[test]
601    fn bad_request_surfaces_status_and_body() {
602        let body = "path/body key disagreement";
603        let err = classify_put_response(StatusCode::BAD_REQUEST, body).unwrap_err();
604        match err {
605            CoreError::Backend(msg) => {
606                assert!(msg.contains("400"), "want status 400 in {msg:?}");
607                assert!(msg.contains(body), "want daemon body in {msg:?}");
608            }
609            other => panic!("expected Backend error, got {other:?}"),
610        }
611    }
612
613    #[test]
614    fn blob_put_ok_returns_the_hash() {
615        let content = b"blob body";
616        let hash = ContentHash::of(content);
617        let result = classify_blob_put_response(StatusCode::OK, "", hash.clone()).unwrap();
618        assert_eq!(result, hash);
619    }
620
621    #[test]
622    fn blob_put_413_surfaces_status_and_body() {
623        let hash = ContentHash::of(b"x");
624        let err = classify_blob_put_response(
625            StatusCode::PAYLOAD_TOO_LARGE,
626            "blob exceeds max size",
627            hash,
628        )
629        .unwrap_err();
630        match err {
631            CoreError::Backend(msg) => {
632                assert!(msg.contains("413"), "want status 413 in {msg:?}");
633                assert!(
634                    msg.contains("blob exceeds max size"),
635                    "want body in {msg:?}"
636                );
637            }
638            other => panic!("expected Backend error, got {other:?}"),
639        }
640    }
641
642    #[test]
643    fn blob_put_400_mismatch_surfaces_status_and_body() {
644        let hash = ContentHash::of(b"x");
645        let err = classify_blob_put_response(
646            StatusCode::BAD_REQUEST,
647            "blob content does not match the URL hash",
648            hash,
649        )
650        .unwrap_err();
651        match err {
652            CoreError::Backend(msg) => {
653                assert!(msg.contains("400"), "want status 400 in {msg:?}");
654                assert!(msg.contains("does not match"), "want body in {msg:?}");
655            }
656            other => panic!("expected Backend error, got {other:?}"),
657        }
658    }
659}
gonzalo_store_server/lib.rs

gonzalo_store_server/
lib.rs
