use std::sync::Arc;
use alkcall::client::{AdapterError, OperationAdapter};
use alkcall::core::types::Capabilities;
use alkcall::registry::context::OperationContext;
use alkcall::registry::registration::{
make_handler, make_streaming_handler, HandlerKind, HandlerRegistration, OperationProvenance,
};
use alkcall::registry::spec::{OperationSpec, OperationType, Visibility};
use crate::adapters::input_validation::CompiledInputSchema;
use async_trait::async_trait;
use reqwest::Method;
use serde_json::Value;
use super::forward::{forward, forward_stream, validate_path_template, HttpServiceConfig};
use crate::client::SharedHttpClient;
pub struct FromJsonSchema {
spec: OperationSpec,
config: HttpServiceConfig,
path_template: String,
method: String,
http_client: Arc<SharedHttpClient>,
}
impl FromJsonSchema {
pub fn new(
spec: OperationSpec,
config: HttpServiceConfig,
path_template: String,
method: String,
http_client: Arc<SharedHttpClient>,
) -> Result<Self, AdapterError> {
validate_method(&method)?;
validate_path_template(&path_template)?;
Self::validate_base_url(&config.base_url)?;
if spec_name_references_undeclared(&spec, &path_template) {
return Err(AdapterError::SchemaParse {
message: format!(
"path template `{path_template}` references a placeholder the \
operation's input_schema does not declare; the forwarded call \
could never satisfy it"
),
});
}
if config.base_url.is_empty() {
return Err(AdapterError::SchemaParse {
message: "base_url must not be empty".into(),
});
}
let spec = OperationSpec {
visibility: Visibility::Internal,
..spec
};
Ok(Self {
spec,
config,
path_template,
method,
http_client,
})
}
fn validate_base_url(base_url: &str) -> Result<(), AdapterError> {
let parsed = url::Url::parse(base_url).map_err(|e| AdapterError::SchemaParse {
message: format!("invalid base_url `{base_url}`: {e}"),
})?;
let scheme = parsed.scheme();
if scheme != "https" && scheme != "http" {
return Err(AdapterError::SchemaParse {
message: format!("base_url `{base_url}` must be an http(s) URL; `{scheme}` is not"),
});
}
if !parsed.username().is_empty() || parsed.password().is_some() {
return Err(AdapterError::SchemaParse {
message: format!(
"base_url `{base_url}` must not embed userinfo; credentials are \
injected per-call from Capabilities"
),
});
}
Ok(())
}
}
fn validate_method(method: &str) -> Result<(), AdapterError> {
if method.is_empty() {
return Err(AdapterError::SchemaParse {
message: "HTTP method must not be empty".into(),
});
}
Method::from_bytes(method.to_ascii_uppercase().as_bytes()).map_err(|_| {
AdapterError::SchemaParse {
message: format!("invalid HTTP method `{method}`"),
}
})?;
Ok(())
}
fn spec_name_references_undeclared(spec: &OperationSpec, path_template: &str) -> bool {
let properties = spec
.input_schema
.get("properties")
.and_then(|p| p.as_object())
.map(|p| p.keys().cloned().collect::<Vec<_>>())
.unwrap_or_default();
let mut rest = path_template;
while let Some(start) = rest.find('{') {
let Some(end_rel) = rest[start..].find('}') else {
return false;
};
let name = &rest[start + 1..start + end_rel];
if !properties.iter().any(|k| k == name) {
return true;
}
rest = &rest[start + end_rel + 1..];
}
false
}
#[async_trait]
impl OperationAdapter for FromJsonSchema {
async fn import(&self) -> Result<Vec<HandlerRegistration>, AdapterError> {
let input_validator =
CompiledInputSchema::compile(&self.spec.input_schema).map_err(|message| {
AdapterError::SchemaParse {
message: format!(
"operation `{}`: {message} (review 002 OAI-18: an un-validatable input \
contract must fail import, not register unenforced)",
self.spec.name
),
}
})?;
let path_template = self.path_template.clone();
let method_upper = self.method.to_ascii_uppercase();
let auth_scheme = self.config.auth.clone();
let default_headers = self.config.default_headers.clone();
let base_url = self.config.base_url.clone();
let namespace = self.config.namespace.clone();
let http_client = Arc::clone(&self.http_client);
let op_type = self.spec.op_type;
let input_schema = self.spec.input_schema.clone();
let error_status_codes: Vec<(u16, String)> = self
.spec
.error_schemas
.iter()
.map(|e| (e.http_status.unwrap_or(0), e.code.clone()))
.collect();
let handler = if op_type == OperationType::Sub {
let stream_handler =
make_streaming_handler(move |input: Value, context: OperationContext| {
let path_template = path_template.clone();
let method_upper = method_upper.clone();
let auth_scheme = auth_scheme.clone();
let default_headers = default_headers.clone();
let base_url = base_url.clone();
let namespace = namespace.clone();
let http_client = Arc::clone(&http_client);
let error_status_codes = error_status_codes.clone();
let input_schema = input_schema.clone();
let input_validator = input_validator.clone();
forward_stream(
&http_client,
&base_url,
&path_template,
&method_upper,
&auth_scheme,
&default_headers,
&namespace,
&input_schema,
&error_status_codes,
Some(&input_validator),
input,
context,
)
});
HandlerKind::Stream(stream_handler)
} else {
let once_handler = make_handler(move |input: Value, context: OperationContext| {
let path_template = path_template.clone();
let method_upper = method_upper.clone();
let auth_scheme = auth_scheme.clone();
let default_headers = default_headers.clone();
let base_url = base_url.clone();
let namespace = namespace.clone();
let http_client = Arc::clone(&http_client);
let error_status_codes = error_status_codes.clone();
let input_schema = input_schema.clone();
let input_validator = input_validator.clone();
async move {
forward(
&http_client,
&base_url,
&path_template,
&method_upper,
&auth_scheme,
&default_headers,
&namespace,
&input_schema,
&error_status_codes,
Some(&input_validator),
input,
context,
)
.await
}
});
HandlerKind::Once(once_handler)
};
let capabilities = Capabilities::new();
Ok(vec![HandlerRegistration::new(
self.spec.clone(),
handler,
OperationProvenance::FromJsonSchema,
None,
None,
capabilities,
)])
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::adapters::forward::{build_request, HttpAuthScheme};
use crate::client::HttpClientConfig;
use alkcall::protocol::wire::ResponseEnvelope;
use alkcall::registry::context::AbortPolicy;
use alkcall::registry::env::OperationEnv;
use alkcall::registry::spec::{AccessControl, ErrorDefinition, Visibility};
use futures::StreamExt;
use reqwest::header::AUTHORIZATION;
use reqwest::Method;
use std::collections::HashMap;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
fn noop_context(request_id: &str, capabilities: Capabilities) -> OperationContext {
struct NoopEnv;
#[async_trait]
impl OperationEnv for NoopEnv {
async fn invoke_with_policy(
&self,
_ns: &str,
_op: &str,
_input: Value,
parent: &OperationContext,
_policy: AbortPolicy,
) -> ResponseEnvelope {
ResponseEnvelope::ok(parent.request_id.clone(), Value::Null)
}
fn contains(&self, _name: &str) -> bool {
false
}
}
OperationContext {
request_id: request_id.to_string(),
parent_request_id: None,
identity: None,
handler_identity: None,
forwarded_for: None,
capabilities,
metadata: HashMap::new(),
scoped_env: alkcall::registry::context::ScopedPeerEnv::empty(),
env: Arc::new(NoopEnv),
abort_policy: AbortPolicy::default(),
deadline: Some(std::time::Instant::now() + Duration::from_secs(30)),
internal: true,
ownership: None,
}
}
fn test_spec(name: &str, op_type: OperationType) -> OperationSpec {
OperationSpec::new(
name,
op_type,
Visibility::Internal,
serde_json::json!({"type":"object","properties":{"id":{"type":"string"}}}),
serde_json::json!({"type":"object"}),
vec![],
AccessControl::default(),
None,
)
}
fn test_spec_with_errors(
name: &str,
op_type: OperationType,
errors: Vec<ErrorDefinition>,
) -> OperationSpec {
OperationSpec::new(
name,
op_type,
Visibility::Internal,
serde_json::json!({"type":"object"}),
serde_json::json!({"type":"object"}),
errors,
AccessControl::default(),
None,
)
}
fn test_config(namespace: &str, base_url: &str) -> HttpServiceConfig {
HttpServiceConfig {
namespace: namespace.to_string(),
base_url: base_url.to_string(),
auth: None,
default_headers: HashMap::new(),
}
}
fn test_http_client() -> Arc<SharedHttpClient> {
Arc::new(SharedHttpClient::new(HttpClientConfig::default()).unwrap())
}
#[tokio::test]
async fn import_produces_one_handler_registration() {
let adapter = FromJsonSchema::new(
test_spec("svc/getWidget", OperationType::Query),
test_config("svc", "https://api.example.com"),
"/widgets".to_string(),
"GET".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
assert_eq!(bundles.len(), 1);
assert_eq!(bundles[0].spec.name, "svc/getWidget");
assert_eq!(bundles[0].provenance, OperationProvenance::FromJsonSchema);
assert!(bundles[0].composition_authority.is_none());
assert!(bundles[0].scoped_env.is_none());
}
#[tokio::test]
async fn query_op_registration_is_handler_kind_once() {
let adapter = FromJsonSchema::new(
test_spec("svc/getWidget", OperationType::Query),
test_config("svc", "https://api.example.com"),
"/widgets".to_string(),
"GET".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
assert!(matches!(bundles[0].handler, HandlerKind::Once(_)));
}
#[tokio::test]
async fn sub_op_registration_is_handler_kind_stream() {
let adapter = FromJsonSchema::new(
test_spec("svc/stream", OperationType::Sub),
test_config("svc", "https://api.example.com"),
"/stream".to_string(),
"POST".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
assert!(matches!(bundles[0].handler, HandlerKind::Stream(_)));
}
#[tokio::test]
async fn mutation_op_registration_is_handler_kind_once() {
let adapter = FromJsonSchema::new(
test_spec("svc/createWidget", OperationType::Mutation),
test_config("svc", "https://api.example.com"),
"/widgets".to_string(),
"POST".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
assert!(matches!(bundles[0].handler, HandlerKind::Once(_)));
}
#[tokio::test]
async fn build_request_injects_bearer_from_capabilities() {
let caps = Capabilities::new().with_http_token("github", "tok-123".to_string());
let ctx = noop_context("req-1", caps);
let (method, url, _body, headers) = build_request(
"https://api.github.com",
"/repos/{owner}/{repo}/issues",
"GET",
&Some(HttpAuthScheme::Bearer),
&HashMap::new(),
"github",
&serde_json::json!({"type": "object", "additionalProperties": true}),
None,
&serde_json::json!({"owner":"a","repo":"b"}),
&ctx,
)
.unwrap();
assert_eq!(method, Method::GET);
assert_eq!(url.path(), "/repos/a/b/issues");
assert_eq!(url.host_str(), Some("api.github.com"));
let auth = headers.get(AUTHORIZATION).unwrap();
assert_eq!(auth.to_str().unwrap(), "Bearer tok-123");
}
#[tokio::test]
async fn build_request_path_and_query_split() {
let ctx = noop_context("req-2", Capabilities::new());
let (_, url, _, _) = build_request(
"https://api.example.com",
"/widgets/{id}",
"GET",
&None,
&HashMap::new(),
"svc",
&serde_json::json!({"type": "object", "additionalProperties": true}),
None,
&serde_json::json!({"id":42,"filter":"active"}),
&ctx,
)
.unwrap();
assert_eq!(url.path(), "/widgets/42");
assert_eq!(url.query().unwrap(), "filter=active");
}
async fn spawn_echo_server(
status: u16,
body: &'static str,
content_type: &'static str,
) -> String {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(pair) => pair,
Err(_) => break,
};
let status_line = match status {
200 => "200 OK",
201 => "201 Created",
404 => "404 Not Found",
500 => "500 Internal Server Error",
_ => "200 OK",
};
let body_bytes = body.as_bytes();
let response = format!(
"HTTP/1.1 {status_line}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
body_bytes.len(),
body
);
let mut buf = [0u8; 4096];
let _ = sock.read(&mut buf).await;
sock.write_all(response.as_bytes()).await.unwrap();
sock.flush().await.unwrap();
}
});
format!("http://{addr}")
}
#[tokio::test]
async fn integration_forwarding_handler_calls_external_endpoint() {
let base = spawn_echo_server(200, r#"{"ok":true}"#, "application/json").await;
let adapter = FromJsonSchema::new(
test_spec("svc/data", OperationType::Query),
test_config("svc", &base),
"/data".to_string(),
"GET".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
let registration = &bundles[0];
let ctx = noop_context("req-10", Capabilities::new());
let response = match ®istration.handler {
HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
_ => panic!("expected Once handler"),
};
assert_eq!(response.request_id, "req-10");
match response.result {
Ok(v) => assert_eq!(v, serde_json::json!({"ok":true})),
Err(e) => panic!("expected Ok, got {e:?}"),
}
}
#[tokio::test]
async fn integration_non_2xx_returns_declared_error() {
let base = spawn_echo_server(404, r#"{"error":"missing"}"#, "application/json").await;
let errors = vec![ErrorDefinition {
code: "HTTP_404".to_string(),
description: "Not found".to_string(),
schema: serde_json::json!({"type":"object"}),
http_status: Some(404),
}];
let adapter = FromJsonSchema::new(
test_spec_with_errors("svc/missing", OperationType::Query, errors),
test_config("svc", &base),
"/missing".to_string(),
"GET".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
let registration = &bundles[0];
let ctx = noop_context("req-11", Capabilities::new());
let response = match ®istration.handler {
HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
_ => panic!("expected Once handler"),
};
match response.result {
Err(e) => {
assert_eq!(e.code, "HTTP_404");
assert!(!e.retryable);
}
other => panic!("expected HTTP_404 error, got {other:?}"),
}
}
#[tokio::test]
async fn integration_undeclared_error_status_returns_http_status_code() {
let base = spawn_echo_server(500, "boom", "text/plain").await;
let adapter = FromJsonSchema::new(
test_spec("svc/x", OperationType::Query),
test_config("svc", &base),
"/x".to_string(),
"GET".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
let registration = &bundles[0];
let ctx = noop_context("req-12", Capabilities::new());
let response = match ®istration.handler {
HandlerKind::Once(h) => h(serde_json::json!({}), ctx).await,
_ => panic!("expected Once handler"),
};
match response.result {
Err(e) => assert_eq!(e.code, "HTTP_500"),
other => panic!("expected HTTP_500, got {other:?}"),
}
}
#[tokio::test]
async fn integration_sse_subscription_streams_responded_events() {
let sse_body = "data: {\"n\":1}\n\ndata: {\"n\":2}\n\n";
let base = spawn_echo_server(200, sse_body, "text/event-stream").await;
let adapter = FromJsonSchema::new(
test_spec("svc/stream", OperationType::Sub),
test_config("svc", &base),
"/stream".to_string(),
"POST".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
let registration = &bundles[0];
let ctx = noop_context("req-13", Capabilities::new());
let stream = match ®istration.handler {
HandlerKind::Stream(h) => h(serde_json::json!({}), ctx),
_ => panic!("expected Stream handler"),
};
let collected: Vec<ResponseEnvelope> = stream.collect().await;
assert_eq!(collected.len(), 2);
assert_eq!(collected[0].result, Ok(serde_json::json!({"n":1})));
assert_eq!(collected[1].result, Ok(serde_json::json!({"n":2})));
assert_eq!(collected[0].request_id, "req-13");
assert_eq!(collected[1].request_id, "req-13");
}
#[tokio::test]
async fn integration_sse_subscription_invalid_input_yields_one_error_envelope_and_ends() {
for (input, expected_fragment) in [
(
serde_json::json!({"debug": true}),
"not declared by the operation's input schema",
),
(
serde_json::json!({"id": {"nested": "object"}}),
"[keyword: type",
),
] {
let base = spawn_echo_server(200, "data: {\"n\":1}\n\n", "text/event-stream").await;
let adapter = FromJsonSchema::new(
test_spec("svc/stream", OperationType::Sub),
test_config("svc", &base),
"/stream/{id}".to_string(),
"POST".to_string(),
test_http_client(),
)
.unwrap();
let bundles = adapter.import().await.unwrap();
let registration = &bundles[0];
let ctx = noop_context("req-13b", Capabilities::new());
let stream = match ®istration.handler {
HandlerKind::Stream(h) => h(input.clone(), ctx),
_ => panic!("expected Stream handler"),
};
let collected: Vec<ResponseEnvelope> = stream.collect().await;
assert_eq!(
collected.len(),
1,
"exactly one error envelope, then the stream ends (never an empty-200-subscribe)"
);
match &collected[0].result {
Err(e) => {
assert_eq!(e.code, "INVALID_INPUT", "message was: {}", e.message);
assert!(
e.message.contains(expected_fragment),
"message `{}` lacks `{expected_fragment}`",
e.message
);
}
other => panic!("expected INVALID_INPUT, got {other:?}"),
}
assert_eq!(collected[0].request_id, "req-13b");
}
}
#[test]
fn no_env_vars_read_in_build_request() {
std::env::set_var("OPENAI_API_KEY", "should-not-be-used");
let ctx = noop_context("req-14", Capabilities::new());
let err = build_request(
"https://api.openai.com",
"/v1/chat",
"POST",
&Some(HttpAuthScheme::Bearer),
&HashMap::new(),
"openai",
&serde_json::json!({"type": "object", "additionalProperties": true}),
None,
&serde_json::json!({"body":{"prompt":"hi"}}),
&ctx,
)
.expect_err("absent capability is a loud error (FWD-16), never an unauthenticated send");
assert_eq!(err.code, "INTERNAL");
assert!(
err.message
.contains("capability for namespace `openai` is absent"),
"message was: {}",
err.message
);
assert!(
!err.message.contains("should-not-be-used"),
"error must not echo env material"
);
std::env::remove_var("OPENAI_API_KEY");
}
#[tokio::test]
async fn uncompilable_input_schema_fails_import_loudly() {
let mut spec = test_spec("svc/broken", OperationType::Query);
spec.input_schema = serde_json::json!({"type": "object", "required": "n"});
let adapter = FromJsonSchema::new(
spec,
test_config("svc", "https://api.example.com"),
"/x".to_string(),
"GET".to_string(),
test_http_client(),
)
.unwrap();
let result = adapter.import().await;
match result {
Err(AdapterError::SchemaParse { message }) => {
assert!(
message.contains("failed to compile"),
"message was: {message}"
);
assert!(message.contains("svc/broken"), "message was: {message}");
}
Ok(_) => panic!("an un-validatable input contract must fail import (OAI-18)"),
Err(e) => panic!("expected SchemaParse, got {e:?}"),
}
}
#[test]
fn malformed_method_path_template_and_base_url_fail_at_construction() {
for (method, path_template, base_url, expected_fragment) in [
("NOT/*A METHOD", "/x", "https://x", "invalid HTTP method"),
("GET", "/x/{open", "https://x", "unterminated placeholder"),
("GET", "/x/}", "https://x", "without a matching"),
("GET", "/x", "https://u:p@x", "userinfo"),
("GET", "/x", "ftp://x", "must be an http(s) URL"),
("GET", "/x", "not a url", "invalid base_url"),
] {
let result = FromJsonSchema::new(
test_spec("svc/x", OperationType::Query),
test_config("svc", base_url),
path_template.to_string(),
method.to_string(),
test_http_client(),
);
match result {
Err(AdapterError::SchemaParse { message }) => {
assert!(
message.contains(expected_fragment),
"message `{message}` lacks `{expected_fragment}`"
);
}
Ok(_) => panic!(
"malformed adapter config ({method}, {path_template}, {base_url}) must fail at construction"
),
Err(e) => panic!("expected SchemaParse, got {e}"),
}
}
}
#[test]
fn construction_forces_visibility_internal_and_validates_placeholder_binding() {
let spec = OperationSpec::new(
"svc/getWidget",
OperationType::Query,
Visibility::External,
serde_json::json!({"type":"object","properties":{"id":{"type":"string"}}}),
serde_json::json!({"type":"object"}),
vec![],
AccessControl::default(),
None,
);
let adapter = FromJsonSchema::new(
spec,
test_config("svc", "https://api.example.com"),
"/widgets/{id}".to_string(),
"GET".to_string(),
test_http_client(),
)
.expect("adapter builds");
let bundles = futures::executor::block_on(adapter.import()).unwrap();
assert_eq!(bundles[0].spec.visibility, Visibility::Internal);
let spec = OperationSpec::new(
"svc/getWidget",
OperationType::Query,
Visibility::Internal,
serde_json::json!({"type":"object","properties":{"widget_id":{"type":"string"}}}),
serde_json::json!({"type":"object"}),
vec![],
AccessControl::default(),
None,
);
let result = FromJsonSchema::new(
spec,
test_config("svc", "https://api.example.com"),
"/widgets/{id}".to_string(),
"GET".to_string(),
test_http_client(),
);
match result {
Err(AdapterError::SchemaParse { message }) => {
assert!(message.contains("placeholder"), "message was: {message}");
assert!(message.contains("{id}"), "message was: {message}");
}
Ok(_) => panic!("unbound placeholder must fail at construction"),
Err(e) => panic!("expected SchemaParse, got {e}"),
}
}
#[test]
fn construction_rejects_placeholder_when_input_schema_has_no_properties() {
let spec = OperationSpec::new(
"svc/getWidget",
OperationType::Query,
Visibility::Internal,
serde_json::json!({"type":"object"}),
serde_json::json!({"type":"object"}),
vec![],
AccessControl::default(),
None,
);
let result = FromJsonSchema::new(
spec,
test_config("svc", "https://api.example.com"),
"/widgets/{id}".to_string(),
"GET".to_string(),
test_http_client(),
);
match result {
Err(AdapterError::SchemaParse { message }) => {
assert!(message.contains("placeholder"), "message was: {message}");
assert!(message.contains("{id}"), "message was: {message}");
}
Ok(_) => {
panic!("no-properties schema with a placeholder template must fail at construction")
}
Err(e) => panic!("expected SchemaParse, got {e}"),
}
let spec = OperationSpec::new(
"svc/listWidgets",
OperationType::Query,
Visibility::Internal,
serde_json::json!({"type":"object"}),
serde_json::json!({"type":"object"}),
vec![],
AccessControl::default(),
None,
);
let adapter = FromJsonSchema::new(
spec,
test_config("svc", "https://api.example.com"),
"/widgets".to_string(),
"GET".to_string(),
test_http_client(),
)
.expect("placeholder-free template with a no-properties schema builds");
let bundles = futures::executor::block_on(adapter.import()).unwrap();
assert_eq!(bundles[0].spec.visibility, Visibility::Internal);
}
}