axess-core 0.11.0

//! Denormalised authentication telemetry for operational analytics.
//!
//! axess's regulatory audit path is [`AuthEvent`] →
//! [`IdentityAuthnLog::record_event`](crate::authn::store::IdentityAuthnLog::record_event):
//! every login attempt, factor verification, session creation, and
//! revocation event lands in the adopter's `authn_hist` table.
//! That table answers regulator questions ("did this user authenticate
//! on date X?") but is wrong-shaped for SOC dashboards, fraud
//! investigation, or product analytics: those want denormalised,
//! join-free records they can stream into a columnar store (DuckDB,
//! ClickHouse, Snowflake) and aggregate cheaply.
//!
//! This module defines the **shape** of that analytics stream so
//! multiple adopters can share downstream consumers. axess does not
//! ship a concrete sink; adopters wire their own pipeline (typical:
//! rkyv-serialise → Iggy / Kafka → columnar store).
//!
//! # Why not extend `AuthEvent`?
//!
//! `AuthEvent` is the regulatory record. Adding analytics fields to it
//! would either bloat every audit row (extra storage) or make some
//! fields silently optional (regulator can't tell whether the field
//! was missing or unrecorded: a finding waiting to happen).
//!
//! `RichAuthnEvent` carries `event: AuthEvent` as its core plus
//! adopter-supplied enrichment that doesn't survive in the audit
//! table. Adopters who enable the analytics path get the rich record;
//! adopters who don't pay nothing.
//!
//! # Hot path
//!
//! The auth hot path MUST NOT block on analytics writes.
//! [`AuditLogWithAnalytics`] enqueues each event on a bounded channel
//! with a non-blocking `try_send` and a worker the caller owns
//! ([`AnalyticsWorker`]) drains it into the sink. When the queue is
//! full the analytics copy is dropped and counted
//! ([`dropped_events`](AuditLogWithAnalytics::dropped_events)); the
//! regulatory `AuthEvent` is on a separate synchronous path and is never
//! dropped. A crash between emission and the sink's flush loses analytics
//! data, which is the trade the stream makes.

use crate::authn::event::AuthEvent;
#[cfg(feature = "device")]
use crate::device::types::DeviceTrustLevel;

#[cfg(feature = "serde")]
use serde::{Deserialize, Serialize};

// ── UserAgentSummary ─────────────────────────────────────────────────────────

/// Parsed user-agent fields useful for analytics. axess does not run a
/// UA parser; adopters populate this from their own parsing path
/// (e.g. the `uaparser` crate fed from a `User-Agent` header captured
/// upstream of the auth flow).
///
/// All fields are optional because UA strings vary in completeness and
/// adopters may choose to parse only the subset they need (bot
/// detection alone is a common case).
#[derive(Debug, Clone, Default)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[cfg_attr(
    feature = "rkyv",
    derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)
)]
pub struct UserAgentSummary {
    /// Browser family: `"Chrome"`, `"Safari"`, `"Firefox"`, …
    pub browser_family: Option<String>,
    /// Browser version, free-form: `"139.0.7233"`, `"17.6"`, …
    pub browser_version: Option<String>,
    /// Operating-system family: `"macOS"`, `"Windows"`, `"iOS"`, …
    pub os_family: Option<String>,
    /// Operating-system version, free-form.
    pub os_version: Option<String>,
    /// Adopter's bot detection result. axess has no opinion on what
    /// counts as a bot; leave `false` if you don't run a detector.
    pub is_bot: bool,
}

// ── RichAuthnEvent ───────────────────────────────────────────────────────────

/// Denormalised authentication event for the analytics stream.
///
/// Combines the regulatory [`AuthEvent`] (always present) with optional
/// enrichment fields. Adopters construct this from the `AuthEvent`
/// their `IdentityAuthnLog::record_event` impl receives plus whatever
/// context they have available (GeoIP lookup, UA parser, device
/// trust resolution, …) and hand it to the configured
/// [`AuthnAnalyticsSink`].
///
/// # rkyv compatibility
///
/// All fields are rkyv-derivable under the `rkyv` feature. Adopters
/// streaming to Apache Iggy / Kafka / similar can serialise with
/// `rkyv::to_bytes::<_, 1024>(&event)` and consume zero-copy on the
/// downstream side. The schema is intentionally stable: adding new
/// fields is a non-breaking change as long as they are `Option<T>` or
/// have a `Default`.
#[derive(Debug, Clone)]
#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
#[cfg_attr(
    feature = "rkyv",
    derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)
)]
pub struct RichAuthnEvent {
    /// The regulatory core: the same `AuthEvent` that gets persisted
    /// to the adopter's `authn_hist` table via
    /// [`IdentityAuthnLog::record_event`](crate::authn::store::IdentityAuthnLog::record_event).
    pub event: AuthEvent,

