batpak 0.10.0

/// Positional types for locating events within a DAG chain.
pub mod position;
pub use position::DagPosition;

use crate::event::EventKind;
use batpak_macros::Error;
use serde::{Deserialize, Serialize};
use std::fmt;
use std::sync::Arc;

/// Hard cap for each coordinate component. Prevents accidental or hostile
/// cardinality bombs from turning entity/scope keys into unbounded memory sinks.
pub const MAX_COORDINATE_COMPONENT_LEN: usize = 1024;

/// Coordinate: WHO (entity) + WHERE (scope). The address of an event stream.

#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize)]
#[serde(into = "CoordinateWire")]
pub struct Coordinate {
    entity: Arc<str>, // WHO — stream key, hash chain anchor
    scope: Arc<str>,  // WHERE — isolation boundary
}

/// Wire form of [`Coordinate`] used by serde so that every deserialised
/// value routes back through [`Coordinate::new`] and picks up the same
/// validation as in-process construction.
#[derive(Serialize, Deserialize)]
struct CoordinateWire {
    entity: String,
    scope: String,
}

impl From<Coordinate> for CoordinateWire {
    fn from(coord: Coordinate) -> Self {
        Self {
            entity: coord.entity.as_ref().to_owned(),
            scope: coord.scope.as_ref().to_owned(),
        }
    }
}

impl<'de> Deserialize<'de> for Coordinate {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let wire = CoordinateWire::deserialize(deserializer)?;
        Coordinate::new(&wire.entity, &wire.scope).map_err(serde::de::Error::custom)
    }
}

/// Errors returned when constructing a [`Coordinate`].
#[derive(Clone, Debug, PartialEq, Eq, Error)]
#[non_exhaustive]
pub enum CoordinateError {
    /// The entity string was empty.
    #[error("entity cannot be empty")]
    EmptyEntity,
    /// The scope string was empty.
    #[error("scope cannot be empty")]
    EmptyScope,
    /// The entity string exceeded the maximum supported length.
    #[error("entity length {len} exceeds maximum {max}")]
    EntityTooLong {
        /// Actual entity string length.
        len: usize,
        /// Maximum permitted length.
        max: usize,
    },
    /// The scope string exceeded the maximum supported length.
    #[error("scope length {len} exceeds maximum {max}")]
    ScopeTooLong {
        /// Actual scope string length.
        len: usize,
        /// Maximum permitted length.
        max: usize,
    },
    /// A coordinate component contained a NUL byte (`'\0'`).
    #[error("coordinate component contains a NUL byte")]
    NulByte,
    /// A coordinate component contained a forbidden ASCII control character.
    #[error("coordinate component contains a forbidden ASCII control character")]
    ControlChar,
    /// A coordinate component contained a path-traversal substring (`..` or `/`).
    #[error("coordinate component contains a forbidden path-traversal substring (`..` or `/`)")]
    PathTraversal,
    /// A coordinate component contained a checkpoint identity separator (`|` or `=`).
    #[error("coordinate component contains a forbidden identity-separator character (`|` or `=`)")]
    ForbiddenSeparator,
}

/// Errors returned when constructing a region filter component.
#[derive(Clone, Debug, PartialEq, Eq, Error)]
#[non_exhaustive]
pub enum RegionFilterError {
    /// A clock range whose start exceeds its end.
    #[error("clock range start {start} exceeds end {end}")]
    InvertedClockRange {
        /// Inclusive lower bound supplied by the caller.
        start: u32,
        /// Inclusive upper bound supplied by the caller.
        end: u32,
    },
    /// An event category outside the valid 4-bit range (`0..16`).
    #[error("event category {category} is out of range (must be < 16)")]
    CategoryOutOfRange {
        /// Out-of-range category supplied by the caller.
        category: u8,
    },
}

/// Inclusive per-entity clock range used as a region filter.
///
/// Bounds are per-entity logical clocks, not global sequences, and do not apply
/// to live filtering.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ClockRange {
    start: u32,
    end: u32,
}

