use serde::Serialize;
use serde::de::DeserializeOwned;
use snafu::ResultExt;
use crate::cassettes::discovery::Discovery;
use crate::core::contract::{self, core, ops};
use crate::core::models::params::ContractParams;
use crate::core::models::{
CreateSkillRequest, ExportSessionParams, ExportSessionsParams, GenerateSkillRequest,
PublishSkillRequest, RawTurnListResponse, SearchSpansParams, SeedDemoRequest, SeedResult,
SessionDetailResponse, SessionItem, SessionListParams, SessionListResponse,
SessionSkillsResponse, SessionTracesParams, SessionTracesResponse, SessionUpdateRequest,
SkillResponse, SkillVersionResponse, SkillVersionsResponse, SkillsListParams,
SkillsListResponse, SpanItem, SpanSearchOutput, StatsParams, StatsResponse, TraceDetail,
TraceListParams, TraceListResponse, TraceParams, UpdateSkillRequest,
};
use crate::decode;
use crate::error::{Result, error};
use crate::page;
use crate::transport::{StreamingTransport, TapesTransport, WireRequest};
#[derive(Debug, Clone, Copy)]
pub struct CoreClient<T> {
transport: T,
}
impl<T> CoreClient<T> {
#[must_use]
pub fn new(transport: T) -> Self {
Self { transport }
}
#[must_use]
pub fn transport(&self) -> &T {
&self.transport
}
#[must_use]
pub fn into_transport(self) -> T {
self.transport
}
}
impl<T: TapesTransport> CoreClient<T> {
pub async fn call<R: DeserializeOwned>(
&self,
operation_id: &str,
values: Vec<(&str, String)>,
) -> Result<R> {
self.call_with_body(operation_id, values, None).await
}
pub async fn call_with_body<R: DeserializeOwned>(
&self,
operation_id: &str,
values: Vec<(&str, String)>,
body: Option<String>,
) -> Result<R> {
let method = core()?.method(operation_id)?;
let request = contract::call_for_with_body(method, values, body)?;
let response = self
.transport
.send(&request)
.await
.context(error::TransportSnafu)?;
decode::json_typed(&response)
}
pub fn request_for(
&self,
operation_id: &str,
values: Vec<(&str, String)>,
) -> Result<WireRequest<'static>> {
contract::call_for(core()?.method(operation_id)?, values)
}
async fn with_params<P: ContractParams, R: DeserializeOwned>(&self, params: &P) -> Result<R> {
self.call(P::OPERATION, params.values()).await
}
async fn with_params_at<P: ContractParams, R: DeserializeOwned>(
&self,
params: &P,
path: Vec<(&str, String)>,
) -> Result<R> {
let mut values: Vec<(&str, String)> = params.values();
values.extend(path);
self.call(P::OPERATION, values).await
}
async fn with_body<B: Serialize, R: DeserializeOwned>(
&self,
operation_id: &str,
values: Vec<(&str, String)>,
body: &B,
) -> Result<R> {
let rendered = serde_json::to_string(body).context(error::RenderBodySnafu)?;
self.call_with_body(operation_id, values, Some(rendered))
.await
}
pub async fn list_sessions(&self, params: &SessionListParams) -> Result<SessionListResponse> {
self.with_params(params).await
}
pub async fn list_all_sessions(&self, params: &SessionListParams) -> Result<Vec<SessionItem>> {
page::walk(|cursor| {
let mut params = params.clone();
params.cursor = cursor;
async move { Ok(self.list_sessions(¶ms).await?.into_page()) }
})
.await
}
pub async fn get_session(&self, id: &str) -> Result<SessionDetailResponse> {
self.call(ops::GET_SESSION, vec![("id", id.to_owned())])
.await
}
pub async fn update_session(
&self,
id: &str,
body: &SessionUpdateRequest,
) -> Result<SessionDetailResponse> {
self.with_body(ops::UPDATE_SESSION, vec![("id", id.to_owned())], body)
.await
}
pub async fn delete_session(&self, id: &str) -> Result<()> {
self.call(ops::DELETE_SESSION, vec![("id", id.to_owned())])
.await
}
pub async fn get_session_traces(
&self,
id: &str,
params: &SessionTracesParams,
) -> Result<SessionTracesResponse> {
self.with_params_at(params, vec![("id", id.to_owned())])
.await
}
pub async fn list_raw_turns(&self, id: &str) -> Result<RawTurnListResponse> {
self.call(ops::LIST_RAW_TURNS, vec![("id", id.to_owned())])
.await
}
pub async fn list_session_skills(&self, id: &str) -> Result<SessionSkillsResponse> {
self.call(ops::LIST_SESSION_SKILLS, vec![("id", id.to_owned())])
.await
}
pub async fn list_traces(&self, params: &TraceListParams) -> Result<TraceListResponse> {
self.with_params(params).await
}
pub async fn get_trace(&self, trace_id: &str, params: &TraceParams) -> Result<TraceDetail> {
self.with_params_at(params, vec![("trace_id", trace_id.to_owned())])
.await
}
pub async fn get_span(&self, trace_id: &str, span_id: &str) -> Result<SpanItem> {
self.call(
ops::GET_SPAN,
vec![
("trace_id", trace_id.to_owned()),
("span_id", span_id.to_owned()),
],
)
.await
}
pub async fn search_spans(&self, params: &SearchSpansParams) -> Result<SpanSearchOutput> {
self.with_params(params).await
}
pub async fn get_stats(&self, params: &StatsParams) -> Result<StatsResponse> {
self.with_params(params).await
}
pub async fn list_skills(&self, params: &SkillsListParams) -> Result<SkillsListResponse> {
self.with_params(params).await
}
pub async fn list_all_skills(&self, params: &SkillsListParams) -> Result<Vec<SkillResponse>> {
page::walk(|cursor| {
let mut params = params.clone();
params.cursor = cursor;
async move { Ok(self.list_skills(¶ms).await?.into_page()) }
})
.await
}
pub async fn get_skill(&self, id: &str) -> Result<SkillResponse> {
self.call(ops::GET_SKILL, vec![("id", id.to_owned())]).await
}
pub async fn create_skill(&self, body: &CreateSkillRequest) -> Result<SkillResponse> {
self.with_body(ops::CREATE_SKILL, Vec::new(), body).await
}
pub async fn update_skill(&self, id: &str, body: &UpdateSkillRequest) -> Result<SkillResponse> {
self.with_body(ops::UPDATE_SKILL, vec![("id", id.to_owned())], body)
.await
}
pub async fn delete_skill(&self, id: &str) -> Result<()> {
self.call(ops::DELETE_SKILL, vec![("id", id.to_owned())])
.await
}
pub async fn duplicate_skill(&self, id: &str) -> Result<SkillResponse> {
self.call(ops::DUPLICATE_SKILL, vec![("id", id.to_owned())])
.await
}
pub async fn list_skill_versions(&self, id: &str) -> Result<SkillVersionsResponse> {
self.call(ops::LIST_SKILL_VERSIONS, vec![("id", id.to_owned())])
.await
}
pub async fn publish_skill(
&self,
id: &str,
body: &PublishSkillRequest,
) -> Result<SkillVersionResponse> {
self.with_body(ops::PUBLISH_SKILL, vec![("id", id.to_owned())], body)
.await
}
pub async fn generate_skill(&self, body: &GenerateSkillRequest) -> Result<SkillResponse> {
self.with_body(ops::GENERATE_SKILL, Vec::new(), body).await
}
pub async fn list_cassettes(&self) -> Result<Discovery> {
self.call(ops::LIST_CASSETTES, Vec::new()).await
}
pub async fn seed_demo(&self, body: &SeedDemoRequest) -> Result<SeedResult> {
self.with_body(ops::SEED_DEMO, Vec::new(), body).await
}
}
impl<T: StreamingTransport> CoreClient<T> {
pub async fn stream(&self, operation_id: &str, values: Vec<(&str, String)>) -> Result<T::Body> {
let method = core()?.method(operation_id)?;
let request = contract::call_for(method, values)?;
self.transport.send_stream(&request).await
}
pub async fn export_session(&self, id: &str, params: &ExportSessionParams) -> Result<T::Body> {
let mut values = params.values();
values.push(("id", id.to_owned()));
self.stream(ops::EXPORT_SESSION, values).await
}
pub async fn export_sessions(&self, params: &ExportSessionsParams) -> Result<T::Body> {
self.stream(ops::EXPORT_SESSIONS, params.values()).await
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
mod tests {
use super::*;
use crate::core::models::params::PayloadDetail;
use crate::path::{PathMode, call_url};
use crate::transport::{TransportError, WireResponse};
use serde::Deserialize;
use serde_json::Value;
use std::cell::RefCell;
use url::Url;
struct Recorder {
base: Url,
responses: RefCell<Vec<Value>>,
seen: RefCell<Vec<String>>,
bodies: RefCell<Vec<Option<String>>>,
}
impl Recorder {
fn new(base: &str, responses: Vec<Value>) -> Self {
Self {
base: Url::parse(base).unwrap(),
responses: RefCell::new(responses),
seen: RefCell::new(Vec::new()),
bodies: RefCell::new(Vec::new()),
}
}
}
impl TapesTransport for Recorder {
async fn send(
&self,
request: &WireRequest<'_>,
) -> std::result::Result<WireResponse, TransportError> {
let url = call_url(&self.base, request, PathMode::UnderBase)
.map_err(|error| TransportError::new(error.to_string()))?;
self.seen.borrow_mut().push(url.to_string());
self.bodies.borrow_mut().push(request.body.clone());
let mut responses = self.responses.borrow_mut();
let body = if responses.len() > 1 {
responses.remove(0)
} else {
responses.first().cloned().unwrap_or(Value::Null)
};
Ok(WireResponse::new(
200,
url.to_string(),
Vec::new(),
body.to_string().into_bytes(),
))
}
}
fn client(base: &str, response: Value) -> CoreClient<Recorder> {
CoreClient::new(Recorder::new(base, vec![response]))
}
#[tokio::test]
async fn an_operation_is_routed_through_the_contract_and_the_transport() {
let client = client(
"https://acme.example/primary/tapes/",
serde_json::json!({"traces": []}),
);
let _ = client
.get_session_traces("s-1", &SessionTracesParams::default())
.await
.unwrap();
assert_eq!(
client.transport().seen.borrow()[0],
"https://acme.example/primary/tapes/v1/sessions/s-1/traces",
);
}
#[tokio::test]
async fn a_typed_method_decodes_the_contracts_own_shape() {
let client = client(
"http://127.0.0.1:8081",
serde_json::json!({
"items": [{"id": "s1", "rollup": {"turn_count": 3}}],
"next_cursor": "abc",
}),
);
let listing = client
.list_sessions(&SessionListParams::default())
.await
.unwrap();
assert_eq!(listing.items[0].id, "s1");
assert_eq!(listing.items[0].rollup.turn_count, 3);
assert_eq!(listing.next_cursor, "abc");
}
#[tokio::test]
async fn a_typed_method_survives_a_field_it_has_never_heard_of() {
let client = client(
"http://127.0.0.1:8081",
serde_json::json!({"items": [{"id": "s1", "a_field_from_the_future": 7}]}),
);
let listing = client
.list_sessions(&SessionListParams::default())
.await
.unwrap();
assert_eq!(listing.items[0].id, "s1");
}
#[tokio::test]
async fn the_generic_seam_still_decodes_into_a_callers_own_type() {
#[derive(Debug, Deserialize)]
struct Listing {
next_cursor: String,
}
let client = client(
"http://127.0.0.1:8081",
serde_json::json!({"items": [], "next_cursor": "abc"}),
);
let got: Listing = client.call(ops::LIST_SESSIONS, Vec::new()).await.unwrap();
assert_eq!(got.next_cursor, "abc");
let raw: Value = client.call(ops::LIST_SESSIONS, Vec::new()).await.unwrap();
assert_eq!(raw["next_cursor"], "abc");
}
#[tokio::test]
async fn a_typed_parameter_travels_under_the_contracts_own_name() {
let client = client("http://127.0.0.1:8081", serde_json::json!({"traces": []}));
let _ = client
.get_session_traces(
"s-1",
&SessionTracesParams {
payload: Some(PayloadDetail::Preview),
},
)
.await
.unwrap();
assert!(
client.transport().seen.borrow()[0].ends_with("/traces?payload=preview"),
"got: {:?}",
client.transport().seen.borrow(),
);
}
#[tokio::test]
async fn a_listing_walk_follows_the_cursor_to_the_end() {
let client = CoreClient::new(Recorder::new(
"http://127.0.0.1:8081",
vec![
serde_json::json!({"items": [{"id": "s1"}], "next_cursor": "c1"}),
serde_json::json!({"items": [{"id": "s2"}], "next_cursor": ""}),
],
));
let sessions = client
.list_all_sessions(&SessionListParams::default())
.await
.unwrap();
assert_eq!(
sessions.iter().map(|s| s.id.as_str()).collect::<Vec<_>>(),
vec!["s1", "s2"],
);
assert!(
client.transport().seen.borrow()[1].contains("cursor=c1"),
"got: {:?}",
client.transport().seen.borrow(),
);
}
#[tokio::test]
async fn an_undeclared_parameter_is_refused_before_the_transport_is_reached() {
let client = client("http://127.0.0.1:8081", Value::Null);
let err = client
.call::<Value>(ops::GET_SESSION, vec![("payolad", "full".to_owned())])
.await
.unwrap_err();
assert!(err.to_string().contains("payolad"), "got: {err}");
assert!(
client.transport().seen.borrow().is_empty(),
"nothing may be sent for a call the contract refused",
);
}
#[tokio::test]
async fn a_typed_body_reaches_the_transport_as_the_contracts_own_json() {
let client = client("http://127.0.0.1:8081", serde_json::json!({"id": "sk-1"}));
let skill = client
.create_skill(&CreateSkillRequest {
name: "gum".to_owned(),
..Default::default()
})
.await
.unwrap();
assert_eq!(skill.id, "sk-1");
let bodies = client.transport().bodies.borrow();
let sent: Value = serde_json::from_str(bodies[0].as_deref().unwrap()).unwrap();
assert_eq!(sent["name"], "gum");
}
#[tokio::test]
async fn the_bodyless_facade_refuses_an_operation_that_requires_a_body() {
let client = client("http://127.0.0.1:8081", Value::Null);
let err = client
.call::<Value>(ops::CREATE_SKILL, Vec::new())
.await
.unwrap_err();
assert!(
err.to_string().contains("requires a request body"),
"got: {err}",
);
assert!(client.transport().seen.borrow().is_empty());
}
#[tokio::test]
async fn the_facade_refuses_a_body_on_an_operation_that_declares_none() {
let client = client("http://127.0.0.1:8081", Value::Null);
let err = client
.call_with_body::<Value>(
ops::GET_SESSION,
vec![("id", "s-1".to_owned())],
Some("{}".to_owned()),
)
.await
.unwrap_err();
assert!(
err.to_string().contains("declares no request body"),
"got: {err}",
);
assert!(client.transport().seen.borrow().is_empty());
}
#[tokio::test]
async fn the_bodyless_facade_still_sends_no_body_for_an_ordinary_read() {
let client = client("http://127.0.0.1:8081", serde_json::json!({"items": []}));
let _ = client
.list_sessions(&SessionListParams::default())
.await
.unwrap();
assert_eq!(client.transport().bodies.borrow().as_slice(), [None]);
}
#[tokio::test]
async fn a_named_method_and_its_operation_id_build_the_same_request() {
let named = client("http://127.0.0.1:8081", serde_json::json!({}));
let _ = named.get_span("t-1", "sp-1").await.unwrap();
let raw = client("http://127.0.0.1:8081", serde_json::json!({}));
let _: Value = raw
.call(
ops::GET_SPAN,
vec![
("trace_id", "t-1".to_owned()),
("span_id", "sp-1".to_owned()),
],
)
.await
.unwrap();
assert_eq!(
*named.transport().seen.borrow(),
*raw.transport().seen.borrow()
);
}
}