1use serde::Serialize;
30use serde::de::DeserializeOwned;
31use snafu::ResultExt;
32
33use crate::cassettes::discovery::Discovery;
34use crate::core::contract::{self, core, ops};
35use crate::core::models::params::ContractParams;
36use crate::core::models::{
37 RawTurnListResponse, SeedDemoRequest, SeedResult, SessionDetailResponse, SessionItem,
38 SessionListParams, SessionListResponse, SessionTracesParams, SessionTracesResponse,
39 SessionUpdateRequest, SpanItem, StatsParams, StatsResponse, TraceDetail, TraceListParams,
40 TraceListResponse, TraceParams,
41};
42use crate::decode;
43use crate::error::{Result, error};
44use crate::page;
45use crate::transport::{StreamingTransport, TapesTransport, WireRequest};
46
47#[derive(Debug, Clone, Copy)]
49pub struct CoreClient<T> {
50 transport: T,
51}
52
53impl<T> CoreClient<T> {
54 #[must_use]
56 pub fn new(transport: T) -> Self {
57 Self { transport }
58 }
59
60 #[must_use]
62 pub fn transport(&self) -> &T {
63 &self.transport
64 }
65
66 #[must_use]
68 pub fn into_transport(self) -> T {
69 self.transport
70 }
71}
72
73impl<T: TapesTransport> CoreClient<T> {
74 pub async fn call<R: DeserializeOwned>(
86 &self,
87 operation_id: &str,
88 values: Vec<(&str, String)>,
89 ) -> Result<R> {
90 self.call_with_body(operation_id, values, None).await
91 }
92
93 pub async fn call_with_body<R: DeserializeOwned>(
106 &self,
107 operation_id: &str,
108 values: Vec<(&str, String)>,
109 body: Option<String>,
110 ) -> Result<R> {
111 self.call_shaped(operation_id, values, &[], body).await
112 }
113
114 pub async fn call_with_claimed<R: DeserializeOwned>(
137 &self,
138 operation_id: &str,
139 values: Vec<(&str, String)>,
140 claimed: &[(String, String)],
141 ) -> Result<R> {
142 self.call_shaped(operation_id, values, claimed, None).await
143 }
144
145 async fn call_shaped<R: DeserializeOwned>(
169 &self,
170 operation_id: &str,
171 values: Vec<(&str, String)>,
172 claimed: &[(String, String)],
173 body: Option<String>,
174 ) -> Result<R> {
175 if !claimed.is_empty() && !ops::CLAIM_BEARING_OPS.contains(&operation_id) {
176 return error::ContractClaimsSnafu {
177 operation: operation_id,
178 }
179 .fail();
180 }
181 let method = core()?.method(operation_id)?;
182 let mut request = contract::call_for_with_body(method, values, body)?;
183 request.query.extend(claimed.iter().cloned());
184 let response = self
185 .transport
186 .send(&request)
187 .await
188 .context(error::TransportSnafu)?;
189 decode::json_typed(&response)
190 }
191
192 pub fn request_for(
202 &self,
203 operation_id: &str,
204 values: Vec<(&str, String)>,
205 ) -> Result<WireRequest<'static>> {
206 contract::call_for(core()?.method(operation_id)?, values)
207 }
208
209 async fn with_params<P: ContractParams, R: DeserializeOwned>(&self, params: &P) -> Result<R> {
211 self.call(P::OPERATION, params.values()).await
212 }
213
214 async fn with_params_at<P: ContractParams, R: DeserializeOwned>(
216 &self,
217 params: &P,
218 path: Vec<(&str, String)>,
219 ) -> Result<R> {
220 let mut values: Vec<(&str, String)> = params.values();
221 values.extend(path);
222 self.call(P::OPERATION, values).await
223 }
224
225 async fn with_body<B: Serialize, R: DeserializeOwned>(
227 &self,
228 operation_id: &str,
229 values: Vec<(&str, String)>,
230 body: &B,
231 ) -> Result<R> {
232 let rendered = serde_json::to_string(body).context(error::RenderBodySnafu)?;
233 self.call_with_body(operation_id, values, Some(rendered))
234 .await
235 }
236
237 pub async fn list_sessions(&self, params: &SessionListParams) -> Result<SessionListResponse> {
243 self.call_shaped(ops::LIST_SESSIONS, params.values(), ¶ms.claimed, None)
244 .await
245 }
246
247 pub async fn list_all_sessions(&self, params: &SessionListParams) -> Result<Vec<SessionItem>> {
257 page::walk(|cursor| {
258 let mut params = params.clone();
259 params.cursor = cursor;
260 async move { Ok(self.list_sessions(¶ms).await?.into_page()) }
261 })
262 .await
263 }
264
265 pub async fn get_session(&self, id: &str) -> Result<SessionDetailResponse> {
271 self.call(ops::GET_SESSION, vec![("id", id.to_owned())])
272 .await
273 }
274
275 pub async fn update_session(
281 &self,
282 id: &str,
283 body: &SessionUpdateRequest,
284 ) -> Result<SessionDetailResponse> {
285 self.with_body(ops::UPDATE_SESSION, vec![("id", id.to_owned())], body)
286 .await
287 }
288
289 pub async fn delete_session(&self, id: &str) -> Result<()> {
295 self.call(ops::DELETE_SESSION, vec![("id", id.to_owned())])
296 .await
297 }
298
299 pub async fn get_session_traces(
305 &self,
306 id: &str,
307 params: &SessionTracesParams,
308 ) -> Result<SessionTracesResponse> {
309 self.with_params_at(params, vec![("id", id.to_owned())])
310 .await
311 }
312
313 pub async fn list_raw_turns(&self, id: &str) -> Result<RawTurnListResponse> {
319 self.call(ops::LIST_RAW_TURNS, vec![("id", id.to_owned())])
320 .await
321 }
322
323 pub async fn list_traces(&self, params: &TraceListParams) -> Result<TraceListResponse> {
329 self.with_params(params).await
330 }
331
332 pub async fn get_trace(&self, trace_id: &str, params: &TraceParams) -> Result<TraceDetail> {
338 self.with_params_at(params, vec![("trace_id", trace_id.to_owned())])
339 .await
340 }
341
342 pub async fn get_span(&self, trace_id: &str, span_id: &str) -> Result<SpanItem> {
348 self.call(
349 ops::GET_SPAN,
350 vec![
351 ("trace_id", trace_id.to_owned()),
352 ("span_id", span_id.to_owned()),
353 ],
354 )
355 .await
356 }
357
358 pub async fn get_stats(&self, params: &StatsParams) -> Result<StatsResponse> {
364 self.with_params(params).await
365 }
366
367 pub async fn list_cassettes(&self) -> Result<Discovery> {
378 self.call(ops::LIST_CASSETTES, Vec::new()).await
379 }
380
381 pub async fn seed_demo(&self, body: &SeedDemoRequest) -> Result<SeedResult> {
387 self.with_body(ops::SEED_DEMO, Vec::new(), body).await
388 }
389}
390
391impl<T: StreamingTransport> CoreClient<T> {
392 pub async fn stream(&self, operation_id: &str, values: Vec<(&str, String)>) -> Result<T::Body> {
405 let method = core()?.method(operation_id)?;
406 let request = contract::call_for(method, values)?;
407 self.transport.send_stream(&request).await
408 }
409}
410
411#[cfg(test)]
412#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
413mod tests {
414 use super::*;
415 use crate::core::models::params::PayloadDetail;
416 use crate::path::{PathMode, call_url};
417 use crate::transport::{TransportError, WireResponse};
418 use serde::Deserialize;
419 use serde_json::Value;
420 use std::cell::RefCell;
421 use url::Url;
422
423 struct Recorder {
431 base: Url,
432 responses: RefCell<Vec<Value>>,
433 seen: RefCell<Vec<String>>,
434 bodies: RefCell<Vec<Option<String>>>,
435 }
436
437 impl Recorder {
438 fn new(base: &str, responses: Vec<Value>) -> Self {
439 Self {
440 base: Url::parse(base).unwrap(),
441 responses: RefCell::new(responses),
442 seen: RefCell::new(Vec::new()),
443 bodies: RefCell::new(Vec::new()),
444 }
445 }
446 }
447
448 impl TapesTransport for Recorder {
449 async fn send(
450 &self,
451 request: &WireRequest<'_>,
452 ) -> std::result::Result<WireResponse, TransportError> {
453 let url = call_url(&self.base, request, PathMode::UnderBase)
454 .map_err(|error| TransportError::new(error.to_string()))?;
455 self.seen.borrow_mut().push(url.to_string());
456 self.bodies.borrow_mut().push(request.body.clone());
457 let mut responses = self.responses.borrow_mut();
458 let body = if responses.len() > 1 {
459 responses.remove(0)
460 } else {
461 responses.first().cloned().unwrap_or(Value::Null)
462 };
463 Ok(WireResponse::new(
464 200,
465 url.to_string(),
466 Vec::new(),
467 body.to_string().into_bytes(),
468 ))
469 }
470 }
471
472 fn client(base: &str, response: Value) -> CoreClient<Recorder> {
473 CoreClient::new(Recorder::new(base, vec![response]))
474 }
475
476 #[tokio::test]
477 async fn an_operation_is_routed_through_the_contract_and_the_transport() {
478 let client = client(
479 "https://acme.example/primary/tapes/",
480 serde_json::json!({"traces": []}),
481 );
482 let _ = client
483 .get_session_traces("s-1", &SessionTracesParams::default())
484 .await
485 .unwrap();
486
487 assert_eq!(
488 client.transport().seen.borrow()[0],
489 "https://acme.example/primary/tapes/v1/sessions/s-1/traces",
490 );
491 }
492
493 #[tokio::test]
494 async fn a_typed_method_decodes_the_contracts_own_shape() {
495 let client = client(
498 "http://127.0.0.1:8081",
499 serde_json::json!({
500 "items": [{"id": "s1", "rollup": {"turn_count": 3}}],
501 "next_cursor": "abc",
502 }),
503 );
504 let listing = client
505 .list_sessions(&SessionListParams::default())
506 .await
507 .unwrap();
508
509 assert_eq!(listing.items[0].id, "s1");
510 assert_eq!(listing.items[0].rollup.turn_count, 3);
511 assert_eq!(listing.next_cursor, "abc");
512 }
513
514 #[tokio::test]
515 async fn a_typed_method_survives_a_field_it_has_never_heard_of() {
516 let client = client(
519 "http://127.0.0.1:8081",
520 serde_json::json!({"items": [{"id": "s1", "a_field_from_the_future": 7}]}),
521 );
522 let listing = client
523 .list_sessions(&SessionListParams::default())
524 .await
525 .unwrap();
526 assert_eq!(listing.items[0].id, "s1");
527 }
528
529 #[tokio::test]
530 async fn the_generic_seam_still_decodes_into_a_callers_own_type() {
531 #[derive(Debug, Deserialize)]
534 struct Listing {
535 next_cursor: String,
536 }
537
538 let client = client(
539 "http://127.0.0.1:8081",
540 serde_json::json!({"items": [], "next_cursor": "abc"}),
541 );
542 let got: Listing = client.call(ops::LIST_SESSIONS, Vec::new()).await.unwrap();
543 assert_eq!(got.next_cursor, "abc");
544
545 let raw: Value = client.call(ops::LIST_SESSIONS, Vec::new()).await.unwrap();
546 assert_eq!(raw["next_cursor"], "abc");
547 }
548
549 #[tokio::test]
550 async fn a_typed_parameter_travels_under_the_contracts_own_name() {
551 let client = client("http://127.0.0.1:8081", serde_json::json!({"traces": []}));
552 let _ = client
553 .get_session_traces(
554 "s-1",
555 &SessionTracesParams {
556 payload: Some(PayloadDetail::Preview),
557 },
558 )
559 .await
560 .unwrap();
561 assert!(
562 client.transport().seen.borrow()[0].ends_with("/traces?payload=preview"),
563 "got: {:?}",
564 client.transport().seen.borrow(),
565 );
566 }
567
568 #[tokio::test]
569 async fn a_listing_walk_follows_the_cursor_to_the_end() {
570 let client = CoreClient::new(Recorder::new(
573 "http://127.0.0.1:8081",
574 vec![
575 serde_json::json!({"items": [{"id": "s1"}], "next_cursor": "c1"}),
576 serde_json::json!({"items": [{"id": "s2"}], "next_cursor": ""}),
577 ],
578 ));
579 let sessions = client
580 .list_all_sessions(&SessionListParams::default())
581 .await
582 .unwrap();
583
584 assert_eq!(
585 sessions.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(),
586 vec!["s1", "s2"],
587 );
588 assert!(
589 client.transport().seen.borrow()[1].contains("cursor=c1"),
590 "got: {:?}",
591 client.transport().seen.borrow(),
592 );
593 }
594
595 #[tokio::test]
596 async fn an_undeclared_parameter_is_refused_before_the_transport_is_reached() {
597 let client = client("http://127.0.0.1:8081", Value::Null);
598 let err = client
599 .call::<Value>(ops::GET_SESSION, vec![("payolad", "full".to_owned())])
600 .await
601 .unwrap_err();
602 assert!(err.to_string().contains("payolad"), "got: {err}");
603 assert!(
604 client.transport().seen.borrow().is_empty(),
605 "nothing may be sent for a call the contract refused",
606 );
607 }
608
609 #[tokio::test]
610 async fn a_typed_body_reaches_the_transport_as_the_contracts_own_json() {
611 let client = client(
615 "http://127.0.0.1:8081",
616 serde_json::json!({"session": {"id": "s-1"}}),
617 );
618 let updated = client
619 .update_session(
620 "s-1",
621 &SessionUpdateRequest {
622 display_name: Some("gum glow charm".to_owned()),
623 },
624 )
625 .await
626 .unwrap();
627
628 assert_eq!(updated.session.id, "s-1");
629 let bodies = client.transport().bodies.borrow();
630 let sent: Value = serde_json::from_str(bodies[0].as_deref().unwrap()).unwrap();
631 assert_eq!(sent["display_name"], "gum glow charm");
632 }
633
634 #[tokio::test]
635 async fn the_bodyless_facade_refuses_an_operation_that_requires_a_body() {
636 let client = client("http://127.0.0.1:8081", Value::Null);
640 let err = client
641 .call::<Value>(ops::UPDATE_SESSION, vec![("id", "s-1".to_owned())])
642 .await
643 .unwrap_err();
644
645 assert!(
646 err.to_string().contains("requires a request body"),
647 "got: {err}",
648 );
649 assert!(client.transport().seen.borrow().is_empty());
650 }
651
652 #[tokio::test]
653 async fn the_facade_refuses_a_body_on_an_operation_that_declares_none() {
654 let client = client("http://127.0.0.1:8081", Value::Null);
655 let err = client
656 .call_with_body::<Value>(
657 ops::GET_SESSION,
658 vec![("id", "s-1".to_owned())],
659 Some("{}".to_owned()),
660 )
661 .await
662 .unwrap_err();
663
664 assert!(
665 err.to_string().contains("declares no request body"),
666 "got: {err}",
667 );
668 assert!(client.transport().seen.borrow().is_empty());
669 }
670
671 #[tokio::test]
672 async fn the_bodyless_facade_still_sends_no_body_for_an_ordinary_read() {
673 let client = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
676 let _ = client
677 .list_sessions(&SessionListParams::default())
678 .await
679 .unwrap();
680 assert_eq!(client.transport().bodies.borrow().as_slice(), [None]);
681 }
682
683 impl crate::transport::StreamingTransport for Recorder {
684 type Body = Vec<u8>;
685
686 async fn send_stream(&self, request: &WireRequest<'_>) -> Result<Self::Body> {
687 let url = call_url(&self.base, request, PathMode::UnderBase).map_err(|error| {
688 crate::Error::Transport {
689 source: TransportError::new(error.to_string()),
690 }
691 })?;
692 self.seen.borrow_mut().push(url.to_string());
693 Ok(Vec::new())
694 }
695 }
696
697 #[tokio::test]
698 async fn the_stream_escape_hatch_builds_the_contract_url() {
699 let client = CoreClient::new(Recorder::new(
703 "http://127.0.0.1:8081",
704 vec![serde_json::json!({})],
705 ));
706 let _ = client
707 .stream(ops::LIST_RAW_TURNS, vec![("id", "s-1".to_owned())])
708 .await
709 .unwrap();
710 let seen = client.transport().seen.borrow();
711 assert_eq!(seen[0], "http://127.0.0.1:8081/v1/sessions/s-1/raw_turns");
712 }
713
714 #[tokio::test]
715 async fn a_named_method_and_its_operation_id_build_the_same_request() {
716 let named = client("http://127.0.0.1:8081", serde_json::json!({}));
720 let _ = named.get_span("t-1", "sp-1").await.unwrap();
721
722 let raw = client("http://127.0.0.1:8081", serde_json::json!({}));
723 let _: Value = raw
724 .call(
725 ops::GET_SPAN,
726 vec![
727 ("trace_id", "t-1".to_owned()),
728 ("span_id", "sp-1".to_owned()),
729 ],
730 )
731 .await
732 .unwrap();
733
734 assert_eq!(
735 *named.transport().seen.borrow(),
736 *raw.transport().seen.borrow()
737 );
738 }
739
740 #[tokio::test]
741 async fn claimed_params_append_to_the_query_in_order() {
742 let client = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
748 let _ = client
749 .list_sessions(&SessionListParams {
750 limit: Some(25),
751 claimed: vec![
752 ("flavor".to_owned(), "grape".to_owned()),
753 ("flavor".to_owned(), "sour cherry".to_owned()),
754 ("vintage".to_owned(), "1998".to_owned()),
755 ],
756 ..Default::default()
757 })
758 .await
759 .unwrap();
760 assert_eq!(
761 client.transport().seen.borrow()[0],
762 "http://127.0.0.1:8081/v1/sessions?limit=25&flavor=grape&flavor=sour+cherry&vintage=1998",
763 );
764 }
765
766 #[tokio::test]
767 async fn claimed_values_are_percent_encoded_and_nothing_more() {
768 let client = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
773 let _ = client
774 .list_sessions(&SessionListParams {
775 claimed: vec![("flavor".to_owned(), "Grüße 🍇".to_owned())],
776 ..Default::default()
777 })
778 .await
779 .unwrap();
780 assert_eq!(
781 client.transport().seen.borrow()[0],
782 "http://127.0.0.1:8081/v1/sessions?flavor=Gr%C3%BC%C3%9Fe+%F0%9F%8D%87",
783 );
784 }
785
786 #[tokio::test]
787 async fn an_empty_claimed_set_leaves_the_request_as_it_always_was() {
788 let client = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
791 let _ = client
792 .list_sessions(&SessionListParams::default())
793 .await
794 .unwrap();
795 assert_eq!(
796 client.transport().seen.borrow()[0],
797 "http://127.0.0.1:8081/v1/sessions",
798 );
799 }
800
801 #[tokio::test]
802 async fn the_page_walk_carries_claimed_params_onto_every_page() {
803 let client = CoreClient::new(Recorder::new(
807 "http://127.0.0.1:8081",
808 vec![
809 serde_json::json!({"items": [{"id": "s1"}], "next_cursor": "c1"}),
810 serde_json::json!({"items": [{"id": "s2"}], "next_cursor": ""}),
811 ],
812 ));
813 let _ = client
814 .list_all_sessions(&SessionListParams {
815 claimed: vec![("flavor".to_owned(), "grape".to_owned())],
816 ..Default::default()
817 })
818 .await
819 .unwrap();
820 let seen = client.transport().seen.borrow();
821 assert_eq!(seen.len(), 2);
822 assert!(seen[1].contains("cursor=c1"), "got: {seen:?}");
823 assert!(
824 seen.iter().all(|url| url.contains("flavor=grape")),
825 "every page of a filtered walk must carry the claimed pairs: {seen:?}",
826 );
827 }
828
829 #[tokio::test]
830 async fn the_typed_and_untyped_claimed_spellings_build_the_same_request() {
831 let named = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
836 let _ = named
837 .list_sessions(&SessionListParams {
838 limit: Some(1),
839 claimed: vec![("flavor".to_owned(), "grape".to_owned())],
840 ..Default::default()
841 })
842 .await
843 .unwrap();
844
845 let raw = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
846 let _: Value = raw
847 .call_with_claimed(
848 ops::LIST_SESSIONS,
849 vec![("limit", "1".to_owned())],
850 &[("flavor".to_owned(), "grape".to_owned())],
851 )
852 .await
853 .unwrap();
854
855 assert_eq!(
856 *named.transport().seen.borrow(),
857 *raw.transport().seen.borrow()
858 );
859 }
860
861 #[tokio::test]
862 async fn claimed_params_do_not_loosen_the_declared_parameter_refusal() {
863 let client = client("http://127.0.0.1:8081", Value::Null);
867 let err = client
868 .call_with_claimed::<Value>(
869 ops::LIST_SESSIONS,
870 vec![("limt", "25".to_owned())],
871 &[("flavor".to_owned(), "grape".to_owned())],
872 )
873 .await
874 .unwrap_err();
875 assert!(err.to_string().contains("limt"), "got: {err}");
876 assert!(
877 client.transport().seen.borrow().is_empty(),
878 "nothing may be sent for a call the contract refused",
879 );
880 }
881
882 #[tokio::test]
883 async fn claimed_pairs_on_a_non_claim_bearing_operation_are_refused() {
884 let client = client("http://127.0.0.1:8081", Value::Null);
890 let err = client
891 .call_with_claimed::<Value>(
892 ops::GET_SESSION,
893 vec![("id", "s-1".to_owned())],
894 &[("flavor".to_owned(), "grape".to_owned())],
895 )
896 .await
897 .unwrap_err();
898 assert!(
899 err.to_string().contains("no claimed filter params"),
900 "got: {err}",
901 );
902 assert!(
903 client.transport().seen.borrow().is_empty(),
904 "nothing may be sent for a call the contract refused",
905 );
906 }
907
908 #[tokio::test]
909 async fn an_empty_claimed_set_is_permitted_on_every_operation() {
910 let client = client(
914 "http://127.0.0.1:8081",
915 serde_json::json!({"session": {"id": "s-1"}}),
916 );
917 let _: Value = client
918 .call_with_claimed(ops::GET_SESSION, vec![("id", "s-1".to_owned())], &[])
919 .await
920 .unwrap();
921 assert_eq!(
922 client.transport().seen.borrow()[0],
923 "http://127.0.0.1:8081/v1/sessions/s-1",
924 );
925 }
926
927 #[test]
928 fn every_claim_bearing_operation_is_in_the_vendored_contract() {
929 for operation in ops::CLAIM_BEARING_OPS {
933 assert!(
934 core().unwrap().method(operation).is_ok(),
935 "ops::CLAIM_BEARING_OPS names {operation:?}, which the vendored contract lacks",
936 );
937 }
938 }
939}