impl ClockRange {
    /// Construct an inclusive clock range.
    ///
    /// # Errors
    /// Returns [`RegionFilterError::InvertedClockRange`] when `start > end`.
    pub fn new(start: u32, end: u32) -> Result<Self, RegionFilterError> {
        if start > end {
            return Err(RegionFilterError::InvertedClockRange { start, end });
        }
        Ok(Self { start, end })
    }

    /// Inclusive lower bound.
    #[must_use]
    pub fn start(&self) -> u32 {
        self.start
    }

    /// Inclusive upper bound.
    #[must_use]
    pub fn end(&self) -> u32 {
        self.end
    }

    pub(crate) fn as_tuple(&self) -> (u32, u32) {
        (self.start, self.end)
    }
}

/// A 4-bit event category used as a region filter.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct EventCategory(u8);

impl EventCategory {
    /// Construct a category, validating that it fits in 4 bits (`0..16`).
    ///
    /// # Errors
    /// Returns [`RegionFilterError::CategoryOutOfRange`] when `category >= 16`.
    pub fn new(category: u8) -> Result<Self, RegionFilterError> {
        if category >= 16 {
            return Err(RegionFilterError::CategoryOutOfRange { category });
        }
        Ok(Self(category))
    }

    /// The category of a concrete [`EventKind`].
    #[must_use]
    pub fn of_kind(kind: EventKind) -> Self {
        Self(kind.category())
    }

    /// The raw 4-bit category value.
    #[must_use]
    pub fn get(&self) -> u8 {
        self.0
    }
}

/// Region: the ONE predicate type for query, subscription, cursor, traversal.
#[derive(Clone, Debug, Default)]
pub struct Region {
    /// Optional entity name prefix; matches any entity whose name starts with this string.
    pub(crate) entity_prefix: Option<Arc<str>>,
    /// Optional exact scope to match.
    pub(crate) scope: Option<Arc<str>>,
    /// Optional event-kind filter applied to matched events.
    pub(crate) fact: Option<KindFilter>,
    /// Optional inclusive per-entity clock range; does not apply to live filtering.
    pub(crate) clock_range: Option<ClockRange>, // per-entity clock, not global_sequence
    /// Optional exact DAG lane filter.
    pub(crate) lane: Option<u32>,
}

/// Filter on [`EventKind`] used within a [`Region`] query.
#[derive(Clone, Debug)]
#[non_exhaustive]
pub enum KindFilter {
    /// Matches only events with this exact kind.
    Exact(EventKind),
    /// Matches any event whose kind falls within this 4-bit category.
    Category(u8), // matches any EventKind in this 4-bit category
    /// Matches events of any kind.
    Any,
}

impl Coordinate {
    /// Creates a new `Coordinate` from an entity and scope string.
    ///
    /// Coordinate components are logical stream identifiers, not path or
    /// checkpoint-identity fragments. They must be non-empty, bounded, free of
    /// control bytes, free of path traversal shapes, and free of the `|` / `=`
    /// separators reserved by `Region::checkpoint_identity`.
    ///
    /// # Errors
    /// Returns any `CoordinateError` produced by the validation rules described
    /// above.
    pub fn new(entity: impl AsRef<str>, scope: impl AsRef<str>) -> Result<Self, CoordinateError> {
        let entity = entity.as_ref();
        let scope = scope.as_ref();
        Self::validate_parts(entity, scope)?;
        Ok(Self {
            entity: Arc::from(entity),
            scope: Arc::from(scope),
        })
    }

    /// Returns the entity string.
    pub fn entity(&self) -> &str {
        &self.entity
    }
    /// Returns the scope string.
    pub fn scope(&self) -> &str {
        &self.scope
    }
    pub(crate) fn entity_arc(&self) -> Arc<str> {
        Arc::clone(&self.entity)
    }
    pub(crate) fn scope_arc(&self) -> Arc<str> {
        Arc::clone(&self.scope)
    }

    pub(crate) fn from_shared_parts(
        entity: Arc<str>,
        scope: Arc<str>,
    ) -> Result<Self, CoordinateError> {
        Self::validate_parts(entity.as_ref(), scope.as_ref())?;
        Ok(Self { entity, scope })
    }

