use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use cratefield_core::{
Auth, Blob, BoundedHttpClient, Captcha, Classifier, Clock, CustomHostnames, Database, Embedder,
HarnessConfig, KeyValue, Mailer, Payments, Port, Ports, Push, RateLimiter, Realtime, Runtime,
TextModel, Tracker, UlidIdGen, VectorIndex,
};
use crate::config::EnvConfig;
use crate::ports::{ReqwestClient, SpawnDefer, TokioClock};
static WARNED_SIGNER: AtomicBool = AtomicBool::new(false);
fn warn_once(flag: &AtomicBool, message: &str) {
if !flag.swap(true, Ordering::Relaxed) {
tracing::warn!("{message}");
}
}
#[derive(Clone, Default)]
pub struct Native {
db: Option<Arc<dyn Database>>,
rate_limiter: Option<Arc<dyn RateLimiter>>,
kv: Option<Arc<dyn KeyValue>>,
blob: Option<Arc<dyn Blob>>,
push: Option<Arc<dyn Push>>,
payments: Option<Arc<dyn Payments>>,
tracker: Option<Arc<dyn Tracker>>,
realtime: Option<Arc<dyn Realtime>>,
text_model: Option<Arc<dyn TextModel>>,
classifier: Option<Arc<dyn Classifier>>,
vector_index: Option<Arc<dyn VectorIndex>>,
embedder: Option<Arc<dyn Embedder>>,
custom_hostnames: Option<Arc<dyn CustomHostnames>>,
mailer: Option<Arc<dyn Mailer>>,
captcha: Option<Arc<dyn Captcha>>,
auth: Option<Arc<dyn Auth>>,
auth_from_env: bool,
#[cfg(feature = "push")]
push_from_env: bool,
#[cfg(feature = "push")]
assembled_push: Arc<std::sync::OnceLock<(Arc<dyn Push>, cratefield_push_wiring::PushWiring)>>,
}
impl Native {
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn db(mut self, db: impl Database + 'static) -> Self {
self.db = Some(Arc::new(db));
self
}
#[must_use]
pub fn db_arc(mut self, db: Arc<dyn Database>) -> Self {
self.db = Some(db);
self
}
#[must_use]
pub fn rate_limiter(mut self, limiter: impl RateLimiter + 'static) -> Self {
self.rate_limiter = Some(Arc::new(limiter));
self
}
#[must_use]
pub fn rate_limiter_arc(mut self, limiter: Arc<dyn RateLimiter>) -> Self {
self.rate_limiter = Some(limiter);
self
}
#[must_use]
pub fn kv(mut self, kv: impl KeyValue + 'static) -> Self {
self.kv = Some(Arc::new(kv));
self
}
#[must_use]
pub fn kv_arc(mut self, kv: Arc<dyn KeyValue>) -> Self {
self.kv = Some(kv);
self
}
#[must_use]
pub fn blob_arc(mut self, blob: Arc<dyn Blob>) -> Self {
self.blob = Some(blob);
self
}
#[must_use]
pub fn push_arc(mut self, push: Arc<dyn Push>) -> Self {
self.push = Some(push);
self
}
#[cfg(feature = "push")]
#[must_use]
pub fn push_from_env(mut self) -> Self {
self.push_from_env = true;
self
}
#[cfg(feature = "push")]
fn env_push(&self) -> Option<&(Arc<dyn Push>, cratefield_push_wiring::PushWiring)> {
if !self.push_from_env || self.push.is_some() {
return None;
}
Some(self.assembled_push.get_or_init(|| {
let clock: Arc<dyn Clock> = Arc::new(TokioClock);
let http: Arc<dyn cratefield_core::HttpClient> = Arc::new(BoundedHttpClient::new(
Arc::new(ReqwestClient::new()),
Arc::clone(&clock),
));
cratefield_push_wiring::build_push(&EnvConfig, &http, &clock)
}))
}
#[cfg(feature = "push")]
#[must_use]
pub fn push_wiring(&self) -> Option<&cratefield_push_wiring::PushWiring> {
self.env_push().map(|(_, wiring)| wiring)
}
#[must_use]
pub fn payments_arc(mut self, payments: Arc<dyn Payments>) -> Self {
self.payments = Some(payments);
self
}
#[must_use]
pub fn tracker_arc(mut self, tracker: Arc<dyn Tracker>) -> Self {
self.tracker = Some(tracker);
self
}
#[must_use]
pub fn realtime_arc(mut self, realtime: Arc<dyn Realtime>) -> Self {
self.realtime = Some(realtime);
self
}
#[must_use]
pub fn text_model(mut self, text_model: impl TextModel + 'static) -> Self {
self.text_model = Some(Arc::new(text_model));
self
}
#[must_use]
pub fn text_model_arc(mut self, text_model: Arc<dyn TextModel>) -> Self {
self.text_model = Some(text_model);
self
}
#[must_use]
pub fn classifier(mut self, classifier: impl Classifier + 'static) -> Self {
self.classifier = Some(Arc::new(classifier));
self
}
#[must_use]
pub fn classifier_arc(mut self, classifier: Arc<dyn Classifier>) -> Self {
self.classifier = Some(classifier);
self
}
#[must_use]
pub fn vector_index_arc(mut self, index: Arc<dyn VectorIndex>) -> Self {
self.vector_index = Some(index);
self
}
#[must_use]
pub fn embedder_arc(mut self, embedder: Arc<dyn Embedder>) -> Self {
self.embedder = Some(embedder);
self
}
#[must_use]
pub fn custom_hostnames(mut self, custom_hostnames: impl CustomHostnames + 'static) -> Self {
self.custom_hostnames = Some(Arc::new(custom_hostnames));
self
}
#[must_use]
pub fn custom_hostnames_arc(mut self, custom_hostnames: Arc<dyn CustomHostnames>) -> Self {
self.custom_hostnames = Some(custom_hostnames);
self
}
#[must_use]
pub fn mailer(mut self, mailer: impl Mailer + 'static) -> Self {
self.mailer = Some(Arc::new(mailer));
self
}
#[must_use]
pub fn mailer_arc(mut self, mailer: Arc<dyn Mailer>) -> Self {
self.mailer = Some(mailer);
self
}
#[must_use]
pub fn captcha(mut self, captcha: impl Captcha + 'static) -> Self {
self.captcha = Some(Arc::new(captcha));
self
}
#[must_use]
pub fn captcha_arc(mut self, captcha: Arc<dyn Captcha>) -> Self {
self.captcha = Some(captcha);
self
}
#[must_use]
pub fn auth(mut self, auth: impl Auth + 'static) -> Self {
self.auth = Some(Arc::new(auth));
self
}
#[must_use]
pub fn auth_arc(mut self, auth: Arc<dyn Auth>) -> Self {
self.auth = Some(auth);
self
}
#[must_use]
pub fn auth_from_env(mut self) -> Self {
self.auth_from_env = true;
self
}
#[must_use]
pub fn ports(&self) -> Ports {
let config: Arc<dyn cratefield_core::Config> = Arc::new(EnvConfig);
let mut ports = Ports::with_config(Arc::clone(&config));
ports.db.clone_from(&self.db);
ports.rate_limiter.clone_from(&self.rate_limiter);
ports.kv.clone_from(&self.kv);
ports.blob.clone_from(&self.blob);
ports.push.clone_from(&self.push);
ports.payments.clone_from(&self.payments);
ports.tracker.clone_from(&self.tracker);
ports.realtime.clone_from(&self.realtime);
ports.text_model.clone_from(&self.text_model);
ports.classifier.clone_from(&self.classifier);
ports.vector_index.clone_from(&self.vector_index);
ports.embedder.clone_from(&self.embedder);
ports.custom_hostnames.clone_from(&self.custom_hostnames);
ports.mailer.clone_from(&self.mailer);
ports.captcha.clone_from(&self.captcha);
ports.auth.clone_from(&self.auth);
#[cfg(feature = "push")]
if ports.push.is_none()
&& let Some((push, _)) = self.env_push()
{
ports.push = Some(Arc::clone(push));
}
match HarnessConfig::from_config(&*config) {
Ok(parsed) => {
cratefield_core::set_log_pseudonym_key(parsed.harness_secret.as_bytes());
ports.signer = Some(Arc::new(parsed.signer()));
}
Err(err) => warn_once(&WARNED_SIGNER, &format!("Signer port not provided: {err}")),
}
let clock: Arc<dyn Clock> = Arc::new(TokioClock);
ports.clock = Some(Arc::clone(&clock));
let http: Arc<dyn cratefield_core::HttpClient> = Arc::new(BoundedHttpClient::new(
Arc::new(ReqwestClient::new()),
Arc::clone(&clock),
));
ports.http = Some(Arc::clone(&http));
if ports.auth.is_none() && self.auth_from_env {
ports.auth = Some(cratefield_auth_client::from_config(
ports.config.as_ref(),
http,
clock,
));
}
ports.id_gen = Some(Arc::new(UlidIdGen));
ports.defer = Some(Arc::new(SpawnDefer));
ports
}
}
impl Runtime for Native {
fn provides(&self) -> Vec<Port> {
let mut provided = vec![
Port::Signer,
Port::HttpClient,
Port::Clock,
Port::IdGen,
Port::Defer,
];
if self.db.is_some() {
provided.push(Port::Db);
}
if self.auth.is_some() || self.auth_from_env {
provided.push(Port::Auth);
}
if self.rate_limiter.is_some() {
provided.push(Port::RateLimiter);
}
if self.kv.is_some() {
provided.push(Port::KeyValue);
}
if self.blob.is_some() {
provided.push(Port::Blob);
}
#[cfg(feature = "push")]
let push_provided = self.push.is_some() || self.push_from_env;
#[cfg(not(feature = "push"))]
let push_provided = self.push.is_some();
if push_provided {
provided.push(Port::Push);
}
if self.payments.is_some() {
provided.push(Port::Payments);
}
if self.tracker.is_some() {
provided.push(Port::Tracker);
}
if self.realtime.is_some() {
provided.push(Port::Realtime);
}
if self.text_model.is_some() {
provided.push(Port::TextModel);
}
if self.classifier.is_some() {
provided.push(Port::Classifier);
}
if self.vector_index.is_some() {
provided.push(Port::VectorIndex);
}
if self.embedder.is_some() {
provided.push(Port::Embedder);
}
if self.custom_hostnames.is_some() {
provided.push(Port::CustomHostnames);
}
if self.mailer.is_some() {
provided.push(Port::Mailer);
}
if self.captcha.is_some() {
provided.push(Port::Captcha);
}
provided
}
fn effectively_configured(&self, port: Port) -> bool {
match port {
Port::Captcha => self.captcha.as_ref().is_some_and(|captcha| {
captcha
.binding()
.is_none_or(|binding| binding.hostname_bound && !binding.fail_open)
}),
_ => self.provides().contains(&port),
}
}
}
pub(crate) fn clone_ports(ports: &Ports) -> Ports {
let mut snapshot = Ports::with_config(Arc::clone(&ports.config));
snapshot.db.clone_from(&ports.db);
snapshot.auth.clone_from(&ports.auth);
snapshot.mailer.clone_from(&ports.mailer);
snapshot.captcha.clone_from(&ports.captcha);
snapshot.rate_limiter.clone_from(&ports.rate_limiter);
snapshot.signer.clone_from(&ports.signer);
snapshot.kv.clone_from(&ports.kv);
snapshot.blob.clone_from(&ports.blob);
snapshot.push.clone_from(&ports.push);
snapshot.payments.clone_from(&ports.payments);
snapshot.tracker.clone_from(&ports.tracker);
snapshot.realtime.clone_from(&ports.realtime);
snapshot.text_model.clone_from(&ports.text_model);
snapshot.classifier.clone_from(&ports.classifier);
snapshot.vector_index.clone_from(&ports.vector_index);
snapshot.embedder.clone_from(&ports.embedder);
snapshot
.custom_hostnames
.clone_from(&ports.custom_hostnames);
snapshot.http.clone_from(&ports.http);
snapshot.clock.clone_from(&ports.clock);
snapshot.id_gen.clone_from(&ports.id_gen);
snapshot.defer.clone_from(&ports.defer);
snapshot.url_safety.clone_from(&ports.url_safety);
snapshot.actors.clone_from(&ports.actors);
snapshot
}
#[cfg(test)]
mod tests {
use super::clone_ports;
use cratefield_core::{Port, Runtime};
use std::sync::Arc;
#[test]
fn clone_ports_carries_every_port() {
let source = cratefield_testing::full_fake_ports();
assert!(
Port::ALL.iter().all(|port| source.has(*port)),
"the source bundle must wire every port for this to prove anything"
);
let snapshot = clone_ports(&source);
for port in Port::ALL {
assert!(snapshot.has(*port), "clone_ports drops {}", port.name());
}
}
#[test]
fn the_vector_ports_are_provided_only_when_wired() {
let wired = super::Native::new()
.vector_index_arc(Arc::new(cratefield_core::ExactVectorIndex::new(3)))
.embedder_arc(Arc::new(cratefield_testing::FakeEmbedder));
assert!(wired.provides().contains(&Port::VectorIndex));
assert!(wired.provides().contains(&Port::Embedder));
let bare = super::Native::new();
assert!(!bare.provides().contains(&Port::VectorIndex));
assert!(!bare.provides().contains(&Port::Embedder));
}
#[test]
fn custom_hostnames_is_provided_only_when_wired() {
let wired = super::Native::new().custom_hostnames(
cratefield_testing::FakeCustomHostnames::new("cratefield.app"),
);
assert!(wired.provides().contains(&Port::CustomHostnames));
let bare = super::Native::new();
assert!(!bare.provides().contains(&Port::CustomHostnames));
}
mod warn_once_race {
use super::super::warn_once;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Barrier, OnceLock};
const MARKER: &str = "warn-once-race-marker";
struct Counter(Arc<AtomicUsize>);
impl tracing::Subscriber for Counter {
fn enabled(&self, _metadata: &tracing::Metadata<'_>) -> bool {
true
}
fn new_span(&self, _attributes: &tracing::span::Attributes<'_>) -> tracing::span::Id {
tracing::span::Id::from_u64(1)
}
fn record(&self, _id: &tracing::span::Id, _values: &tracing::span::Record<'_>) {}
fn record_follows_from(&self, _id: &tracing::span::Id, _follows: &tracing::span::Id) {}
fn event(&self, event: &tracing::Event<'_>) {
struct Marker<'a>(&'a str, &'a AtomicUsize);
impl tracing::field::Visit for Marker<'_> {
fn record_debug(
&mut self,
field: &tracing::field::Field,
value: &dyn std::fmt::Debug,
) {
if field.name() == "message" && format!("{value:?}").contains(self.0) {
self.1.fetch_add(1, Ordering::SeqCst);
}
}
}
event.record(&mut Marker(MARKER, &self.0));
}
fn enter(&self, _id: &tracing::span::Id) {}
fn exit(&self, _id: &tracing::span::Id) {}
}
fn seen() -> &'static Arc<AtomicUsize> {
static SEEN: OnceLock<Arc<AtomicUsize>> = OnceLock::new();
SEEN.get_or_init(|| {
let seen = Arc::new(AtomicUsize::new(0));
let _ = tracing::subscriber::set_global_default(Counter(Arc::clone(&seen)));
seen
})
}
#[test]
fn a_warn_once_flag_is_claimed_by_exactly_one_thread() {
let seen = seen();
for _ in 0..200 {
let flag = Arc::new(AtomicBool::new(false));
let gate = Arc::new(Barrier::new(32));
let threads: Vec<_> = (0..32)
.map(|_| {
let gate = Arc::clone(&gate);
let flag = Arc::clone(&flag);
std::thread::spawn(move || {
gate.wait();
warn_once(&flag, MARKER);
})
})
.collect();
for thread in threads {
thread.join().expect("thread");
}
let logged = seen.swap(0, Ordering::SeqCst);
assert_eq!(logged, 1, "{logged} threads logged a once-only warning");
}
}
}
}