use bytes::Bytes;
use helix_core::effect::{HttpRequest, HttpResponse};
use helix_core::ports::HttpRequester;
use helix_core::PortError;
use reqwest::header::USER_AGENT;
use reqwest::{Client, Method};
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Instant;
use crate::metrics::{
AsyncMetricSink, LabelKey, MetricEvent, MetricId, MetricLabels, NoopMetricSink,
};
use super::network_util::{
header_name, header_value, headers_to_strings, resolve_http_url, validate_url,
};
use super::{HostHeaderRegistry, HostNetworkConfig};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PreparedHttpRequest {
pub method: String,
pub url: String,
pub headers: BTreeMap<String, String>,
pub body: Option<Bytes>,
}
#[derive(Clone)]
pub struct SharedHttpClient {
client: Client,
config: HostNetworkConfig,
headers: HostHeaderRegistry,
metrics: Arc<dyn AsyncMetricSink>,
inflight: Arc<AtomicUsize>,
}
impl SharedHttpClient {
pub fn new(config: HostNetworkConfig) -> Result<Self, PortError> {
Self::with_registry(config, HostHeaderRegistry::default())
}
pub fn with_registry(
config: HostNetworkConfig,
headers: HostHeaderRegistry,
) -> Result<Self, PortError> {
validate_url(&config.api_base_url, "api_base_url")?;
if !config.default_api_base_url.is_empty() {
validate_url(&config.default_api_base_url, "default_api_base_url")?;
}
let mut builder = Client::builder().timeout(config.timeout);
if let Some(user_agent) = &config.user_agent {
builder = builder.user_agent(user_agent);
}
let client = builder
.build()
.map_err(|e| PortError::Http(format!("reqwest client build error: {e}")))?;
Ok(Self {
client,
config,
headers,
metrics: Arc::new(NoopMetricSink),
inflight: Arc::new(AtomicUsize::new(0)),
})
}
pub fn headers(&self) -> HostHeaderRegistry {
self.headers.clone()
}
pub fn with_metric_sink(mut self, metrics: Arc<dyn AsyncMetricSink>) -> Self {
self.metrics = metrics;
self
}
pub async fn prepare_for_test(
&self,
req: HttpRequest,
) -> Result<PreparedHttpRequest, PortError> {
self.prepare(req).await
}
async fn prepare(&self, req: HttpRequest) -> Result<PreparedHttpRequest, PortError> {
let method = Method::from_bytes(req.method.as_bytes())
.map_err(|e| PortError::Http(format!("invalid method '{}': {e}", req.method)))?;
let url = resolve_http_url(&self.config.api_base_url, &req.url)?;
let mut headers = self.headers.snapshot().await;
if let Some(user_agent) = &self.config.user_agent {
headers.insert(USER_AGENT, header_value("user-agent", user_agent)?);
}
for (name, value) in &req.headers {
let name = header_name(name)?;
let value = header_value(name.as_str(), value)?;
headers.insert(name, value);
}
Ok(PreparedHttpRequest {
method: method.as_str().to_string(),
url,
headers: headers_to_strings(&headers),
body: req.body,
})
}
fn classify_error(e: &reqwest::Error) -> PortError {
if e.is_timeout() {
PortError::Transport(format!("timeout: {e}"))
} else if e.is_connect() || e.is_request() {
PortError::Transport(format!("network: {e}"))
} else {
PortError::Http(e.to_string())
}
}
async fn execute(&self, req: HttpRequest) -> Result<HttpResponse, PortError> {
let prepared = self.prepare(req).await?;
crate::network_debug::dump_http_request(&prepared);
let method = Method::from_bytes(prepared.method.as_bytes())
.map_err(|e| PortError::Http(format!("invalid method '{}': {e}", prepared.method)))?;
let mut builder = self.client.request(method, &prepared.url);
for (name, value) in &prepared.headers {
builder = builder.header(name, value);
}
if let Some(body) = prepared.body {
builder = builder.body(body);
}
let resp = builder.send().await.map_err(|e| Self::classify_error(&e))?;
let status = resp.status().as_u16();
let headers = resp
.headers()
.iter()
.map(|(k, v)| {
(
k.as_str().to_string(),
v.to_str().unwrap_or("<non-utf8>").to_string(),
)
})
.collect::<Vec<_>>();
let body = resp.bytes().await.map_err(|e| Self::classify_error(&e))?;
crate::network_debug::dump_http_response(status, &headers, &body);
Ok(HttpResponse {
status,
headers,
body,
})
}
}
#[async_trait::async_trait]
impl HttpRequester for SharedHttpClient {
async fn request(&self, req: HttpRequest) -> Result<HttpResponse, PortError> {
if !self.metrics.is_enabled() {
return self.execute(req).await;
}
let request_bytes = req.body.as_ref().map_or(0, Bytes::len);
let started = Instant::now();
let inflight = self.inflight.fetch_add(1, Ordering::Relaxed) + 1;
record_http_gauge(self.metrics.as_ref(), inflight);
let result = self.execute(req).await;
let remaining = self
.inflight
.fetch_sub(1, Ordering::Relaxed)
.saturating_sub(1);
record_http_gauge(self.metrics.as_ref(), remaining);
let failed = match &result {
Ok(response) => response.status >= 400,
Err(_) => true,
};
let status = if failed { "error" } else { "ok" };
let labels = MetricLabels::one(LabelKey::Stage, "http")
.with(LabelKey::Protocol, "http")
.with(LabelKey::Operation, "request")
.with(LabelKey::Status, status);
let _ = self.metrics.try_record(MetricEvent::histogram(
MetricId::HttpRequestDurationSeconds,
started.elapsed().as_secs_f64(),
labels,
));
let _ = self.metrics.try_record(MetricEvent::histogram(
MetricId::HttpRequestBytes,
request_bytes as f64,
labels,
));
if let Ok(response) = &result {
let status_labels = labels.with(LabelKey::StatusClass, status_class(response.status));
let _ = self.metrics.try_record(MetricEvent::histogram(
MetricId::HttpResponseBytes,
response.body.len() as f64,
status_labels,
));
let _ = self.metrics.try_record(MetricEvent::counter(
MetricId::HttpStatusTotal,
1.0,
status_labels,
));
}
let _ =
self.metrics
.try_record(MetricEvent::counter(MetricId::OperationsTotal, 1.0, labels));
if failed {
if result
.as_ref()
.err()
.is_some_and(|error| error.to_string().to_ascii_lowercase().contains("timeout"))
{
let _ = self.metrics.try_record(MetricEvent::counter(
MetricId::HttpTimeoutTotal,
1.0,
labels,
));
}
let error_kind = if result.is_err() {
"http_request_failed"
} else {
"http_status_error"
};
let _ = self.metrics.try_record(MetricEvent::counter(
MetricId::ErrorsTotal,
1.0,
labels.with(LabelKey::ErrorKind, error_kind),
));
}
result
}
}
fn status_class(status: u16) -> &'static str {
match status {
100..=199 => "1xx",
200..=299 => "2xx",
300..=399 => "3xx",
400..=499 => "4xx",
500..=599 => "5xx",
_ => "other",
}
}
fn record_http_gauge(metrics: &dyn AsyncMetricSink, inflight: usize) {
let _ = metrics.try_record(MetricEvent::gauge(
MetricId::HttpInflight,
inflight as f64,
MetricLabels::one(LabelKey::Stage, "http"),
));
}