saddle-core 0.3.26

//! Request facts only. Allocation permission and cancellation remain with Runtime/Admission.
//! Root creation and each local allocation must be preceded by their storage reservation.
use crate::{CallContext, DbScopeDiagnosticIdentity};
use serde::{Serialize, Serializer, ser::SerializeStruct};
use std::sync::{
    Arc, OnceLock,
    atomic::{AtomicU64, Ordering},
};

static NEXT_ROOT: AtomicU64 = AtomicU64::new(1);

/// Closed absence vocabulary, separate from output availability.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(tag = "state", content = "value", rename_all = "snake_case")]
pub enum ContextFact<T> {
    Present(T),
    NotApplicable,
    NotEstablished,
    Unavailable,
}

/// Exact bounded protocol identity. Not an authorization or a user-data container.
#[derive(Clone, Copy, Eq, PartialEq)]
pub struct ContextIdentity {
    bytes: [u8; 256],
    len: u16,
}
impl ContextIdentity {
    pub fn checked(value: &str) -> Result<Self, ContextConflict> {
        if value.is_empty() || value.len() > 256 || value.chars().any(char::is_control) {
            return Err(ContextConflict::InvalidIdentity);
        }
        let mut out = Self {
            bytes: [0; 256],
            len: value.len() as u16,
        };
        out.bytes[..value.len()].copy_from_slice(value.as_bytes());
        Ok(out)
    }
    fn as_str(&self) -> &str {
        std::str::from_utf8(&self.bytes[..usize::from(self.len)]).expect("validated UTF-8")
    }
}
impl Serialize for ContextIdentity {
    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        s.serialize_str(self.as_str())
    }
}

/// Safe metadata, bounded and exact; credentials/URL/query syntax is rejected.
#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
#[serde(transparent)]
pub struct ContextLabel(ContextIdentity);
impl ContextLabel {
    pub fn checked(value: &str) -> Result<Self, ContextConflict> {
        let value = ContextIdentity::checked(value)?;
        if value.as_str().contains("://")
            || !value
                .as_str()
                .chars()
                .all(|c| c.is_alphanumeric() || "_.:/{}*-".contains(c))
        {
            return Err(ContextConflict::UnsafeMetadata);
        }
        Ok(Self(value))
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ContextConflict {
    InvalidIdentity,
    UnsafeMetadata,
    IdentityGroup,
    ForeignRoot,
    ChildRelation,
    CounterExhausted,
}

/// Complete validated identity group. Consumed atomically by the entry publisher.
/// No partial-field setters; rejected publication returns this same group.
#[derive(Eq, PartialEq)]
pub struct RequestIdentityGroup {
    application: ContextLabel,
    module: ContextLabel,
    service: ContextLabel,
    operation: ContextLabel,
    trace: ContextIdentity,
    rpc: ContextFact<ContextIdentity>,
    span: u64,
    request: ContextIdentity,
    route: ContextLabel,
    attempt: u32,
    zone: ContextFact<ContextLabel>,
}
impl RequestIdentityGroup {
    pub fn from_validated(
        call: &CallContext,
        request: &str,
        route: &str,
        attempt: u32,
        zone: ContextFact<ContextLabel>,
    ) -> Result<Self, ContextConflict> {
        if attempt == 0 {
            return Err(ContextConflict::InvalidIdentity);
        }
        Ok(Self {
            application: ContextLabel::checked(call.application().as_str())?,
            module: ContextLabel::checked(call.module().as_str())?,
            service: ContextLabel::checked(call.service().as_str())?,
            operation: ContextLabel::checked(call.operation().as_str())?,
            trace: ContextIdentity::checked(call.trace_correlation_id().as_str())?,
            rpc: match call.rpc_correlation_id() {
                Some(id) => ContextFact::Present(ContextIdentity::checked(id.as_str())?),
                None => ContextFact::Unavailable,
            },
            span: call.span_id().as_u64(),
            request: ContextIdentity::checked(request)?,
            route: ContextLabel::checked(route)?,
            attempt,
            zone,
        })
    }
}

struct RequestRoot {
    local: u64,
    application: ContextLabel,
    initial: ContextFact<()>,
    identity: OnceLock<RequestIdentityGroup>,
}
impl Drop for RequestRoot {
    fn drop(&mut self) {
        observe(self.local, "root_drop");
    }
}

/// Unique entry binding capability. Does not own an account or output handle.
///
/// ```compile_fail
/// use saddle_core::RequestRootPublisher;
/// fn duplicate(p: RequestRootPublisher) { let _ = p.clone(); }
/// ```
pub struct RequestRootPublisher {
    root: Arc<RequestRoot>,
}

/// Read-only shared reference; cloning shares one allocation, never root fields.
pub struct RequestRootRef {
    root: Arc<RequestRoot>,
}
impl Clone for RequestRootRef {
    fn clone(&self) -> Self {
        observe(self.root.local, "root_share");
        Self {
            root: Arc::clone(&self.root),
        }
    }
}
impl Drop for RequestRootRef {
    fn drop(&mut self) {
        observe(self.root.local, "root_release");
    }
}

impl RequestRootPublisher {
    /// Storage-only constructor, not admission. Caller must reserve before calling;
    /// C supplies layout, R0/A/R supply and enforce the actual reservation seam.
    pub fn create(
        application: ContextLabel,
        initial: ContextFact<()>,
    ) -> Result<Self, ContextConflict> {
        if matches!(initial, ContextFact::Present(())) {
            return Err(ContextConflict::InvalidIdentity);
        }
        let local = NEXT_ROOT
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| n.checked_add(1))
            .map_err(|_| ContextConflict::CounterExhausted)?;
        let root = Arc::new(RequestRoot {
            local,
            application,
            initial,
            identity: OnceLock::new(),
        });
        observe(local, "root_create");
        Ok(Self { root })
    }
    pub fn reference(&self) -> RequestRootRef {
        observe(self.root.local, "root_share");
        RequestRootRef {
            root: Arc::clone(&self.root),
        }
    }
    #[allow(clippy::result_large_err)] // Recover the fixed input without allocating on rejection.
    pub fn publish(
        &mut self,
        group: RequestIdentityGroup,
    ) -> Result<(), (ContextConflict, RequestIdentityGroup)> {
        if group.application != self.root.application {
            return Err((ContextConflict::IdentityGroup, group));
        }
        if let Some(old) = self.root.identity.get() {
            return if old == &group {
                Ok(())
            } else {
                Err((ContextConflict::IdentityGroup, group))
            };
        }
        self.root
            .identity
            .set(group)
            .map_err(|group| (ContextConflict::IdentityGroup, group))
    }
}

/// Closed lifecycle metadata, not a state machine controlling the request.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum RequestViewPhase {
    SocketAccepted,
    Reading,
    Admitted,
    Dispatch,
    Handler,
    Database,
    Outbound,
    Response,
    Finalizing,
    Finished,
}