    /// Revalidate an existing coordinate against the current validation rules.
    ///
    /// Used at API boundaries (e.g. `submit_batch`) to defend against
    /// coordinates constructed through internal routes that bypass `new`,
    /// or produced by older on-disk data under tightened rules.
    ///
    /// # Errors
    /// Returns any [`CoordinateError`] that [`Coordinate::new`] would produce
    /// if called with the same entity/scope strings.
    pub fn validate(&self) -> Result<(), CoordinateError> {
        Self::validate_parts(self.entity.as_ref(), self.scope.as_ref())
    }

    fn validate_parts(entity: &str, scope: &str) -> Result<(), CoordinateError> {
        if entity.is_empty() {
            return Err(CoordinateError::EmptyEntity);
        }
        if scope.is_empty() {
            return Err(CoordinateError::EmptyScope);
        }
        if entity.len() > MAX_COORDINATE_COMPONENT_LEN {
            return Err(CoordinateError::EntityTooLong {
                len: entity.len(),
                max: MAX_COORDINATE_COMPONENT_LEN,
            });
        }
        if scope.len() > MAX_COORDINATE_COMPONENT_LEN {
            return Err(CoordinateError::ScopeTooLong {
                len: scope.len(),
                max: MAX_COORDINATE_COMPONENT_LEN,
            });
        }
        Self::validate_component_bytes(entity)?;
        Self::validate_component_bytes(scope)?;
        Ok(())
    }

    fn validate_component_bytes(value: &str) -> Result<(), CoordinateError> {
        for byte in value.bytes() {
            if byte == 0 {
                return Err(CoordinateError::NulByte);
            }
            // ASCII control range 0x00..=0x1F and DEL 0x7F. NUL is handled
            // above for a more specific error; the rest fall through here.
            if byte < 0x20 || byte == 0x7F {
                return Err(CoordinateError::ControlChar);
            }
        }
        if value.contains('/') || value.contains("..") {
            return Err(CoordinateError::PathTraversal);
        }
        if value.contains('|') || value.contains('=') {
            return Err(CoordinateError::ForbiddenSeparator);
        }
        Ok(())
    }
}

impl fmt::Display for Coordinate {
    /// "entity@scope"
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}@{}", self.entity, self.scope)
    }
}

/// Region builder with method chaining.
impl Region {
    /// Returns a region that matches all events.
    pub fn all() -> Self {
        Self::default()
    }

    /// Returns a region scoped to entities whose names start with `prefix`.
    pub fn entity(prefix: impl AsRef<str>) -> Self {
        Self {
            entity_prefix: Some(Arc::from(prefix.as_ref())),
            ..Self::default()
        }
    }

    /// Returns a region scoped to a specific scope string.
    pub fn scope(scope: impl AsRef<str>) -> Self {
        Self {
            scope: Some(Arc::from(scope.as_ref())),
            ..Self::default()
        }
    }

    /// Chainable setters
    pub fn with_scope(mut self, scope: impl AsRef<str>) -> Self {
        self.scope = Some(Arc::from(scope.as_ref()));
        self
    }

    /// Filters events by the given kind filter.
    pub fn with_fact(mut self, filter: KindFilter) -> Self {
        self.fact = Some(filter);
        self
    }

    /// Filters events to those whose kind matches the given category.
    pub fn with_fact_category(mut self, category: EventCategory) -> Self {
        self.fact = Some(KindFilter::Category(category.get()));
        self
    }

    /// Filters events to those within the given per-entity clock range.
    pub fn with_clock_range(mut self, range: ClockRange) -> Self {
        self.clock_range = Some(range);
        self
    }

    /// Filters events to an exact DAG lane.
    pub fn with_lane(mut self, lane: u32) -> Self {
        self.lane = Some(lane);
        self
    }

    /// Returns the configured entity prefix, if any.
    pub fn entity_prefix(&self) -> Option<&str> {
        self.entity_prefix.as_deref()
    }