    /// Device trust level resolved at event time. `None` when the
    /// `device` feature is off, the event fires pre-device-resolution
    /// (login attempt), or the adopter doesn't run the device
    /// subsystem.
    #[cfg(feature = "device")]
    pub device_trust_level: Option<DeviceTrustLevel>,

    /// ISO 3166-1 alpha-2 country code inferred from the client IP.
    /// Adopter resolves this via their own GeoIP path (MaxMind,
    /// IPinfo, …) before constructing the rich event. axess does not
    /// pull a GeoIP database into the library.
    pub geo_country: Option<String>,

    /// Autonomous system number of the client's network. Same
    /// adopter-supplied path as `geo_country`.
    pub geo_asn: Option<u32>,

    /// Parsed user-agent summary. See [`UserAgentSummary`].
    pub user_agent_summary: Option<UserAgentSummary>,

    /// Free-form adopter-controlled tags. Keep keys short and the
    /// vocabulary stable: these often end up as column headers in
    /// the analytics store. Typical: `"channel" => "mobile"`,
    /// `"experiment" => "step-up-v2"`.
    pub tags: Vec<(String, String)>,
}

impl RichAuthnEvent {
    /// Construct an enrichment-free rich event from a regulatory
    /// `AuthEvent`. Use the `.with_*` fluent helpers to populate
    /// enrichment fields before handing to the sink.
    pub fn from_event(event: AuthEvent) -> Self {
        Self {
            event,
            #[cfg(feature = "device")]
            device_trust_level: None,
            geo_country: None,
            geo_asn: None,
            user_agent_summary: None,
            tags: Vec::new(),
        }
    }

    /// Set the resolved device trust level. Requires the `device` feature.
    #[cfg(feature = "device")]
    pub fn with_device_trust_level(mut self, level: DeviceTrustLevel) -> Self {
        self.device_trust_level = Some(level);
        self
    }

    /// Set the ISO-3166 alpha-2 country code.
    pub fn with_geo_country(mut self, country: impl Into<String>) -> Self {
        self.geo_country = Some(country.into());
        self
    }

    /// Set the network ASN.
    pub fn with_geo_asn(mut self, asn: u32) -> Self {
        self.geo_asn = Some(asn);
        self
    }

    /// Set the parsed UA summary.
    pub fn with_user_agent_summary(mut self, ua: UserAgentSummary) -> Self {
        self.user_agent_summary = Some(ua);
        self
    }

    /// Append a free-form tag.
    pub fn with_tag(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.tags.push((key.into(), value.into()));
        self
    }
}

// ── AuthnAnalyticsSink ───────────────────────────────────────────────────────

/// Sink for [`RichAuthnEvent`] records. Adopters implement this against
/// their analytics infrastructure; typical patterns include
/// rkyv-serialising to Apache Iggy or Kafka, JSONL-appending to a
/// rotating filesystem path, or HTTP-posting to a managed analytics
/// service.
///
/// # Hot path
///
/// `record_rich` is not called from the auth path: [`AnalyticsWorker`]
/// calls it, one event at a time in the order the events were recorded,
/// from a queue that is bounded, so a slow sink costs queue space and then
/// dropped analytics events, never authentication latency. Because calls
/// are sequential, a sink that needs throughput should buffer internally
/// and flush in batches; one that blocks for a long time per event will fill
/// the queue. Do not put an unbounded channel in front of it: the bounded
/// queue is what keeps a stalled sink from growing memory.
///
/// # Error handling
///
/// The returned `Result` is logged inside axess but does not propagate
/// to the caller. A failing sink does not fail authentication; the
/// regulatory record on [`IdentityAuthnLog`](crate::authn::store::IdentityAuthnLog)
/// is the source of truth, and the analytics path is best-effort by
/// design.
pub trait AuthnAnalyticsSink: Send + Sync + 'static {
    /// Sink-specific error type. Most implementations will wrap an I/O
    /// error, a channel-send error, or a serialisation error.
    type Error: std::error::Error + Send + Sync + 'static;

    /// Sink a denormalised analytics event.
    ///
    /// Called from the [`AnalyticsWorker`], never from the auth path, so a
    /// sink that takes seconds to flush does not slow login. The event is at
    /// risk of being lost on process shutdown if the sink has not flushed its
    /// own buffer; see [`AuditLogWithAnalytics::shutdown`].
    fn record_rich(
        &self,
        event: RichAuthnEvent,
    ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;

    /// Human-readable name for log lines + dashboards.
    fn name(&self) -> &'static str;
}

