use std::{
collections::BTreeMap,
sync::Arc,
time::{Duration, SystemTime},
};
use odp_core::{
Collection, CollectionSearchRequest, Offering, OfferingPage, OfferingSearchRequest, Operation,
Page, ParseError, Representation, ServiceDocument, build_operation_url, derive_service_origin,
normalize_agent_response, parse_agent_service_document, parse_collection, parse_offering,
parse_offering_search_response, parse_page, parse_problem_response, resolve_continuation,
};
use odp_directory::{HttpRequest, ReqwestTransport, Transport, TransportError};
use sha2::{Digest, Sha256};
use thiserror::Error;
use url::Url;
use crate::{Cache, CacheFallbacks, CacheRecord, default_cache};
const MEDIA_TYPE: &str = "application/odp+json";
const MAX_DOCUMENT_BYTES: usize = 65_536;
const MAX_RESOURCE_BYTES: usize = 524_288;
const MAX_REDIRECTS: usize = 5;
const MAX_TRAVERSAL_ITEMS: usize = 10_000;
const MAX_TRAVERSAL_PAGES: usize = 16;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TraversalOptions {
pub max_items: usize,
pub max_pages: usize,
}
impl Default for TraversalOptions {
fn default() -> Self {
Self {
max_items: MAX_TRAVERSAL_ITEMS,
max_pages: MAX_TRAVERSAL_PAGES,
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct Inspection {
pub document: ServiceDocument,
pub final_url: String,
pub freshness: Freshness,
pub requested_url: String,
pub service_origin: String,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum Freshness {
Fetched,
Fresh,
Revalidated,
}
#[derive(Debug, Error)]
pub enum AgentError {
#[error(transparent)]
Transport(#[from] TransportError),
#[error(transparent)]
Parse(#[from] ParseError),
#[error("invalid Agent request: {0}")]
InvalidRequest(String),
#[error("invalid ODP response: {0}")]
InvalidResponse(String),
#[error("ODP cache failed: {0}")]
Cache(String),
#[error("ODP Directory failed: {0}")]
Directory(String),
#[error("ODP Service does not advertise {0:?}")]
UnsupportedOperation(Operation),
#[error("ODP request failed with HTTP {status}: {message}")]
Request { message: String, status: u16 },
}
#[derive(Clone)]
pub struct ServiceClient {
accept_language: Option<String>,
cache: Arc<dyn Cache>,
cache_fallbacks: CacheFallbacks,
cache_partition: String,
service_origin: String,
supporting_transport: Arc<dyn Transport>,
transport: Arc<dyn Transport>,
}
impl ServiceClient {
pub fn new(service_url: &str) -> Result<Self, AgentError> {
Self::with_transport(service_url, Arc::new(ReqwestTransport::new()?))
}
pub fn with_transport(
service_url: &str,
transport: Arc<dyn Transport>,
) -> Result<Self, AgentError> {
let supporting_transport = Arc::new(ReqwestTransport::new()?);
Ok(Self {
accept_language: None,
cache: default_cache(),
cache_fallbacks: CacheFallbacks::default(),
cache_partition: "anonymous".to_owned(),
service_origin: derive_service_origin(service_url)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?,
supporting_transport,
transport,
})
}
pub fn with_accept_language(mut self, language: impl Into<String>) -> Self {
self.accept_language = Some(language.into());
self
}
pub fn with_cache(mut self, cache: Arc<dyn Cache>) -> Self {
self.cache = cache;
self
}
pub fn with_cache_fallbacks(mut self, fallbacks: CacheFallbacks) -> Self {
self.cache_fallbacks = fallbacks;
self
}
pub fn with_cache_partition(mut self, partition: impl Into<String>) -> Self {
self.cache_partition = partition.into();
self
}
pub fn with_supporting_transport(mut self, transport: Arc<dyn Transport>) -> Self {
self.supporting_transport = transport;
self
}
pub async fn inspect(&self) -> Result<Inspection, AgentError> {
let requested_url = format!("{}/.well-known/odp", self.service_origin);
let target = Url::parse(&requested_url)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?;
let response = self
.request_cached(
"GET",
target,
Vec::new(),
MAX_DOCUMENT_BYTES,
self.cache_fallbacks.service_document,
validate_service_document_bytes,
)
.await?;
let document = parse_agent_service_document(&response.body)?;
Ok(Inspection {
document,
final_url: response.final_url,
freshness: response.freshness,
requested_url,
service_origin: self.service_origin.clone(),
})
}
pub async fn list_collections(
&self,
representation: Representation,
limit: usize,
) -> Result<Page<Collection>, AgentError> {
let page = self
.get_page(Operation::ListCollections, None, representation, limit)
.await?;
validate_collections(page)
}
pub async fn get_collection(&self, id: &str) -> Result<Collection, AgentError> {
let data = self
.get_resource(Operation::GetCollection, id, Representation::Full)
.await?;
Ok(parse_agent_collection(&data)?)
}
pub async fn search_collections(
&self,
request: &CollectionSearchRequest,
representation: Representation,
) -> Result<Page<Collection>, AgentError> {
let data = self
.post_search(
Operation::SearchCollections,
serde_json::to_vec(request)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?,
representation,
)
.await?;
validate_collections(parse_agent_collection_page(&data)?)
}
pub async fn list_offerings(
&self,
representation: Representation,
limit: usize,
) -> Result<OfferingPage<Offering>, AgentError> {
self.get_offering_page(Operation::ListOfferings, None, representation, limit)
.await
}
pub async fn list_collection_offerings(
&self,
collection_id: &str,
representation: Representation,
limit: usize,
) -> Result<OfferingPage<Offering>, AgentError> {
self.get_offering_page(
Operation::ListCollectionOfferings,
Some(collection_id),
representation,
limit,
)
.await
}
pub async fn get_offering(&self, id: &str) -> Result<Offering, AgentError> {
let data = self
.get_resource(Operation::GetOffering, id, Representation::Full)
.await?;
Ok(parse_agent_offering(&data)?)
}
pub async fn search_offerings(
&self,
request: &OfferingSearchRequest,
representation: Representation,
) -> Result<OfferingPage<Offering>, AgentError> {
let data = self
.post_search(
Operation::SearchOfferings,
serde_json::to_vec(request)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?,
representation,
)
.await?;
Ok(parse_agent_offering_search_response(&data)?)
}
pub async fn continue_collections(&self, next: &str) -> Result<Page<Collection>, AgentError> {
let target = resolve_continuation(next, &self.service_origin)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?;
let response = self
.request_cached(
"GET",
target,
Vec::new(),
MAX_RESOURCE_BYTES,
self.cache_fallbacks.collection,
validate_collection_page_bytes,
)
.await?;
validate_collections(parse_agent_collection_page(&response.body)?)
}
pub async fn continue_offerings(
&self,
next: &str,
) -> Result<OfferingPage<Offering>, AgentError> {
let target = resolve_continuation(next, &self.service_origin)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?;
let response = self
.request_cached(
"GET",
target,
Vec::new(),
MAX_RESOURCE_BYTES,
self.cache_fallbacks.offering,
validate_offering_page_bytes,
)
.await?;
Ok(parse_agent_offering_search_response(&response.body)?)
}
pub async fn list_all_collections(
&self,
representation: Representation,
limit: usize,
options: TraversalOptions,
) -> Result<Vec<Collection>, AgentError> {
let mut page = self.list_collections(representation, limit).await?;
self.collect_collections(&mut page, options).await
}
pub async fn list_all_offerings(
&self,
representation: Representation,
limit: usize,
options: TraversalOptions,
) -> Result<Vec<Offering>, AgentError> {
let mut page = self.list_offerings(representation, limit).await?;
self.collect_offerings(&mut page, options).await
}
pub async fn search_all_offerings(
&self,
request: &OfferingSearchRequest,
representation: Representation,
options: TraversalOptions,
) -> Result<Vec<Offering>, AgentError> {
let mut page = self.search_offerings(request, representation).await?;
self.collect_offerings(&mut page, options).await
}
async fn collect_collections(
&self,
page: &mut Page<Collection>,
options: TraversalOptions,
) -> Result<Vec<Collection>, AgentError> {
let (maximum_items, maximum_pages) = traversal_bounds(options)?;
let mut result = Vec::new();
for page_number in 0..maximum_pages {
result.extend(page.items.drain(..).take(maximum_items - result.len()));
if result.len() == maximum_items || page.next.is_empty() {
return Ok(result);
}
if page_number + 1 < maximum_pages {
*page = self.continue_collections(&page.next).await?;
}
}
Ok(result)
}
async fn collect_offerings(
&self,
page: &mut OfferingPage<Offering>,
options: TraversalOptions,
) -> Result<Vec<Offering>, AgentError> {
let (maximum_items, maximum_pages) = traversal_bounds(options)?;
let mut result = Vec::new();
for page_number in 0..maximum_pages {
result.extend(page.items.drain(..).take(maximum_items - result.len()));
if result.len() == maximum_items || page.next.is_empty() {
return Ok(result);
}
if page_number + 1 < maximum_pages {
*page = self.continue_offerings(&page.next).await?;
}
}
Ok(result)
}
async fn get_page<T: serde::de::DeserializeOwned>(
&self,
operation: Operation,
id: Option<&str>,
representation: Representation,
limit: usize,
) -> Result<Page<T>, AgentError> {
let data = self
.get_page_bytes(operation, id, representation, limit)
.await?;
let kind = if matches!(
operation,
Operation::ListCollections | Operation::SearchCollections
) {
"collection-page"
} else {
"offering-page"
};
Ok(parse_page(&normalize_agent_response(&data, kind)?)?)
}
async fn get_offering_page(
&self,
operation: Operation,
id: Option<&str>,
representation: Representation,
limit: usize,
) -> Result<OfferingPage<Offering>, AgentError> {
let data = self
.get_page_bytes(operation, id, representation, limit)
.await?;
Ok(parse_agent_offering_search_response(&data)?)
}
async fn get_page_bytes(
&self,
operation: Operation,
id: Option<&str>,
representation: Representation,
limit: usize,
) -> Result<Vec<u8>, AgentError> {
let inspection = self.require_operation(operation).await?;
let mut target = build_operation_url(
&inspection.document.http.endpoint_base,
operation,
&self.service_origin,
id,
)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?;
target
.query_pairs_mut()
.append_pair("representation", representation_name(representation));
if limit != 0 {
target
.query_pairs_mut()
.append_pair("limit", &limit.to_string());
}
let fallback = if matches!(
operation,
Operation::GetCollection | Operation::ListCollections | Operation::SearchCollections
) {
self.cache_fallbacks.collection
} else {
self.cache_fallbacks.offering
};
let validator = response_validator(operation);
Ok(self
.request_cached(
"GET",
target,
Vec::new(),
MAX_RESOURCE_BYTES,
fallback,
validator,
)
.await?
.body)
}
async fn get_resource(
&self,
operation: Operation,
id: &str,
representation: Representation,
) -> Result<Vec<u8>, AgentError> {
self.get_page_bytes(operation, Some(id), representation, 0)
.await
}
async fn post_search(
&self,
operation: Operation,
body: Vec<u8>,
representation: Representation,
) -> Result<Vec<u8>, AgentError> {
let inspection = self.require_operation(operation).await?;
let mut target = build_operation_url(
&inspection.document.http.endpoint_base,
operation,
&self.service_origin,
None,
)
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?;
target
.query_pairs_mut()
.append_pair("representation", representation_name(representation));
let fallback = if operation == Operation::SearchCollections {
self.cache_fallbacks.collection
} else {
self.cache_fallbacks.offering
};
let validator = response_validator(operation);
Ok(self
.request_cached(
"POST",
target,
body,
MAX_RESOURCE_BYTES,
fallback,
validator,
)
.await?
.body)
}
async fn require_operation(&self, operation: Operation) -> Result<Inspection, AgentError> {
let inspection = self.inspect().await?;
if inspection
.document
.operations
.iter()
.any(|descriptor| descriptor.name == operation)
{
Ok(inspection)
} else {
Err(AgentError::UnsupportedOperation(operation))
}
}
async fn request_cached(
&self,
method: &str,
target: Url,
body: Vec<u8>,
maximum_bytes: usize,
fallback: Duration,
validate: fn(&[u8]) -> Result<(), AgentError>,
) -> Result<Response, AgentError> {
let key = self.cache_key(method, target.as_str(), &body);
let request_origin = derive_service_origin(target.as_str())
.map_err(|error| AgentError::InvalidRequest(error.to_string()))?;
let cached = self.cache.get(&key).map_err(AgentError::Cache)?;
let now = SystemTime::now();
if let Some(record) = &cached {
if now < record.expires_at {
return Ok(Response {
body: record.body.clone(),
final_url: record.final_url.clone(),
freshness: Freshness::Fresh,
});
}
}
let mut conditional = BTreeMap::new();
let mut request_target = target;
if let Some(record) = &cached {
if let Ok(cached_target) = Url::parse(&record.final_url) {
if derive_service_origin(cached_target.as_str())
.ok()
.as_deref()
== Some(&request_origin)
{
request_target = cached_target;
}
}
if let Some(etag) = &record.etag {
conditional.insert("if-none-match".to_owned(), etag.clone());
}
if let Some(last_modified) = &record.last_modified {
conditional.insert("if-modified-since".to_owned(), last_modified.clone());
}
}
let raw = self
.request_raw(method, request_target, body, conditional, &request_origin)
.await?;
if raw.status == 304 {
let Some(mut record) = cached else {
return Err(AgentError::InvalidResponse(
"ODP response returned 304 without a cached representation".to_owned(),
));
};
if no_store(&raw.headers) {
self.cache.delete(&key).map_err(AgentError::Cache)?;
return Ok(Response {
body: record.body,
final_url: record.final_url,
freshness: Freshness::Revalidated,
});
}
record.expires_at = revalidated_expiration(&raw.headers, &record, fallback, now);
record.stored_at = now;
record.final_url = raw.final_url;
self.cache
.set(key, record.clone())
.map_err(AgentError::Cache)?;
return Ok(Response {
body: record.body,
final_url: record.final_url,
freshness: Freshness::Revalidated,
});
}
let response = consume(raw, maximum_bytes)?;
validate(&response.body)?;
if !cacheable(method, &response.headers, fallback) {
self.cache.delete(&key).map_err(AgentError::Cache)?;
} else {
self.cache
.set(
key,
CacheRecord {
body: response.body.clone(),
etag: response.headers.get("etag").cloned(),
expires_at: expiration(&response.headers, fallback, now),
final_url: response.final_url.clone(),
last_modified: response.headers.get("last-modified").cloned(),
status: response.status,
stored_at: now,
},
)
.map_err(AgentError::Cache)?;
}
Ok(Response {
body: response.body,
final_url: response.final_url,
freshness: Freshness::Fetched,
})
}
async fn request_raw(
&self,
mut method: &str,
mut target: Url,
mut body: Vec<u8>,
conditional: BTreeMap<String, String>,
redirect_origin: &str,
) -> Result<RawResponse, AgentError> {
for redirect in 0..=MAX_REDIRECTS {
let mut headers = BTreeMap::from([("accept".to_owned(), MEDIA_TYPE.to_owned())]);
if let Some(language) = &self.accept_language {
headers.insert("accept-language".to_owned(), language.clone());
}
if !body.is_empty() {
headers.insert("content-type".to_owned(), MEDIA_TYPE.to_owned());
}
headers.extend(conditional.clone());
let response = self
.transport
.send(HttpRequest {
body: body.clone(),
headers,
method: method.to_owned(),
url: target.to_string(),
})
.await?;
if matches!(response.status, 301 | 302 | 303 | 307 | 308) {
if redirect == MAX_REDIRECTS {
return Err(AgentError::InvalidResponse(
"ODP response exceeded five redirects".to_owned(),
));
}
let location = response.headers.get("location").ok_or_else(|| {
AgentError::InvalidResponse("ODP redirect omitted Location".to_owned())
})?;
let next = target
.join(location)
.map_err(|error| AgentError::InvalidResponse(error.to_string()))?;
if derive_service_origin(next.as_str()).ok().as_deref() != Some(redirect_origin) {
return Err(AgentError::InvalidResponse(
"ODP redirect changed Service origin".to_owned(),
));
}
if response.status == 303
|| (matches!(response.status, 301 | 302) && method == "POST")
{
method = "GET";
body.clear();
}
target = next;
continue;
}
return Ok(RawResponse {
body: response.body,
final_url: target.to_string(),
headers: response.headers,
status: response.status,
});
}
Err(AgentError::InvalidResponse(
"ODP response exceeded its redirect limit".to_owned(),
))
}
fn cache_key(&self, method: &str, target: &str, body: &[u8]) -> String {
format!(
"{}\n{}\n{}\n{}\n{}",
self.cache_partition,
method,
target,
self.accept_language.as_deref().unwrap_or_default(),
sha256_hex(body)
)
}
pub(crate) fn service_origin(&self) -> &str {
&self.service_origin
}
pub(crate) async fn linked_odp(
&self,
target: Url,
fallback: Duration,
validate: fn(&[u8]) -> Result<(), AgentError>,
) -> Result<Vec<u8>, AgentError> {
Ok(self
.request_cached(
"GET",
target,
Vec::new(),
MAX_RESOURCE_BYTES,
fallback,
validate,
)
.await?
.body)
}
pub(crate) async fn supporting_json(
&self,
target: &str,
resource_class: &str,
accept: &str,
media_types: &[&str],
maximum_bytes: usize,
) -> Result<serde_json::Value, AgentError> {
let mut current =
Url::parse(target).map_err(|error| AgentError::InvalidRequest(error.to_string()))?;
if current.scheme() != "https" || current.host_str().is_none() {
return Err(AgentError::InvalidRequest(
"ODP supporting document URL must use HTTPS".to_owned(),
));
}
let key = format!("anonymous:{resource_class}\nGET\n{target}\n{accept}");
let cached = self.cache.get(&key).map_err(AgentError::Cache)?;
let now = SystemTime::now();
if let Some(record) = &cached {
if now < record.expires_at {
return decode_json_object(&record.body);
}
}
for redirects in 0..=MAX_REDIRECTS {
let mut headers = BTreeMap::from([("accept".to_owned(), accept.to_owned())]);
if let Some(record) = &cached {
if let Some(etag) = &record.etag {
headers.insert("if-none-match".to_owned(), etag.clone());
}
if let Some(last_modified) = &record.last_modified {
headers.insert("if-modified-since".to_owned(), last_modified.clone());
}
}
let response = self
.supporting_transport
.send(HttpRequest {
body: Vec::new(),
headers,
method: "GET".to_owned(),
url: current.to_string(),
})
.await?;
if matches!(response.status, 301 | 302 | 303 | 307 | 308) {
if redirects == MAX_REDIRECTS {
return Err(AgentError::InvalidResponse(
"ODP supporting document exceeded five redirects".to_owned(),
));
}
let location = response.headers.get("location").ok_or_else(|| {
AgentError::InvalidResponse(
"ODP supporting document redirect omitted Location".to_owned(),
)
})?;
let next = current
.join(location)
.map_err(|error| AgentError::InvalidResponse(error.to_string()))?;
if next.scheme() != "https" || next.host_str().is_none() {
return Err(AgentError::InvalidResponse(
"ODP supporting document redirect must use HTTPS".to_owned(),
));
}
current = next;
continue;
}
if response.status == 304 {
let Some(mut record) = cached.clone() else {
return Err(AgentError::InvalidResponse(
"ODP supporting document returned 304 without a cached representation"
.to_owned(),
));
};
if no_store(&response.headers) {
self.cache.delete(&key).map_err(AgentError::Cache)?;
} else {
record.expires_at =
revalidated_expiration(&response.headers, &record, Duration::ZERO, now);
record.stored_at = now;
record.final_url = current.to_string();
self.cache
.set(key.clone(), record.clone())
.map_err(AgentError::Cache)?;
}
return decode_json_object(&record.body);
}
if !(200..300).contains(&response.status) {
return Err(AgentError::Request {
message: format!("ODP supporting document returned HTTP {}", response.status),
status: response.status,
});
}
if response.body.len() > maximum_bytes {
return Err(AgentError::InvalidResponse(
"ODP supporting document exceeds its byte limit".to_owned(),
));
}
let content_type = response
.headers
.get("content-type")
.map(|value| value.split(';').next().unwrap_or_default().trim())
.unwrap_or_default();
if !media_types
.iter()
.any(|value| content_type.eq_ignore_ascii_case(value))
{
return Err(AgentError::InvalidResponse(
"ODP supporting document has an unsupported media type".to_owned(),
));
}
let document = decode_json_object(&response.body)?;
if !cacheable("GET", &response.headers, Duration::ZERO) {
self.cache.delete(&key).map_err(AgentError::Cache)?;
} else {
self.cache
.set(
key.clone(),
CacheRecord {
body: response.body,
etag: response.headers.get("etag").cloned(),
expires_at: expiration(&response.headers, Duration::ZERO, now),
final_url: current.to_string(),
last_modified: response.headers.get("last-modified").cloned(),
status: response.status,
stored_at: now,
},
)
.map_err(AgentError::Cache)?;
}
return Ok(document);
}
Err(AgentError::InvalidResponse(
"ODP supporting document exceeded its redirect limit".to_owned(),
))
}
}
fn sha256_hex(data: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let digest = Sha256::digest(data);
let mut encoded = String::with_capacity(digest.len() * 2);
for byte in digest {
encoded.push(HEX[usize::from(byte >> 4)] as char);
encoded.push(HEX[usize::from(byte & 0x0f)] as char);
}
encoded
}
fn parse_agent_collection(data: &[u8]) -> Result<Collection, ParseError> {
parse_collection(&normalize_agent_response(data, "collection")?)
}
fn parse_agent_offering(data: &[u8]) -> Result<Offering, ParseError> {
parse_offering(&normalize_agent_response(data, "offering")?)
}
fn parse_agent_collection_page(data: &[u8]) -> Result<Page<Collection>, ParseError> {
parse_page(&normalize_agent_response(data, "collection-page")?)
}
fn parse_agent_offering_search_response(data: &[u8]) -> Result<OfferingPage<Offering>, ParseError> {
parse_offering_search_response(&normalize_agent_response(data, "offering-page")?)
}
fn validate_collections(page: Page<Collection>) -> Result<Page<Collection>, AgentError> {
for collection in &page.items {
let mut inherited = collection.clone();
if inherited.odp_version.is_empty() {
inherited.odp_version.clone_from(&page.odp_version);
}
let data = serde_json::to_vec(&inherited)
.map_err(|error| AgentError::InvalidResponse(error.to_string()))?;
parse_agent_collection(&data)?;
}
Ok(page)
}
fn validate_offerings(page: OfferingPage<Offering>) -> Result<OfferingPage<Offering>, AgentError> {
for offering in &page.items {
let mut inherited = offering.clone();
if inherited.odp_version.is_empty() {
inherited.odp_version.clone_from(&page.odp_version);
}
let data = serde_json::to_vec(&inherited)
.map_err(|error| AgentError::InvalidResponse(error.to_string()))?;
parse_agent_offering(&data)?;
}
Ok(page)
}
fn validate_service_document_bytes(data: &[u8]) -> Result<(), AgentError> {
parse_agent_service_document(data)?;
Ok(())
}
fn validate_collection_bytes(data: &[u8]) -> Result<(), AgentError> {
parse_agent_collection(data)?;
Ok(())
}
fn validate_offering_bytes(data: &[u8]) -> Result<(), AgentError> {
parse_agent_offering(data)?;
Ok(())
}
fn validate_collection_page_bytes(data: &[u8]) -> Result<(), AgentError> {
let page = parse_agent_collection_page(data)?;
validate_collections(page)?;
Ok(())
}
fn validate_offering_page_bytes(data: &[u8]) -> Result<(), AgentError> {
validate_offerings(parse_agent_offering_search_response(data)?)?;
Ok(())
}
fn response_validator(operation: Operation) -> fn(&[u8]) -> Result<(), AgentError> {
match operation {
Operation::GetCollection => validate_collection_bytes,
Operation::GetOffering => validate_offering_bytes,
Operation::ListCollections | Operation::SearchCollections => validate_collection_page_bytes,
Operation::ListCollectionOfferings
| Operation::ListOfferings
| Operation::SearchOfferings => validate_offering_page_bytes,
}
}
fn traversal_bounds(options: TraversalOptions) -> Result<(usize, usize), AgentError> {
let maximum_items = if options.max_items == 0 {
MAX_TRAVERSAL_ITEMS
} else {
options.max_items
};
let maximum_pages = if options.max_pages == 0 {
MAX_TRAVERSAL_PAGES
} else {
options.max_pages
};
if maximum_items > MAX_TRAVERSAL_ITEMS || maximum_pages > MAX_TRAVERSAL_PAGES {
return Err(AgentError::InvalidRequest(
"traversal exceeds 10000 items or 16 pages".to_owned(),
));
}
Ok((maximum_items, maximum_pages))
}
struct Response {
body: Vec<u8>,
final_url: String,
freshness: Freshness,
}
struct RawResponse {
body: Vec<u8>,
final_url: String,
headers: BTreeMap<String, String>,
status: u16,
}
fn consume(response: RawResponse, maximum_bytes: usize) -> Result<RawResponse, AgentError> {
if response.body.len() > maximum_bytes {
return Err(AgentError::InvalidResponse(
"ODP response exceeds its byte limit".to_owned(),
));
}
if !(200..300).contains(&response.status) {
let problem = normalize_agent_response(&response.body, "problem")
.unwrap_or_else(|_| response.body.clone());
let message = parse_problem_response(&problem, response.status)
.map(|problem| {
if problem.detail.is_empty() {
problem.title
} else {
problem.detail
}
})
.unwrap_or_else(|_| String::from_utf8_lossy(&response.body).into_owned());
return Err(AgentError::Request {
message,
status: response.status,
});
}
let content_type = response
.headers
.get("content-type")
.map(|value| value.split(';').next().unwrap_or_default().trim())
.unwrap_or_default();
if !content_type.eq_ignore_ascii_case(MEDIA_TYPE) {
return Err(AgentError::InvalidResponse(format!(
"ODP response must use {MEDIA_TYPE}"
)));
}
Ok(response)
}
fn expiration(
headers: &BTreeMap<String, String>,
fallback: Duration,
now: SystemTime,
) -> SystemTime {
let directives = cache_directives(headers);
if directives.contains_key("no-cache") {
return now;
}
let maximum_age = directives
.get("max-age")
.and_then(|value| value.parse::<u64>().ok())
.map(Duration::from_secs);
if let Some(mut duration) = maximum_age {
if let Some(age) = headers
.get("age")
.and_then(|value| value.trim().parse::<u64>().ok())
{
duration = duration.saturating_sub(Duration::from_secs(age));
}
return now.checked_add(duration).unwrap_or(now);
}
if let Some(expires) = headers
.get("expires")
.and_then(|value| httpdate::parse_http_date(value).ok())
{
return expires;
}
now.checked_add(fallback).unwrap_or(now)
}
fn revalidated_expiration(
headers: &BTreeMap<String, String>,
record: &CacheRecord,
fallback: Duration,
now: SystemTime,
) -> SystemTime {
if has_freshness(headers) {
expiration(headers, fallback, now)
} else {
let lifetime = record
.expires_at
.duration_since(record.stored_at)
.unwrap_or(Duration::ZERO);
now.checked_add(lifetime).unwrap_or(now)
}
}
fn no_store(headers: &BTreeMap<String, String>) -> bool {
cache_directives(headers).contains_key("no-store")
}
fn cacheable(method: &str, headers: &BTreeMap<String, String>, fallback: Duration) -> bool {
if !matches!(method, "GET" | "POST") || !supported_vary(headers) || no_store(headers) {
return false;
}
let directives = cache_directives(headers);
let no_cache = directives.contains_key("no-cache");
(method == "GET" && (!fallback.is_zero() || no_cache)) || explicit_freshness(headers)
}
fn supported_vary(headers: &BTreeMap<String, String>) -> bool {
headers.get("vary").is_none_or(|value| {
value.split(',').all(|name| {
matches!(
name.trim().to_ascii_lowercase().as_str(),
"" | "accept" | "accept-language" | "content-type"
)
})
})
}
fn explicit_freshness(headers: &BTreeMap<String, String>) -> bool {
cache_directives(headers).contains_key("max-age") || headers.contains_key("expires")
}
fn has_freshness(headers: &BTreeMap<String, String>) -> bool {
let directives = cache_directives(headers);
directives.contains_key("max-age")
|| directives.contains_key("no-cache")
|| directives.contains_key("no-store")
|| headers.contains_key("expires")
}
fn cache_directives(headers: &BTreeMap<String, String>) -> BTreeMap<String, String> {
headers
.get("cache-control")
.into_iter()
.flat_map(|value| value.split(','))
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| {
let (name, setting) = value.split_once('=').unwrap_or((value, ""));
(
name.to_ascii_lowercase(),
setting.trim_matches('"').to_owned(),
)
})
.collect()
}
fn decode_json_object(data: &[u8]) -> Result<serde_json::Value, AgentError> {
let value = serde_json::from_slice::<serde_json::Value>(data)
.map_err(|error| AgentError::InvalidResponse(error.to_string()))?;
if !value.is_object() {
return Err(AgentError::InvalidResponse(
"ODP supporting document must be a JSON object".to_owned(),
));
}
Ok(value)
}
const fn representation_name(value: Representation) -> &'static str {
match value {
Representation::Terse => "terse",
Representation::Full => "full",
}
}
#[cfg(test)]
mod tests {
use std::{
collections::VecDeque,
sync::{
Mutex,
atomic::{AtomicUsize, Ordering},
},
};
use async_trait::async_trait;
use odp_directory::HttpResponse;
use super::*;
struct MockTransport {
responses: Mutex<VecDeque<HttpResponse>>,
}
#[async_trait]
impl Transport for MockTransport {
async fn send(&self, _request: HttpRequest) -> Result<HttpResponse, TransportError> {
Ok(self.responses.lock().unwrap().pop_front().unwrap())
}
}
fn response(body: &'static [u8]) -> HttpResponse {
HttpResponse {
body: body.to_vec(),
headers: BTreeMap::from([("content-type".to_owned(), MEDIA_TYPE.to_owned())]),
status: 200,
}
}
struct ConditionalTransport {
calls: AtomicUsize,
}
#[async_trait]
impl Transport for ConditionalTransport {
async fn send(&self, request: HttpRequest) -> Result<HttpResponse, TransportError> {
let call = self.calls.fetch_add(1, Ordering::SeqCst);
if call == 0 {
return Ok(HttpResponse {
body: br#"{"description":"Plants","http":{"endpoint_base":"/odp"},"language":"en","localizations":["en"],"name":"Indica Flowers","odp_version":"1.0","operations":[{"authentication":"not-required","name":"get-offering"},{"authentication":"not-required","name":"list-offerings"}]}"#.to_vec(),
headers: BTreeMap::from([
("cache-control".to_owned(), "max-age=0".to_owned()),
("content-type".to_owned(), MEDIA_TYPE.to_owned()),
("etag".to_owned(), "document-1".to_owned()),
]),
status: 200,
});
}
assert_eq!(
request.headers.get("if-none-match").map(String::as_str),
Some("document-1")
);
Ok(HttpResponse {
body: Vec::new(),
headers: BTreeMap::from([("cache-control".to_owned(), "max-age=60".to_owned())]),
status: 304,
})
}
}
struct InvalidTransport {
calls: AtomicUsize,
}
#[async_trait]
impl Transport for InvalidTransport {
async fn send(&self, _request: HttpRequest) -> Result<HttpResponse, TransportError> {
self.calls.fetch_add(1, Ordering::SeqCst);
Ok(response(br#"{}"#))
}
}
#[tokio::test]
async fn inspects_support_before_getting_an_offering() {
let document = br#"{"description":"Plants","http":{"endpoint_base":"/odp"},"language":"en","localizations":["en"],"name":"Indica Flowers","odp_version":"1.0","operations":[{"authentication":"not-required","name":"get-offering"},{"authentication":"not-required","name":"list-offerings"}]}"#;
let offering = br#"{"id":"plant-1","name":"Plant","odp_version":"1.0"}"#;
let client = ServiceClient::with_transport(
"https://demo.inflowpay.ai",
Arc::new(MockTransport {
responses: Mutex::new(VecDeque::from([response(document), response(offering)])),
}),
)
.unwrap();
assert_eq!(client.get_offering("plant-1").await.unwrap().name, "Plant");
}
#[tokio::test]
async fn inspection_preserves_tap_trust_advertisement() {
let document = br#"{"description":"Plants","http":{"endpoint_base":"/odp"},"language":"en","localizations":["en"],"name":"Indica Flowers","odp_version":"1.0","operations":[{"authentication":"not-required","name":"get-offering"},{"authentication":"not-required","name":"list-offerings"}],"protocols":{"payments":[{"authentication":"not-required","name":"future-payment"},{"authentication":"not-required","name":"mpp"}],"trust":[{"name":"future-trust"},{"name":"tap"}]}}"#;
let client = ServiceClient::with_transport(
"https://demo.inflowpay.ai",
Arc::new(MockTransport {
responses: Mutex::new(VecDeque::from([response(document)])),
}),
)
.unwrap();
let inspection = client.inspect().await.unwrap();
let protocols = inspection.document.protocols.unwrap();
assert_eq!(protocols.payments.len(), 1);
assert_eq!(
protocols.trust,
[odp_core::TrustProtocol {
name: odp_core::Protocol::Tap
}]
);
}
#[tokio::test]
async fn caches_the_service_document_with_its_resource_fallback() {
let document = br#"{"description":"Plants","http":{"endpoint_base":"/odp"},"language":"en","localizations":["en"],"name":"Indica Flowers","odp_version":"1.0","operations":[{"authentication":"not-required","name":"get-offering"},{"authentication":"not-required","name":"list-offerings"}]}"#;
let client = ServiceClient::with_transport(
"https://demo.inflowpay.ai",
Arc::new(MockTransport {
responses: Mutex::new(VecDeque::from([response(document)])),
}),
)
.unwrap();
assert_eq!(
client.inspect().await.unwrap().freshness,
Freshness::Fetched
);
assert_eq!(client.inspect().await.unwrap().freshness, Freshness::Fresh);
}
#[tokio::test]
async fn revalidates_a_stale_cached_document() {
let transport = Arc::new(ConditionalTransport {
calls: AtomicUsize::new(0),
});
let client =
ServiceClient::with_transport("https://demo.inflowpay.ai", transport.clone()).unwrap();
assert_eq!(
client.inspect().await.unwrap().freshness,
Freshness::Fetched
);
assert_eq!(
client.inspect().await.unwrap().freshness,
Freshness::Revalidated
);
assert_eq!(client.inspect().await.unwrap().freshness, Freshness::Fresh);
assert_eq!(transport.calls.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn does_not_cache_an_invalid_document() {
let transport = Arc::new(InvalidTransport {
calls: AtomicUsize::new(0),
});
let client =
ServiceClient::with_transport("https://demo.inflowpay.ai", transport.clone()).unwrap();
assert!(client.inspect().await.is_err());
assert!(client.inspect().await.is_err());
assert_eq!(transport.calls.load(Ordering::SeqCst), 2);
}
#[test]
fn post_search_requires_explicit_freshness_before_caching() {
let headers = BTreeMap::new();
assert!(!cacheable("POST", &headers, Duration::from_secs(300)));
let headers = BTreeMap::from([("cache-control".to_owned(), "max-age=30".to_owned())]);
assert!(cacheable("POST", &headers, Duration::from_secs(300)));
}
#[test]
fn validates_embedded_representations_with_the_page_version() {
assert!(
validate_collection_page_bytes(
br#"{"items":[{"id":"plants","name":"Plants"}],"odp_version":"1.0"}"#
)
.is_ok()
);
assert!(
validate_offering_page_bytes(
br#"{"items":[{"id":"plant","name":"Plant"}],"odp_version":"1.0"}"#
)
.is_ok()
);
assert!(
validate_offering_page_bytes(
br#"{"items":[{"id":"bad/id","name":"Plant"}],"odp_version":"1.0"}"#
)
.is_err()
);
}
}