    /// Returns the configured exact scope, if any.
    pub fn scope_value(&self) -> Option<&str> {
        self.scope.as_deref()
    }

    /// Returns the configured kind filter, if any.
    pub fn fact(&self) -> Option<&KindFilter> {
        self.fact.as_ref()
    }

    /// Returns the configured inclusive per-entity clock range, if any.
    pub fn clock_range(&self) -> Option<ClockRange> {
        self.clock_range
    }

    /// Returns the configured exact DAG lane, if any.
    pub fn lane(&self) -> Option<u32> {
        self.lane
    }

    /// Returns `true` when `entity` falls within this region's configured
    /// namespace prefix.
    #[must_use]
    pub(crate) fn matches_entity(&self, entity: &str) -> bool {
        match self.entity_prefix.as_deref() {
            Some(prefix) => namespace_prefix_matches(prefix, entity),
            None => true,
        }
    }

    /// Match against individual fields — avoids circular dep on store::Notification.
    /// Called by Subscription::recv() to filter events. [FILE:src/store/delivery/subscription.rs]
    pub fn matches_event(&self, entity: &str, scope: &str, kind: EventKind) -> bool {
        self.matches_event_on_lane(entity, scope, kind, None)
    }

    /// Match against individual fields with an optional DAG lane.
    pub(crate) fn matches_event_on_lane(
        &self,
        entity: &str,
        scope: &str,
        kind: EventKind,
        lane: Option<u32>,
    ) -> bool {
        if !self.matches_entity(entity) {
            return false;
        }
        if let Some(expected) = self.lane {
            if lane != Some(expected) {
                return false;
            }
        }
        if let Some(ref s) = self.scope {
            if scope != s.as_ref() {
                return false;
            }
        }
        if let Some(ref fact) = self.fact {
            match fact {
                KindFilter::Exact(k) => {
                    if kind != *k {
                        return false;
                    }
                }
                KindFilter::Category(c) => {
                    if kind.category() != *c {
                        return false;
                    }
                }
                KindFilter::Any => {}
            }
        }
        // clock_range is not checked here — it's for index queries, not live filtering.
        true
    }

    /// Stable identity string for persisted cursor checkpoints.
    pub(crate) fn checkpoint_identity(&self) -> String {
        // `Coordinate::validate_component_bytes` rejects `|` and `=`, so the
        // separator grammar below is injective for entity/scope components.
        let entity = self.entity_prefix.as_deref().unwrap_or("*");
        let scope = self.scope.as_deref().unwrap_or("*");
        let fact = match self.fact.as_ref() {
            Some(KindFilter::Exact(kind)) => {
                format!("exact:{:x}:{:x}", kind.category(), kind.type_id())
            }
            Some(KindFilter::Category(cat)) => format!("category:{cat:x}"),
            Some(KindFilter::Any) => "any".to_owned(),
            None => "none".to_owned(),
        };
        let clock = match self.clock_range {
            Some(range) => {
                let (start, end) = range.as_tuple();
                format!("{start}-{end}")
            }
            None => "*".to_owned(),
        };
        let base = format!("entity={entity}|scope={scope}|fact={fact}|clock={clock}");
        match self.lane {
            Some(lane) => format!("{base}|lane={lane}"),
            None => base,
        }
    }
}

/// Returns `true` when `candidate` is exactly `prefix` or is nested beneath it
/// at a `:` namespace boundary.
#[must_use]
pub(crate) fn namespace_prefix_matches(prefix: &str, candidate: &str) -> bool {
    candidate == prefix
        || candidate
            .strip_prefix(prefix)
            .is_some_and(|suffix| suffix.starts_with(':'))
}

#[cfg(test)]
mod tests {
    use super::{namespace_prefix_matches, Coordinate, CoordinateError, Region};
    use crate::event::EventKind;
    use std::sync::Arc;