/// No-op analytics sink. Returns `Ok(())` without persisting anything.
///
/// Used as the default when no analytics infrastructure is wired:
/// makes [`AuthnService`](crate::authn::service::AuthnService) type
/// signatures stable across "no analytics" and "with analytics"
/// deployments without requiring a generic bound every adopter must
/// satisfy.
#[derive(Debug, Default, Clone, Copy)]
pub struct NoopAuthnAnalyticsSink;

impl AuthnAnalyticsSink for NoopAuthnAnalyticsSink {
    type Error = std::convert::Infallible;

    async fn record_rich(&self, event: RichAuthnEvent) -> Result<(), Self::Error> {
        tracing::trace!(
            target: "axess::audit::analytics",
            event_type = ?event.event.event_type,
            event_time = event.event.event_time,
            "NoopAuthnAnalyticsSink: rich event discarded",
        );
        Ok(())
    }

    fn name(&self) -> &'static str {
        "noop"
    }
}

// ── AuditLogWithAnalytics ──────────────────────────────────────────────────────────────

/// Queue capacity used by [`AuditLogWithAnalytics::new`].
pub const DEFAULT_ANALYTICS_QUEUE: usize = 10_000;

enum Message {
    Event(Box<AuthEvent>),
    /// Acknowledged once every event queued before it has been handled.
    Flush(tokio::sync::oneshot::Sender<()>),
}

/// The analytics queue or its worker is gone.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("the analytics worker is not running")]
pub struct AnalyticsClosed;

/// Decorator that splits the regulatory audit path so every
/// [`record_event`](crate::authn::store::IdentityAuthnLog::record_event) call
/// also reaches an [`AuthnAnalyticsSink`].
///
/// Wraps a hot [`IdentityAuthnLog`](crate::authn::store::IdentityAuthnLog) impl
/// plus an analytics sink plus an enricher closure that turns a bare
/// [`AuthEvent`] into a [`RichAuthnEvent`] using whatever adopter
/// context is available (GeoIP, UA parser, device resolver, …).
///
/// The hot path stays synchronous: the inner log's `record_event` is
/// awaited so a regulatory write failure surfaces to the caller. Only after
/// that write returns `Recorded` is the event placed on a bounded queue, with
/// `try_send`, inside the same returned future. Nothing is spawned per event
/// and construction needs no runtime. A full queue drops the analytics copy
/// and counts it ([`dropped_events`](Self::dropped_events)); a warning is
/// logged at the first drop and each time the total reaches a power of two.
///
/// [`new`](Self::new) returns the decorator and an [`AnalyticsWorker`] that
/// the caller runs, so its lifetime is the caller's: spawn it with
/// [`AnalyticsWorker::spawn`] and keep the handle. The worker ends, after
/// handling everything still queued, when every decorator handle is dropped;
/// [`flush`](Self::flush) waits for the queue to empty and
/// [`shutdown`](Self::shutdown) does both ends of a graceful stop.
///
/// ```rust,ignore
/// use axess_core::authn::audit::analytics::{AuditLogWithAnalytics, RichAuthnEvent};
///
/// let (log, worker) = AuditLogWithAnalytics::new(
///     postgres_backend,
///     iggy_sink,
///     |event| async move {
///         let ua = parse_ua(event.user_agent.as_deref()).await;
///         let geo = geoip_lookup(event.ip_address).await;
///         RichAuthnEvent::from_event(event)
///             .with_user_agent_summary(ua)
///             .with_geo_country(geo.country)
///             .with_geo_asn(geo.asn)
///     },
/// );
/// let drain = worker.spawn();
/// // ... serve ...
/// log.shutdown().await;
/// drain.await.ok();
/// ```
pub struct AuditLogWithAnalytics<L, S, E> {
    inner: L,
    sink: std::sync::Arc<S>,
    queue: tokio::sync::mpsc::Sender<Message>,
    dropped: std::sync::atomic::AtomicU64,
    enricher: std::marker::PhantomData<fn() -> E>,
}

