use std::collections::HashMap;
use std::sync::OnceLock;
use std::time::Duration;
use async_trait::async_trait;
use zeroize::Zeroizing;
use crate::backend::{encode_key, Backend, HealthStatus, LockableBackend};
use crate::error::{BackendError, BackendErrorKind};
use crate::metrics::{metrics_headers, MetricsProvider};
use crate::session::session_headers;
use crate::url_validator::validate_cachekitio_url;
pub struct CachekitIO {
client: reqwest::Client,
api_key: Zeroizing<String>,
api_url: String,
metrics_provider: OnceLock<MetricsProvider>,
}
impl std::fmt::Debug for CachekitIO {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CachekitIO")
.field("api_url", &self.api_url)
.field("api_key", &"<redacted>")
.finish()
}
}
impl CachekitIO {
pub fn builder() -> CachekitIOBuilder {
CachekitIOBuilder::default()
}
pub fn api_url(&self) -> &str {
&self.api_url
}
pub(crate) fn client(&self) -> &reqwest::Client {
&self.client
}
pub(crate) fn api_key_str(&self) -> &str {
self.api_key.as_str()
}
fn url(&self, key: &str) -> Result<String, BackendError> {
Ok(format!("{}/v1/cache/{}", self.api_url, encode_key(key)?))
}
pub(crate) fn ttl_url(&self, key: &str) -> Result<String, BackendError> {
Ok(format!("{}/ttl", self.url(key)?))
}
pub(crate) fn lock_url(&self, key: &str) -> Result<String, BackendError> {
Ok(format!("{}/lock", self.url(key)?))
}
fn health_url(&self) -> String {
format!("{}/v1/cache/health", self.api_url)
}
pub(crate) fn with_standard_headers(
&self,
mut req: reqwest::RequestBuilder,
) -> reqwest::RequestBuilder {
for (name, value) in session_headers() {
req = req.header(name, value);
}
for (name, value) in metrics_headers(self.metrics_provider.get()) {
req = req.header(name, value);
}
req
}
pub(crate) async fn error_from_response(&self, resp: reqwest::Response) -> BackendError {
let status = resp.status().as_u16();
let body = resp.bytes().await.unwrap_or_default();
from_http_status_sanitized(status, &body, self.api_key.as_str())
}
}
pub(crate) fn reqwest_err_sanitized(e: reqwest::Error, api_key: &str) -> BackendError {
let kind = if e.is_timeout() {
BackendErrorKind::Timeout
} else {
BackendErrorKind::Transient
};
BackendError {
kind,
message: BackendError::sanitize_message(&e.to_string(), api_key),
source: Some(Box::new(e)),
}
}
pub(crate) fn from_http_status_sanitized(status: u16, body: &[u8], api_key: &str) -> BackendError {
let sanitized =
BackendError::sanitize_message(std::str::from_utf8(body).unwrap_or(""), api_key);
BackendError::from_http_status(status, sanitized.as_bytes())
}
#[cfg(not(target_arch = "wasm32"))]
#[cfg_attr(not(feature = "unsync"), async_trait)]
#[cfg_attr(feature = "unsync", async_trait(?Send))]
impl Backend for CachekitIO {
fn attach_metrics(&self, provider: MetricsProvider) {
self.metrics_provider.get_or_init(|| provider);
}
async fn get(&self, key: &str) -> Result<Option<Vec<u8>>, BackendError> {
let req = self.with_standard_headers(
self.client
.get(self.url(key)?)
.bearer_auth(self.api_key.as_str()),
);
let resp = req
.send()
.await
.map_err(|e| reqwest_err_sanitized(e, self.api_key.as_str()))?;
match resp.status().as_u16() {
200 => {
let bytes = resp
.bytes()
.await
.map_err(|e| reqwest_err_sanitized(e, self.api_key.as_str()))?;
Ok(Some(bytes.to_vec()))
}
404 => Ok(None),
_ => Err(self.error_from_response(resp).await),
}
}
async fn set(
&self,
key: &str,
value: Vec<u8>,
ttl: Option<Duration>,
) -> Result<(), BackendError> {
let mut req = self
.client
.put(self.url(key)?)
.bearer_auth(self.api_key.as_str())
.header(reqwest::header::CONTENT_TYPE, "application/octet-stream")
.body(value);
if let Some(ttl) = ttl {
req = req.header("X-TTL", ttl.as_secs().to_string());
}
let req = self.with_standard_headers(req);
let resp = req
.send()
.await
.map_err(|e| reqwest_err_sanitized(e, self.api_key.as_str()))?;
let status = resp.status().as_u16();
if (200..300).contains(&status) {
Ok(())
} else {
Err(self.error_from_response(resp).await)
}
}
async fn delete(&self, key: &str) -> Result<bool, BackendError> {
let req = self.with_standard_headers(
self.client
.delete(self.url(key)?)
.bearer_auth(self.api_key.as_str()),
);
let resp = req
.send()
.await
.map_err(|e| reqwest_err_sanitized(e, self.api_key.as_str()))?;
match resp.status().as_u16() {
200 | 204 => Ok(true),
404 => Ok(false),
_ => Err(self.error_from_response(resp).await),
}
}
async fn exists(&self, key: &str) -> Result<bool, BackendError> {
let req = self.with_standard_headers(
self.client
.head(self.url(key)?)
.bearer_auth(self.api_key.as_str()),
);
let resp = req
.send()
.await
.map_err(|e| reqwest_err_sanitized(e, self.api_key.as_str()))?;
match resp.status().as_u16() {
200 => Ok(true),
404 => Ok(false),
status => Err(BackendError::from_http_status(status, &[])),
}
}
async fn health(&self) -> Result<HealthStatus, BackendError> {
let start = std::time::Instant::now();
let req = self.with_standard_headers(
self.client
.get(self.health_url())
.bearer_auth(self.api_key.as_str()),
);
let resp = req
.send()
.await
.map_err(|e| reqwest_err_sanitized(e, self.api_key.as_str()))?;
let latency = start.elapsed();
let status = resp.status().as_u16();
if (200..300).contains(&status) {
let mut details = HashMap::new();
details.insert("http_status".to_string(), status.to_string());
Ok(HealthStatus {
is_healthy: true,
latency_ms: latency.as_secs_f64() * 1000.0,
backend_type: "cachekitio".to_string(),
details,
})
} else {
Err(self.error_from_response(resp).await)
}
}
fn as_lockable(&self) -> Option<&dyn LockableBackend> {
Some(self)
}
}
#[derive(Default)]
#[must_use]
pub struct CachekitIOBuilder {
api_key: Option<Zeroizing<String>>,
api_url: Option<String>,
allow_custom_host: bool,
metrics_provider: Option<MetricsProvider>,
}
impl CachekitIOBuilder {
pub fn api_key(mut self, key: impl Into<String>) -> Self {
self.api_key = Some(Zeroizing::new(key.into()));
self
}
pub fn api_url(mut self, url: impl Into<String>) -> Self {
self.api_url = Some(url.into());
self
}
pub fn allow_custom_host(mut self, allow: bool) -> Self {
self.allow_custom_host = allow;
self
}
pub fn metrics_provider(mut self, provider: MetricsProvider) -> Self {
self.metrics_provider = Some(provider);
self
}
pub fn build(self) -> Result<CachekitIO, crate::error::CachekitError> {
use crate::error::CachekitError;
let api_key = self
.api_key
.filter(|k| !k.is_empty())
.ok_or_else(|| CachekitError::Config("api_key is required".to_string()))?;
let api_url = self
.api_url
.unwrap_or_else(|| "https://api.cachekit.io".to_string());
validate_cachekitio_url(&api_url, self.allow_custom_host)?;
let api_url = api_url.trim_end_matches('/').to_string();
let client = reqwest::Client::builder();
#[cfg(not(target_arch = "wasm32"))]
let client = client
.use_rustls_tls()
.redirect(reqwest::redirect::Policy::none())
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10));
let client = client
.build()
.map_err(|e| CachekitError::Config(format!("failed to build HTTP client: {e}")))?;
Ok(CachekitIO {
client,
api_key,
api_url,
metrics_provider: self
.metrics_provider
.map_or_else(OnceLock::new, OnceLock::from),
})
}
}
#[cfg(test)]
#[allow(clippy::expect_used)] mod metrics_wiring_tests {
use std::sync::Arc;
use super::CachekitIO;
use crate::backend::Backend;
use crate::metrics::{metrics_headers, L1Stats, MetricsProvider};
fn provider(l1_hits: u64) -> MetricsProvider {
Arc::new(move || {
Some(L1Stats {
l1_hits,
l2_hits: 0,
misses: 0,
l1_enabled: true,
})
})
}
fn l1_hits_header(backend: &CachekitIO) -> Option<String> {
metrics_headers(backend.metrics_provider.get())
.into_iter()
.find(|h| h.0 == "X-CacheKit-L1-Hits")
.map(|h| h.1)
}
#[test]
fn attach_fills_an_unset_provider() {
let backend = CachekitIO::builder()
.api_key("ck_test_key")
.build()
.expect("builder succeeds");
assert_eq!(l1_hits_header(&backend), None, "nothing attached yet");
backend.attach_metrics(provider(7));
assert_eq!(l1_hits_header(&backend).as_deref(), Some("7"));
}
#[test]
fn builder_provider_is_retained_over_attach() {
let backend = CachekitIO::builder()
.api_key("ck_test_key")
.metrics_provider(provider(1))
.build()
.expect("builder succeeds");
backend.attach_metrics(provider(99));
assert_eq!(
l1_hits_header(&backend).as_deref(),
Some("1"),
"user-supplied provider wins"
);
}
}
#[cfg(test)]
#[allow(clippy::expect_used)] mod path_encoding_tests {
use super::CachekitIO;
use url::Url;
const API: &str = "https://api.cachekit.io";
fn backend() -> CachekitIO {
CachekitIO::builder()
.api_url(API)
.api_key("ck_test_key")
.build()
.expect("builder should succeed for the canonical host")
}
#[test]
fn repro_raw_dot_key_escapes_the_cache_prefix() {
let cases = [
("..", "", "/v1/"),
("..", "/ttl", "/v1/ttl"),
("..", "/lock", "/v1/lock"),
(".", "", "/v1/cache/"),
];
for (key, suffix, escaped) in cases {
let raw = format!("{API}/v1/cache/{}{suffix}", urlencoding::encode(key));
let parsed = Url::parse(&raw).expect("parses");
assert_eq!(
parsed.path(),
escaped,
"raw {key:?}{suffix} should collapse to {escaped}"
);
let pct = key.replace('.', "%2E");
let enc = format!("{API}/v1/cache/{pct}{suffix}");
assert_eq!(
Url::parse(&enc).expect("parses").path(),
escaped,
"%2E-encoded {key:?}{suffix} also collapses — encoding cannot fix this in rust-url"
);
}
}
#[test]
fn reserved_segments_rejected_by_every_builder() {
let b = backend();
for key in [".", "..", "health", "ttl", "lock"] {
assert!(b.url(key).is_err(), "url({key:?}) must be rejected");
assert!(b.ttl_url(key).is_err(), "ttl_url({key:?}) must be rejected");
assert!(
b.lock_url(key).is_err(),
"lock_url({key:?}) must be rejected"
);
}
}
#[test]
fn safe_keys_never_escape_the_cache_prefix() {
let b = backend();
let vectors = [
"a:..",
"default:../../admin",
"k?x=1#f",
"a b",
"healthy", "x/../../health", "ns:default:func:m.f:args:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef:",
];
for key in vectors {
let base = Url::parse(&b.url(key).expect("url")).expect("parse base");
let ttl = Url::parse(&b.ttl_url(key).expect("ttl_url")).expect("parse ttl");
let lock = Url::parse(&b.lock_url(key).expect("lock_url")).expect("parse lock");
assert!(
base.path().starts_with("/v1/cache/") && base.path().len() > "/v1/cache/".len(),
"base path {} escaped prefix for {key:?}",
base.path()
);
assert_eq!(
base.path(),
format!("/v1/cache/{}", urlencoding::encode(key)),
"base wire path mismatch for {key:?}"
);
assert!(
ttl.path().starts_with("/v1/cache/") && ttl.path().ends_with("/ttl"),
"ttl path {} escaped prefix for {key:?}",
ttl.path()
);
assert!(
lock.path().starts_with("/v1/cache/") && lock.path().ends_with("/lock"),
"lock path {} escaped prefix for {key:?}",
lock.path()
);
}
}
}