/// Static registered operation labels are borrowed forever, never copied as text.
#[derive(Clone, Copy, Eq, PartialEq, Serialize)]
pub struct RegisteredContextOperation(&'static str);
impl RegisteredContextOperation {
    pub fn checked(value: &'static str) -> Result<Self, ContextConflict> {
        ContextLabel::checked(value)?;
        Ok(Self(value))
    }
}

#[derive(Clone, Copy)]
pub struct RequestLocalFacts {
    task: ContextFact<u64>,
    scope: ContextFact<DbScopeDiagnosticIdentity>,
    db_operation: ContextFact<RegisteredContextOperation>,
    phase: RequestViewPhase,
}
impl RequestLocalFacts {
    pub fn new(phase: RequestViewPhase) -> Self {
        Self {
            task: ContextFact::NotEstablished,
            scope: ContextFact::NotEstablished,
            db_operation: ContextFact::NotApplicable,
            phase,
        }
    }
    /// Numeric observation supplied by the real task owner; not execution authority.
    pub fn with_task(mut self, task: ContextFact<u64>) -> Self {
        self.task = task;
        self
    }
    pub fn without_scope(mut self, state: ContextFact<()>) -> Self {
        self.scope = absent(state);
        self
    }
    pub fn with_db_operation(mut self, operation: RegisteredContextOperation) -> Self {
        self.db_operation = ContextFact::Present(operation);
        self
    }
}

// Only a real child needs variable local identities. Sibling views share this
// allocation; the root's original request/trace/zone never appear in this object.
struct ChildCall {
    #[cfg(test)]
    observation: LocalObservation,
    application: ContextLabel,
    module: ContextLabel,
    service: ContextLabel,
    operation: ContextLabel,
    rpc: ContextIdentity,
    span: u64,
    route: ContextLabel,
    attempt: u32,
}
struct LocalView {
    #[cfg(test)]
    observation: LocalObservation,
    root: Arc<RequestRoot>,
    published: bool,
    facts: RequestLocalFacts,
    child: Option<Arc<ChildCall>>,
}
impl Drop for LocalView {
    fn drop(&mut self) {
        observe(self.root.local, "view_drop");
        #[cfg(test)]
        self.observation.record("destroy");
    }
}
impl Drop for ChildCall {
    fn drop(&mut self) {
        #[cfg(test)]
        self.observation.record("destroy");
    }
}

/// Immutable local facts + a frozen visibility stage, sharing the single root.
/// Source receipts retain this object, not another projection or diagnostic body.
///
/// ```compile_fail
/// use saddle_core::RequestExecutionView;
/// fn rewrite(view: RequestExecutionView) { view.inner.published = false; }
/// ```
pub struct RequestExecutionView {
    inner: Arc<LocalView>,
}
impl Clone for RequestExecutionView {
    fn clone(&self) -> Self {
        #[cfg(test)]
        self.inner.observation.record("share");
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}
impl Drop for RequestExecutionView {
    fn drop(&mut self) {
        #[cfg(test)]
        self.inner.observation.record("release");
    }
}
impl RequestRootRef {
    pub fn view(&self, facts: RequestLocalFacts) -> RequestExecutionView {
        RequestExecutionView::allocate(
            Arc::clone(&self.root),
            self.root.identity.get().is_some(),
            facts,
            None,
        )
    }
    pub fn same_request(&self, view: &RequestExecutionView) -> bool {
        Arc::ptr_eq(&self.root, &view.inner.root)
    }
}
impl RequestExecutionView {
    fn allocate(
        root: Arc<RequestRoot>,
        published: bool,
        facts: RequestLocalFacts,
        child: Option<Arc<ChildCall>>,
    ) -> Self {
        observe(root.local, "view_create");
        Self {
            inner: Arc::new(LocalView {
                #[cfg(test)]
                observation: LocalObservation::create(root.local, "view"),
                root,
                published,
                facts,
                child,
            }),
        }
    }
    /// New operation, no root-text copy or mutation of an earlier view.
    fn local(&self, facts: RequestLocalFacts) -> Self {
        Self::allocate(
            Arc::clone(&self.inner.root),
            self.inner.published,
            facts,
            self.inner.child.clone(),
        )
    }
    pub fn with_phase(&self, phase: RequestViewPhase) -> Self {
        let mut facts = self.inner.facts;
        facts.phase = phase;
        self.local(facts)
    }
    pub fn with_db_operation(&self, operation: RegisteredContextOperation) -> Self {
        self.local(self.inner.facts.with_db_operation(operation))
    }
    /// Derive the observation for the actual task selected by the runtime. The
    /// numeric task is metadata, never a task permit or a sequence minted here.
    pub fn in_task(&self, task: u64) -> Self {
        self.local(self.inner.facts.with_task(ContextFact::Present(task)))
    }
    pub fn in_db_scope(
        &self,
        scope: &crate::DbScopeDiagnosticContext<Self>,
    ) -> Result<Self, ContextConflict> {
        let (original, identity) = scope.diagnostic_context();
        if !self.same_request(original) {
            return Err(ContextConflict::ForeignRoot);
        }
        let mut facts = self.inner.facts;
        facts.scope = ContextFact::Present(identity);
        Ok(self.local(facts))
    }
    pub fn observed_db_scope(
        &self,
        scope: &crate::DbScopeObservation<Self>,
    ) -> Result<Self, ContextConflict> {
        let (original, identity) = scope.diagnostic_context();
        if !self.same_request(original) {
            return Err(ContextConflict::ForeignRoot);
        }
        let mut facts = self.inner.facts;
        facts.scope = ContextFact::Present(identity);
        Ok(self.local(facts))
    }
    pub fn same_request(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.inner.root, &other.inner.root)
    }
    pub fn same_view(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.inner, &other.inner)
    }
    /// Only the existing entry root can authorize a later visibility stage.
    pub fn refresh(&self, root: &RequestRootRef) -> Result<Self, ContextConflict> {
        if !root.same_request(self) {
            return Err(ContextConflict::ForeignRoot);
        }
        Ok(Self::allocate(
            Arc::clone(&self.inner.root),
            self.inner.root.identity.get().is_some(),
            self.inner.facts,
            self.inner.child.clone(),
        ))
    }
    pub fn child(
        &self,
        call: &CallContext,
        request: &str,
        route: &str,
        attempt: u32,
    ) -> Result<Self, ContextConflict> {
        let identity = self.identity().ok_or(ContextConflict::ChildRelation)?;
        if identity.trace.as_str() != call.trace_correlation_id().as_str()
            || identity.request.as_str() != request
        {
            return Err(ContextConflict::ForeignRoot);
        }
        let parent_rpc = if let Some(child) = &self.inner.child {
            &child.rpc
        } else if let ContextFact::Present(rpc) = &identity.rpc {
            rpc
        } else {
            return Err(ContextConflict::ChildRelation);
        };
        let rpc = call
            .rpc_correlation_id()
            .ok_or(ContextConflict::ChildRelation)?;
        let suffix = rpc
            .as_str()
            .strip_prefix(parent_rpc.as_str())
            .and_then(|s| s.strip_prefix('.'))
            .ok_or(ContextConflict::ChildRelation)?;
        let span = self
            .inner
            .child
            .as_ref()
            .map_or(identity.span, |child| child.span);
        if suffix.is_empty()
            || !suffix.bytes().all(|b| b.is_ascii_digit())
            || span == call.span_id().as_u64()
            || attempt == 0
        {
            return Err(ContextConflict::ChildRelation);
        }
        let child = Arc::new(ChildCall {
            application: ContextLabel::checked(call.application().as_str())?,
            module: ContextLabel::checked(call.module().as_str())?,
            service: ContextLabel::checked(call.service().as_str())?,
            operation: ContextLabel::checked(call.operation().as_str())?,
            rpc: ContextIdentity::checked(rpc.as_str())?,
            span: call.span_id().as_u64(),
            route: ContextLabel::checked(route)?,
            attempt,
            #[cfg(test)]
            observation: LocalObservation::create(self.inner.root.local, "child"),
        });
        Ok(Self::allocate(
            Arc::clone(&self.inner.root),
            self.inner.published,
            self.inner.facts,
            Some(child),
        ))
    }
    fn identity(&self) -> Option<&RequestIdentityGroup> {
        self.inner
            .published
            .then(|| self.inner.root.identity.get())
            .flatten()
    }
}