impl<L, S, E> AuditLogWithAnalytics<L, S, E>
where
    L: crate::authn::store::IdentityAuthnLog,
    S: AuthnAnalyticsSink,
    E: Send + Sync + 'static,
{
    /// Construct a tee decorator with a queue of
    /// [`DEFAULT_ANALYTICS_QUEUE`] events. The `enricher` closure receives the
    /// regulatory `AuthEvent` and returns a future yielding the
    /// denormalised `RichAuthnEvent`: adopters do GeoIP / UA parsing
    /// / device-trust resolution inside this closure.
    ///
    /// Run the returned worker, or no event ever reaches the sink.
    pub fn new(inner: L, sink: S, enricher: E) -> (Self, AnalyticsWorker<S, E>) {
        Self::with_capacity(inner, sink, enricher, DEFAULT_ANALYTICS_QUEUE)
    }

    /// As [`new`](Self::new) with a queue of `capacity` events (minimum 1).
    pub fn with_capacity(
        inner: L,
        sink: S,
        enricher: E,
        capacity: usize,
    ) -> (Self, AnalyticsWorker<S, E>) {
        let (queue, rx) = tokio::sync::mpsc::channel(capacity.max(1));
        let sink = std::sync::Arc::new(sink);
        let worker = AnalyticsWorker {
            rx,
            sink: std::sync::Arc::clone(&sink),
            enricher: std::sync::Arc::new(enricher),
        };
        let log = Self {
            inner,
            sink,
            queue,
            dropped: std::sync::atomic::AtomicU64::new(0),
            enricher: std::marker::PhantomData,
        };
        (log, worker)
    }

    /// Borrow the inner regulatory log.
    pub fn inner(&self) -> &L {
        &self.inner
    }

    /// Shared handle to the analytics sink: useful for tests or
    /// dashboards that want to read sink-side state directly.
    pub fn sink(&self) -> &std::sync::Arc<S> {
        &self.sink
    }

    /// How many analytics events were dropped because the queue was full or
    /// the worker had stopped. Regulatory records are never counted here.
    pub fn dropped_events(&self) -> u64 {
        self.dropped.load(std::sync::atomic::Ordering::Relaxed)
    }

    /// Wait until every event queued so far has been handled by the sink.
    ///
    /// Events dropped for want of queue space are not recoverable and are not
    /// waited for. Fails with [`AnalyticsClosed`] if the worker is not running.
    pub async fn flush(&self) -> Result<(), AnalyticsClosed> {
        let (ack, done) = tokio::sync::oneshot::channel();
        self.queue
            .send(Message::Flush(ack))
            .await
            .map_err(|_| AnalyticsClosed)?;
        done.await.map_err(|_| AnalyticsClosed)
    }

    /// Flush the queue and release this handle. Once the last handle is gone
    /// the worker finishes; await the handle [`AnalyticsWorker::spawn`]
    /// returned to know the sink has seen everything.
    pub async fn shutdown(self) {
        let _ = self.flush().await;
    }

    /// Place `event` on the queue without waiting.
    fn enqueue(&self, event: AuthEvent) {
        use std::sync::atomic::Ordering;
        if self
            .queue
            .try_send(Message::Event(Box::new(event)))
            .is_err()
        {
            let total = self.dropped.fetch_add(1, Ordering::Relaxed) + 1;
            if total.is_power_of_two() {
                tracing::warn!(
                    target: "axess::audit::analytics",
                    sink = %self.sink.name(),
                    dropped_total = total,
                    "analytics queue full or worker stopped; dropping analytics event \
                     (the regulatory record is unaffected)",
                );
            }
        }
    }
}

/// Drains the analytics queue into the sink. Created by
/// [`AuditLogWithAnalytics::new`] and run by the caller.
///
/// Events are handled one at a time in the order they were recorded. The
/// enricher and the sink run in a task of their own for each event, so a panic
/// in either is logged and costs that event only.
pub struct AnalyticsWorker<S, E> {
    rx: tokio::sync::mpsc::Receiver<Message>,
    sink: std::sync::Arc<S>,
    enricher: std::sync::Arc<E>,
}

impl<S, E, EFut> AnalyticsWorker<S, E>
where
    S: AuthnAnalyticsSink,
    E: Fn(AuthEvent) -> EFut + Send + Sync + 'static,
    EFut: std::future::Future<Output = RichAuthnEvent> + Send + 'static,
{
    /// Handle queued events until every [`AuditLogWithAnalytics`] handle has
    /// been dropped and the queue is empty.
    pub async fn run(mut self) {
        while let Some(message) = self.rx.recv().await {
            match message {
                Message::Flush(ack) => {
                    let _ = ack.send(());
                }
                Message::Event(event) => {
                    let sink = std::sync::Arc::clone(&self.sink);
                    let enricher = std::sync::Arc::clone(&self.enricher);
                    let one = tokio::spawn(async move {
                        let rich = enricher(*event).await;
                        let sink_name = sink.name();
                        if let Err(e) = sink.record_rich(rich).await {
                            tracing::warn!(
                                sink = %sink_name,
                                error = %e,
                                "authn analytics sink rejected event; regulatory record is unaffected",
                            );
                        }
                    });
                    if let Err(e) = one.await {
                        tracing::error!(
                            target: "axess::audit::analytics",
                            error = %e,
                            "analytics enricher or sink panicked; event lost, worker continues",
                        );
                    }
                }
            }
        }
    }

    /// Run the worker on the current tokio runtime and return its handle.
    pub fn spawn(self) -> tokio::task::JoinHandle<()> {
        tokio::spawn(self.run())
    }
}

