use canton_core::{Error, Result};
use serde::{Deserialize, Serialize};
use crate::context::{ChoiceContext, ChoiceContextRequest, WireChoiceContext};
const DEFAULT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
#[derive(Clone, Debug)]
pub struct RegistryClient {
base: reqwest::Url,
http: reqwest::Client,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct RegistryInfo {
pub admin_id: String,
#[serde(default)]
pub supported_apis: std::collections::BTreeMap<String, i32>,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct Instrument {
pub id: String,
pub name: String,
pub symbol: String,
#[serde(default)]
pub total_supply: Option<String>,
#[serde(default)]
pub total_supply_as_of: Option<String>,
#[serde(default = "default_decimals")]
pub decimals: i32,
#[serde(default)]
pub supported_apis: std::collections::BTreeMap<String, i32>,
#[serde(default)]
pub paused: bool,
#[serde(default)]
pub pause_info: Option<PauseInfo>,
#[serde(default)]
pub account_input_fields_to_show: Option<Vec<String>>,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
#[non_exhaustive]
pub struct PauseInfo {
#[serde(default)]
pub reason: Option<String>,
#[serde(default)]
pub until: Option<String>,
}
fn default_decimals() -> i32 {
10
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ListInstrumentsResponse {
#[serde(default)]
instruments: Vec<Instrument>,
#[serde(default)]
next_page_token: Option<String>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize)]
#[serde(rename_all = "lowercase")]
#[non_exhaustive]
pub enum TransferKind {
#[serde(rename = "self")]
SelfTransfer,
Direct,
Offer,
#[serde(other)]
Unknown,
}
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct FactoryWithContext {
pub factory_id: String,
pub transfer_kind: Option<TransferKind>,
pub context: ChoiceContext,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct WireTransferFactory {
factory_id: String,
transfer_kind: TransferKind,
choice_context: WireChoiceContext,
}
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct WireFactory {
factory_id: String,
choice_context: WireChoiceContext,
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct GetFactoryRequest<'a> {
choice_arguments: &'a serde_json::Value,
#[serde(skip_serializing_if = "std::ops::Not::not")]
exclude_debug_fields: bool,
}
impl RegistryClient {
pub fn new(base_url: &str) -> Result<Self> {
let http = reqwest::Client::builder()
.timeout(DEFAULT_TIMEOUT)
.build()
.map_err(|e| Error::InvalidRequest(format!("cannot build an HTTP client: {e}")))?;
Self::with_http_client(base_url, http)
}
pub fn with_http_client(base_url: &str, http: reqwest::Client) -> Result<Self> {
let normalised = format!("{}/", base_url.trim_end_matches('/'));
let base = reqwest::Url::parse(&normalised).map_err(|e| {
Error::InvalidRequest(format!("`{base_url}` is not a registry URL: {e}"))
})?;
Ok(Self { base, http })
}
fn url(&self, segments: &[&str]) -> Result<reqwest::Url> {
let mut url = self.base.clone();
url.path_segments_mut()
.map_err(|()| Error::InvalidRequest("the registry URL cannot have a path".to_string()))?
.pop_if_empty()
.extend(segments);
Ok(url)
}
pub async fn info(&self) -> Result<RegistryInfo> {
self.get(self.url(&["registry", "metadata", "v1", "info"])?)
.await
}
pub async fn list_instruments(
&self,
page_size: Option<u32>,
page_token: Option<&str>,
) -> Result<(Vec<Instrument>, Option<String>)> {
let mut url = self.url(&["registry", "metadata", "v1", "instruments"])?;
{
let mut query = url.query_pairs_mut();
if let Some(size) = page_size {
query.append_pair("pageSize", &size.to_string());
}
if let Some(token) = page_token {
query.append_pair("pageToken", token);
}
}
let response: ListInstrumentsResponse = self.get(url).await?;
Ok((response.instruments, response.next_page_token))
}
pub async fn instrument(&self, instrument_id: &str) -> Result<Option<Instrument>> {
let url = self.url(&["registry", "metadata", "v1", "instruments", instrument_id])?;
let response = self
.http
.get(url.clone())
.send()
.await
.map_err(|e| connection(&e))?;
if response.status() == reqwest::StatusCode::NOT_FOUND {
let body = response.text().await.map_err(|e| connection(&e))?;
let trimmed = body.trim();
if trimmed.is_empty() || serde_json::from_str::<serde_json::Value>(trimmed).is_ok() {
return Ok(None);
}
tracing::warn!(%url, "a 404 that is not the registry's own — check the base URL");
return Err(Error::http_at(404, truncate(trimmed), url.as_str()));
}
Ok(Some(Self::decode(response, url.as_str()).await?))
}
pub async fn transfer_factory(
&self,
choice_arguments: &serde_json::Value,
) -> Result<FactoryWithContext> {
let url = self.url(&["registry", "transfer-instruction", "v1", "transfer-factory"])?;
let wire: WireTransferFactory = self.post_factory(url, choice_arguments).await?;
Ok(FactoryWithContext {
factory_id: wire.factory_id,
transfer_kind: Some(wire.transfer_kind),
context: ChoiceContext::from_wire(wire.choice_context)?,
})
}
pub async fn allocation_factory(
&self,
choice_arguments: &serde_json::Value,
) -> Result<FactoryWithContext> {
let url = self.url(&[
"registry",
"allocation-instruction",
"v1",
"allocation-factory",
])?;
let wire: WireFactory = self.post_factory(url, choice_arguments).await?;
Ok(FactoryWithContext {
factory_id: wire.factory_id,
transfer_kind: None,
context: ChoiceContext::from_wire(wire.choice_context)?,
})
}
pub async fn transfer_factory_v2(
&self,
choice_arguments: &serde_json::Value,
) -> Result<FactoryWithContext> {
let url = self.url(&["registry", "transfer-instruction", "v2", "transfer-factory"])?;
let wire: WireTransferFactory = self.post_factory(url, choice_arguments).await?;
Ok(FactoryWithContext {
factory_id: wire.factory_id,
transfer_kind: Some(wire.transfer_kind),
context: ChoiceContext::from_wire(wire.choice_context)?,
})
}
pub async fn allocation_factory_v2(
&self,
choice_arguments: &serde_json::Value,
) -> Result<FactoryWithContext> {
let url = self.url(&[
"registry",
"allocation-instruction",
"v2",
"allocation-factory",
])?;
let wire: WireFactory = self.post_factory(url, choice_arguments).await?;
Ok(FactoryWithContext {
factory_id: wire.factory_id,
transfer_kind: None,
context: ChoiceContext::from_wire(wire.choice_context)?,
})
}
pub async fn settlement_factory_v2(
&self,
choice_arguments: &serde_json::Value,
) -> Result<FactoryWithContext> {
let url = self.url(&["registry", "allocation", "v2", "settlement-factory"])?;
let wire: WireFactory = self.post_factory(url, choice_arguments).await?;
Ok(FactoryWithContext {
factory_id: wire.factory_id,
transfer_kind: None,
context: ChoiceContext::from_wire(wire.choice_context)?,
})
}
pub async fn transfer_instruction_context_v2(
&self,
transfer_instruction_id: &str,
choice: TransferInstructionChoice,
request: &ChoiceContextRequest,
) -> Result<ChoiceContext> {
self.choice_context(
self.url(&[
"registry",
"transfer-instruction",
"v2",
transfer_instruction_id,
"choice-contexts",
choice.path(),
])?,
request,
)
.await
}
pub async fn allocation_context_v2(
&self,
allocation_id: &str,
choice: AllocationChoice,
request: &ChoiceContextRequest,
) -> Result<ChoiceContext> {
if choice == AllocationChoice::ExecuteTransfer {
return Err(Error::InvalidRequest(
"V2 allocations are settled through the settlement factory, not an \
execute-transfer context — see `settlement_factory_v2`"
.to_string(),
));
}
self.choice_context(
self.url(&[
"registry",
"allocations",
"v2",
allocation_id,
"choice-contexts",
choice.path(),
])?,
request,
)
.await
}
pub async fn allocation_instruction_context_v2(
&self,
allocation_instruction_id: &str,
choice: AllocationInstructionChoice,
request: &ChoiceContextRequest,
) -> Result<ChoiceContext> {
self.choice_context(
self.url(&[
"registry",
"allocation-instruction",
"v2",
allocation_instruction_id,
"choice-contexts",
choice.path(),
])?,
request,
)
.await
}
pub async fn transfer_instruction_context(
&self,
transfer_instruction_id: &str,
choice: TransferInstructionChoice,
request: &ChoiceContextRequest,
) -> Result<ChoiceContext> {
self.choice_context(
self.url(&[
"registry",
"transfer-instruction",
"v1",
transfer_instruction_id,
"choice-contexts",
choice.path(),
])?,
request,
)
.await
}
pub async fn allocation_context(
&self,
allocation_id: &str,
choice: AllocationChoice,
request: &ChoiceContextRequest,
) -> Result<ChoiceContext> {
self.choice_context(
self.url(&[
"registry",
"allocations",
"v1",
allocation_id,
"choice-contexts",
choice.path(),
])?,
request,
)
.await
}
async fn post_factory<T: serde::de::DeserializeOwned>(
&self,
url: reqwest::Url,
choice_arguments: &serde_json::Value,
) -> Result<T> {
let response = self
.http
.post(url.clone())
.json(&GetFactoryRequest {
choice_arguments,
exclude_debug_fields: true,
})
.send()
.await
.map_err(|e| connection(&e))?;
Self::decode(response, url.as_str()).await
}
async fn choice_context(
&self,
url: reqwest::Url,
request: &ChoiceContextRequest,
) -> Result<ChoiceContext> {
let response = self
.http
.post(url.clone())
.json(request)
.send()
.await
.map_err(|e| connection(&e))?;
let wire: WireChoiceContext = Self::decode(response, url.as_str()).await?;
ChoiceContext::from_wire(wire)
}
async fn get<T: serde::de::DeserializeOwned>(&self, url: reqwest::Url) -> Result<T> {
let response = self
.http
.get(url.clone())
.send()
.await
.map_err(|e| connection(&e))?;
Self::decode(response, url.as_str()).await
}
async fn decode<T: serde::de::DeserializeOwned>(
response: reqwest::Response,
url: &str,
) -> Result<T> {
let status = response.status();
let body = response.text().await.map_err(|e| connection(&e))?;
if status == reqwest::StatusCode::CONFLICT {
let message = serde_json::from_str::<ErrorResponse>(&body)
.map(|e| e.error)
.unwrap_or(body);
return Err(Error::Connection(format!(
"{url}: the registry reports a contract mid-reassignment (409), retry shortly: {}",
truncate(&message)
)));
}
if !status.is_success() {
tracing::warn!(%url, status = status.as_u16(), "the registry answered with an error");
let message = serde_json::from_str::<ErrorResponse>(&body)
.map(|e| e.error)
.unwrap_or(body);
return Err(Error::http_at(status.as_u16(), truncate(&message), url));
}
serde_json::from_str(&body).map_err(|e| {
Error::UnexpectedResponse(format!(
"{url} returned a body this does not understand: {e}"
))
})
}
}
#[derive(Deserialize)]
struct ErrorResponse {
error: String,
}
fn connection(e: &reqwest::Error) -> Error {
let kind = if e.is_timeout() {
Transport::Timeout
} else if e.is_builder() {
Transport::Builder
} else if e.is_redirect() {
Transport::Redirect
} else if e.is_decode() {
Transport::Decode
} else {
Transport::Connect
};
classify_transport(kind, &canton_core::chain(e))
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Transport {
Timeout,
Builder,
Redirect,
Decode,
Connect,
}
fn classify_transport(kind: Transport, detail: &str) -> Error {
match kind {
Transport::Timeout => return Error::Timeout,
Transport::Builder | Transport::Redirect => {
return Error::InvalidRequest(format!(
"the registry request could not be made: {detail}"
));
}
Transport::Decode => {
return Error::UnexpectedResponse(format!(
"the registry's reply could not be read: {detail}"
));
}
Transport::Connect => {}
}
let lower = detail.to_ascii_lowercase();
if lower.contains("certificate") || lower.contains("tls handshake") {
return Error::InvalidRequest(format!(
"the registry's TLS certificate could not be verified: {detail}"
));
}
Error::Connection(format!("cannot reach the registry: {detail}"))
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum TransferInstructionChoice {
Accept,
Reject,
Withdraw,
}
impl TransferInstructionChoice {
fn path(self) -> &'static str {
match self {
Self::Accept => "accept",
Self::Reject => "reject",
Self::Withdraw => "withdraw",
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum AllocationInstructionChoice {
Accept,
Withdraw,
}
impl AllocationInstructionChoice {
fn path(self) -> &'static str {
match self {
Self::Accept => "accept",
Self::Withdraw => "withdraw",
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum AllocationChoice {
ExecuteTransfer,
Withdraw,
Cancel,
}
impl AllocationChoice {
fn path(self) -> &'static str {
match self {
Self::ExecuteTransfer => "execute-transfer",
Self::Withdraw => "withdraw",
Self::Cancel => "cancel",
}
}
}
fn truncate(body: &str) -> String {
const LIMIT: usize = 200;
if body.len() <= LIMIT {
return body.to_string();
}
let cut = body
.char_indices()
.map(|(i, _)| i)
.take_while(|i| *i <= LIMIT)
.last()
.unwrap_or(0);
format!("{}…", &body[..cut])
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
#[test]
fn the_choice_paths_are_the_ones_the_standard_publishes() {
assert_eq!(TransferInstructionChoice::Accept.path(), "accept");
assert_eq!(TransferInstructionChoice::Reject.path(), "reject");
assert_eq!(TransferInstructionChoice::Withdraw.path(), "withdraw");
assert_eq!(AllocationChoice::ExecuteTransfer.path(), "execute-transfer");
assert_eq!(AllocationChoice::Withdraw.path(), "withdraw");
assert_eq!(AllocationChoice::Cancel.path(), "cancel");
}
#[test]
fn the_transfer_kinds_decode_from_the_words_the_wire_uses() {
let kinds: Vec<TransferKind> =
serde_json::from_str(r#"["self", "direct", "offer"]"#).expect("decodes");
assert_eq!(
kinds,
[
TransferKind::SelfTransfer,
TransferKind::Direct,
TransferKind::Offer
]
);
}
#[test]
fn a_trailing_slash_on_the_base_url_does_not_double_up() {
let client = RegistryClient::new("https://scan.example.com/").unwrap();
assert_eq!(
client
.url(&["registry", "metadata", "v1", "info"])
.unwrap()
.as_str(),
"https://scan.example.com/registry/metadata/v1/info"
);
}
#[test]
fn a_hostile_path_segment_cannot_redirect_the_request() {
let client = RegistryClient::new("https://scan.example.com").unwrap();
for hostile in ["a?b", "a#b", "../../info", "a/b"] {
let url = client
.url(&["registry", "metadata", "v1", "instruments", hostile])
.unwrap();
assert_eq!(
url.query(),
None,
"`{hostile}` must not become a query: {url}"
);
assert_eq!(
url.path_segments().map(Iterator::count),
Some(5),
"`{hostile}` must stay one segment: {url}"
);
assert!(
url.as_str()
.starts_with("https://scan.example.com/registry/metadata/v1/instruments/"),
"`{hostile}` escaped the collection: {url}"
);
}
}
#[test]
fn a_base_url_that_is_not_a_url_is_refused_immediately() {
let err = RegistryClient::new("not a url").expect_err("not a URL");
assert!(!err.is_retriable(), "a typo is not worth retrying: {err}");
assert!(err.to_string().contains("not a url"), "{err}");
}
#[test]
fn the_factory_request_is_the_shape_the_standard_specifies() {
let args = serde_json::json!({ "expectedAdmin": "dso::1220", "transfer": {} });
let request = GetFactoryRequest {
choice_arguments: &args,
exclude_debug_fields: true,
};
assert_eq!(
serde_json::to_value(&request).unwrap(),
serde_json::json!({
"choiceArguments": { "expectedAdmin": "dso::1220", "transfer": {} },
"excludeDebugFields": true
})
);
}
#[test]
fn a_factory_response_decodes_and_its_context_is_translated() {
let json = serde_json::json!({
"factoryId": "00factory",
"transferKind": "direct",
"choiceContext": {
"choiceContextData": { "values": {} },
"disclosedContracts": [{
"templateId": "pkg:Splice.AmuletRules:AmuletRules",
"contractId": "001",
"createdEventBlob": "AQI=",
"synchronizerId": "sync::1220ab"
}]
}
});
let wire: WireTransferFactory = serde_json::from_value(json).expect("decodes");
assert_eq!(wire.factory_id, "00factory");
assert_eq!(wire.transfer_kind, TransferKind::Direct);
let context = ChoiceContext::from_wire(wire.choice_context).expect("translates");
assert_eq!(context.disclosed_contracts().len(), 1);
assert_eq!(context.disclosed_contracts()[0].created_event_blob, [1, 2]);
}
#[test]
fn the_allocation_factory_response_has_no_transfer_kind() {
let json = serde_json::json!({
"factoryId": "00factory",
"choiceContext": { "choiceContextData": {}, "disclosedContracts": [] }
});
let wire: WireFactory = serde_json::from_value(json).expect("decodes");
assert_eq!(wire.factory_id, "00factory");
let json = serde_json::json!({
"factoryId": "00factory",
"choiceContext": { "choiceContextData": {}, "disclosedContracts": [] }
});
assert!(serde_json::from_value::<WireTransferFactory>(json).is_err());
}
#[test]
fn an_unknown_transfer_kind_decodes_instead_of_failing_the_response() {
let json = serde_json::json!({
"factoryId": "00factory",
"transferKind": "escrow-something-new",
"choiceContext": { "choiceContextData": {}, "disclosedContracts": [] }
});
let wire: WireTransferFactory =
serde_json::from_value(json).expect("an unknown kind is a value, not a decode failure");
assert_eq!(wire.transfer_kind, TransferKind::Unknown);
}
#[test]
fn an_absent_decimals_is_the_standards_default_and_not_unknown() {
let instrument: Instrument = serde_json::from_value(serde_json::json!({
"id": "Amulet", "name": "Canton Coin", "symbol": "CC"
}))
.expect("decodes");
assert_eq!(instrument.decimals, 10);
}
#[test]
fn instruments_decode_with_the_optional_fields_absent() {
let json = serde_json::json!({
"instruments": [{ "id": "Amulet", "name": "Canton Coin", "symbol": "CC" }]
});
let response: ListInstrumentsResponse = serde_json::from_value(json).expect("decodes");
assert_eq!(response.instruments.len(), 1);
assert_eq!(response.instruments[0].id, "Amulet");
assert_eq!(response.instruments[0].total_supply, None);
assert_eq!(response.instruments[0].decimals, 10);
assert_eq!(response.next_page_token, None);
}
#[test]
fn a_tls_failure_is_permanent_under_either_phrasing() {
for detail in [
"error sending request: invalid peer certificate: UnknownIssuer",
"error sending request: tls handshake eof",
"CERTIFICATE verify failed",
] {
let error = classify_transport(Transport::Connect, detail);
assert!(
matches!(error, Error::InvalidRequest(_)),
"{detail}: {error:?}"
);
assert!(!error.is_retriable(), "{detail}");
assert!(error.to_string().contains("TLS certificate"), "{error}");
}
let refused = classify_transport(Transport::Connect, "connection refused (os error 61)");
assert!(matches!(refused, Error::Connection(_)), "{refused:?}");
assert!(refused.is_retriable());
assert!(
refused.to_string().contains("connection refused"),
"{refused}"
);
}
#[test]
fn the_typed_verdicts_are_read_before_the_strings() {
let timeout = classify_transport(
Transport::Timeout,
"certificate", );
assert!(matches!(timeout, Error::Timeout), "{timeout:?}");
for kind in [Transport::Builder, Transport::Redirect] {
let error = classify_transport(kind, "x");
assert!(
matches!(&error, Error::InvalidRequest(m) if m.contains("could not be made")),
"{kind:?}: {error:?}"
);
}
let decode = classify_transport(Transport::Decode, "x");
assert!(matches!(decode, Error::UnexpectedResponse(_)), "{decode:?}");
}
#[test]
fn truncate_cuts_past_the_limit_and_on_a_character_boundary() {
let exact = "x".repeat(200);
assert_eq!(truncate(&exact), exact);
let over = "x".repeat(201);
let cut = truncate(&over);
assert!(cut.ends_with('…'), "{cut}");
assert_eq!(cut.trim_end_matches('…').len(), 200);
let straddle = format!("{}€€", "x".repeat(199));
let cut = truncate(&straddle);
assert!(cut.ends_with('…'), "{cut}");
assert!(cut.starts_with(&"x".repeat(199)), "{cut}");
}
}