fn absent<T>(state: ContextFact<()>) -> ContextFact<T> {
    match state {
        ContextFact::NotApplicable => ContextFact::NotApplicable,
        ContextFact::NotEstablished => ContextFact::NotEstablished,
        _ => ContextFact::Unavailable,
    }
}
impl Serialize for RequestExecutionView {
    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
        let mut out = s.serialize_struct("RequestContext", 21)?;
        let identity = self.identity();
        let child = self.inner.child.as_deref();
        let initial = self.inner.root.initial;
        out.serialize_field("schema_version", &2u8)?;
        out.serialize_field("local_request", &self.inner.root.local)?;
        out.serialize_field("publication", &u8::from(self.inner.published))?;
        out.serialize_field(
            "application",
            &ContextFact::Present(&self.inner.root.application),
        )?;
        out.serialize_field(
            "call_application",
            &ContextFact::Present(child.map_or(&self.inner.root.application, |c| &c.application)),
        )?;
        macro_rules! root_field {
            ($name:literal, $field:ident) => {
                out.serialize_field(
                    $name,
                    &identity.map_or_else(|| absent(initial), |g| ContextFact::Present(&g.$field)),
                )?;
            };
        }
        macro_rules! call_field {
            ($name:literal, $field:ident) => {
                out.serialize_field(
                    $name,
                    &child
                        .map(|c| &c.$field)
                        .or_else(|| identity.map(|g| &g.$field))
                        .map_or_else(|| absent(initial), ContextFact::Present),
                )?;
            };
        }
        call_field!("module", module);
        call_field!("service", service);
        call_field!("operation", operation);
        root_field!("trace_id", trace);
        root_field!("request", request);
        let span = child.map(|c| c.span).or_else(|| identity.map(|g| g.span));
        out.serialize_field(
            "span_id",
            &span.map_or_else(
                || absent(initial),
                |s| ContextFact::Present(SpanProjection(s)),
            ),
        )?;
        call_field!("route", route);
        call_field!("attempt", attempt);
        let rpc = child
            .map(|c| ContextFact::Present(c.rpc))
            .or_else(|| identity.map(|g| g.rpc))
            .unwrap_or_else(|| absent(initial));
        out.serialize_field("rpc_id", &rpc)?;
        out.serialize_field(
            "zone",
            &identity.map_or_else(|| absent(initial), |g| g.zone),
        )?;
        out.serialize_field("db_operation", &self.inner.facts.db_operation)?;
        out.serialize_field("scope", &self.inner.facts.scope)?;
        out.serialize_field("task", &self.inner.facts.task)?;
        out.serialize_field("lifecycle", &ContextFact::Present(self.inner.facts.phase))?;
        // Target is the registered route, never endpoint/URL or credentials.
        out.serialize_field(
            "target",
            &child.map_or(ContextFact::NotApplicable, |c| {
                ContextFact::Present(&c.route)
            }),
        )?;
        out.end()
    }
}