// `IdentityAuthnLog` and `IdentityLookup` are blanket-forwarded to the
// inner log. The analytics-side spawn lives in `record_event` only;
// lookups and admin-side calls are unaffected.
impl<L, S, E, EFut> crate::authn::store::IdentityLookup for AuditLogWithAnalytics<L, S, E>
where
    L: crate::authn::store::IdentityAuthnLog,
    S: AuthnAnalyticsSink,
    E: Fn(AuthEvent) -> EFut + Send + Sync + 'static,
    EFut: std::future::Future<Output = RichAuthnEvent> + Send + 'static,
{
    type Error = <L as crate::authn::store::IdentityLookup>::Error;

    fn find_user(
        &self,
        identifier: &str,
        tenant_id: &crate::authn::ids::TenantId,
    ) -> impl std::future::Future<Output = Result<Option<crate::authn::types::User>, Self::Error>> + Send
    {
        self.inner.find_user(identifier, tenant_id)
    }

    fn get_user(
        &self,
        user_id: &crate::authn::ids::UserId,
    ) -> impl std::future::Future<Output = Result<Option<crate::authn::types::User>, Self::Error>> + Send
    {
        self.inner.get_user(user_id)
    }

    fn get_user_tenant(
        &self,
        user_id: &crate::authn::ids::UserId,
    ) -> impl std::future::Future<Output = Result<Option<crate::authn::ids::TenantId>, Self::Error>> + Send
    {
        self.inner.get_user_tenant(user_id)
    }

    fn find_tenant(
        &self,
        identifier: &str,
    ) -> impl std::future::Future<Output = Result<Option<crate::authn::types::Tenant>, Self::Error>> + Send
    {
        self.inner.find_tenant(identifier)
    }

    fn default_tenant(
        &self,
    ) -> impl std::future::Future<Output = Result<crate::authn::types::Tenant, Self::Error>> + Send
    {
        self.inner.default_tenant()
    }

    fn account_status(
        &self,
        user_id: &crate::authn::ids::UserId,
    ) -> impl std::future::Future<Output = Result<crate::authn::types::EntityState, Self::Error>> + Send
    {
        self.inner.account_status(user_id)
    }

    fn lockout_policy_for_tenant(
        &self,
        tenant_id: &crate::authn::ids::TenantId,
    ) -> crate::authn::types::LockoutPolicy {
        self.inner.lockout_policy_for_tenant(tenant_id)
    }
}

impl<L, S, E, EFut> crate::authn::store::IdentityAuthnLog for AuditLogWithAnalytics<L, S, E>
where
    L: crate::authn::store::IdentityAuthnLog,
    S: AuthnAnalyticsSink,
    E: Fn(AuthEvent) -> EFut + Send + Sync + 'static,
    EFut: std::future::Future<Output = RichAuthnEvent> + Send + 'static,
{
    async fn record_event(
        &self,
        event: AuthEvent,
    ) -> Result<crate::authn::store::AuditOutcome, <L as crate::authn::store::IdentityLookup>::Error>
    {
        // The regulatory write comes first. Analytics sees only events
        // the regulatory log holds: a failed write fails the flow, and
        // a shed event is a deliberate gap, so neither may appear in
        // the analytics stream as if it had happened on the record.
        let for_analytics = event.clone();
        let outcome = self.inner.record_event(event).await?;
        if outcome == crate::authn::store::AuditOutcome::Recorded {
            self.enqueue(for_analytics);
        }
        Ok(outcome)
    }

    fn record_failed_attempt(
        &self,
        user_id: &crate::authn::ids::UserId,
    ) -> impl std::future::Future<
        Output = Result<u32, <L as crate::authn::store::IdentityLookup>::Error>,
    > + Send {
        self.inner.record_failed_attempt(user_id)
    }

    fn reset_failed_attempts(
        &self,
        user_id: &crate::authn::ids::UserId,
    ) -> impl std::future::Future<
        Output = Result<(), <L as crate::authn::store::IdentityLookup>::Error>,
    > + Send {
        self.inner.reset_failed_attempts(user_id)
    }
}

#[cfg(test)]
mod tests;