    #[test]
    fn region_filter_error_display_renders_each_variant_detail() {
        use super::RegionFilterError;
        // `<RegionFilterError as Display>::fmt -> Ok(Default::default())` would
        // write NOTHING, producing an empty string for every variant. Each
        // variant must render its operative numbers and intent.
        let inverted = RegionFilterError::InvertedClockRange { start: 9, end: 3 }.to_string();
        assert!(
            inverted.contains("clock range start 9") && inverted.contains("end 3"),
            "InvertedClockRange Display must name its start/end; got {inverted:?}"
        );
        let out_of_range = RegionFilterError::CategoryOutOfRange { category: 99 }.to_string();
        assert!(
            out_of_range.contains("99") && out_of_range.contains("out of range"),
            "CategoryOutOfRange Display must name the bad category; got {out_of_range:?}"
        );
    }

    #[test]
    fn namespace_prefix_matches_exact_and_descendants() {
        assert!(namespace_prefix_matches("alice", "alice"));
        assert!(namespace_prefix_matches("alice", "alice:child"));
        assert!(namespace_prefix_matches("alice", "alice:child:grandchild"));
    }

    #[test]
    fn namespace_prefix_rejects_adjacent_namespaces() {
        assert!(!namespace_prefix_matches("alice", "alice2"));
        assert!(!namespace_prefix_matches("alpha-a", "alpha-aa"));
        assert!(!namespace_prefix_matches("alice", "alice-prod"));
        assert!(!namespace_prefix_matches("alice", "alіce"));
    }

    #[test]
    fn region_entity_uses_namespace_matcher() {
        let region = Region::entity("alpha:a");
        assert!(region.matches_entity("alpha:a"));
        assert!(region.matches_entity("alpha:a:child"));
        assert!(!region.matches_entity("alpha:aa"));
    }

    #[test]
    fn matches_event_rejects_non_matching_entity_and_scope() {
        // A region scoped to entity `alpha:a` in scope `room` must filter out
        // events that fall outside either dimension. The negative assertions
        // pin `matches_event`'s predicate: a body that always returned `true`
        // would let these foreign events through.
        let region = Region::entity("alpha:a").with_scope("room");
        let kind = EventKind::custom(0xF, 1);

        // Positive: same entity prefix + exact scope must match.
        assert!(
            region.matches_event("alpha:a", "room", kind),
            "region must accept events on its own entity prefix and scope"
        );
        assert!(
            region.matches_event("alpha:a:child", "room", kind),
            "region must accept descendants of its entity prefix"
        );

        // Negative: a different entity must NOT match.
        assert!(
            !region.matches_event("beta", "room", kind),
            "region must reject events on a foreign entity"
        );
        // Negative: an adjacent (non-namespace-boundary) entity must NOT match.
        assert!(
            !region.matches_event("alpha:aa", "room", kind),
            "region must reject adjacent entity namespaces"
        );
        // Negative: a different scope must NOT match.
        assert!(
            !region.matches_event("alpha:a", "lobby", kind),
            "region must reject events outside its exact scope"
        );
    }

    #[test]
    fn coordinate_rejects_checkpoint_identity_separators() {
        assert_eq!(
            Coordinate::new("entity|injection", "scope"),
            Err(CoordinateError::ForbiddenSeparator)
        );
        assert_eq!(
            Coordinate::new("entity", "scope=injection"),
            Err(CoordinateError::ForbiddenSeparator)
        );
        assert_eq!(
            Coordinate::new("entity", "*|fact=any|clock=*"),
            Err(CoordinateError::ForbiddenSeparator)
        );
    }

    #[test]
    fn coordinate_validate_rejects_internally_forged_separator_values() {
        let coord = Coordinate {
            entity: Arc::from("entity"),
            scope: Arc::from("*|fact=any|clock=*"),
        };

        assert_eq!(coord.validate(), Err(CoordinateError::ForbiddenSeparator));
    }

    #[test]
    fn coordinate_separator_error_is_displayable_std_error() {
        fn assert_error_trait(_: &dyn std::error::Error) {}

        let error = CoordinateError::ForbiddenSeparator;
        assert_error_trait(&error);
        assert!(error.to_string().contains("`|` or `=`"));
    }
}