/// Payload and shared allocation layouts, not a reservation or full task charge.
/// Arc control block calculation follows std's two-AtomicUsize header, with
/// Layout::extend padding. Allocator metadata is NOT included; R0 must account it.
pub fn request_context_layouts() -> [(std::alloc::Layout, std::alloc::Layout); 3] {
    fn pair<T>() -> (std::alloc::Layout, std::alloc::Layout) {
        let payload = std::alloc::Layout::new::<T>();
        let header = std::alloc::Layout::new::<[std::sync::atomic::AtomicUsize; 2]>();
        (
            payload,
            header
                .extend(payload)
                .expect("fixed layout")
                .0
                .pad_to_align(),
        )
    }
    [
        pair::<RequestRoot>(),
        pair::<LocalView>(),
        pair::<ChildCall>(),
    ]
}

#[cfg(not(test))]
fn observe(_: u64, _: &'static str) {}
#[cfg(test)]
fn observe(root: u64, event: &'static str) {
    EVENTS.lock().unwrap().push((root, event));
    record_observation(root, root, "root", event);
}
#[cfg(test)]
static EVENTS: std::sync::Mutex<Vec<(u64, &'static str)>> = std::sync::Mutex::new(Vec::new());

struct SpanProjection(u64);
impl Serialize for SpanProjection {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let mut bytes = [b'0'; 16];
        for (i, byte) in bytes.iter_mut().enumerate() {
            *byte = b"0123456789abcdef"[((self.0 >> ((15 - i) * 4)) & 15) as usize];
        }
        serializer.serialize_str(std::str::from_utf8(&bytes).expect("hex"))
    }
}

// Private observation holds numbers only, never Arc/Weak, callbacks or objects.
// Instrumentation storage is test-only and excluded from production layouts.
#[cfg(test)]
struct LocalObservation {
    id: u64,
    root: u64,
    kind: &'static str,
}
#[cfg(test)]
static LOCAL_EVENTS: std::sync::Mutex<Vec<(u64, u64, &'static str, &'static str)>> =
    std::sync::Mutex::new(Vec::new());
#[cfg(test)]
impl LocalObservation {
    fn create(root: u64, kind: &'static str) -> Self {
        static NEXT: AtomicU64 = AtomicU64::new(1);
        let observation = Self {
            id: NEXT.fetch_add(1, Ordering::Relaxed),
            root,
            kind,
        };
        observation.record("create");
        observation
    }
    fn record(&self, event: &'static str) {
        LOCAL_EVENTS
            .lock()
            .unwrap()
            .push((self.id, self.root, self.kind, event));
        record_observation(self.root, self.id, self.kind, event);
    }
}

#[cfg(test)]
type ObservationEvent = (u64, u64, u64, &'static str, &'static str);
#[cfg(test)]
static ORDERED_EVENTS: std::sync::Mutex<Vec<ObservationEvent>> = std::sync::Mutex::new(Vec::new());
#[cfg(test)]
fn record_observation(root: u64, object: u64, kind: &'static str, event: &'static str) {
    static SEQUENCE: AtomicU64 = AtomicU64::new(1);
    let sequence = SEQUENCE.fetch_add(1, Ordering::Relaxed);
    ORDERED_EVENTS
        .lock()
        .unwrap()
        .push((sequence, root, object, kind, event));
}

#[cfg(test)]
mod tests;