road-runner-common 0.8.0

//! Async decision-audit sink (feature `authz-kafka`).
//!
//! Emits every allow/deny decision to the `identity_gateway.admin-audit-events`
//! topic as a `DECISION` event, consumed by cex-audit into its hash chain. The
//! request path must never block on this: a bounded channel with drop-on-full (plus
//! a dropped counter) decouples enforcement from audit delivery.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

use rdkafka::config::ClientConfig;
use rdkafka::producer::{FutureProducer, FutureRecord};
use serde::Serialize;
use tokio::sync::mpsc;

/// One authorization decision, shaped for cex-audit's `DECISION` decode branch.
#[derive(Debug, Clone, Serialize)]
pub struct DecisionEvent {
    #[serde(rename = "eventType")]
    pub event_type: &'static str,
    pub sub: String,
    pub permission: String,
    /// `allow` | `deny` | `step_up_required` | `quorum_required`.
    pub decision: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    #[serde(rename = "policyVersion", skip_serializing_if = "Option::is_none")]
    pub policy_version: Option<u64>,
    pub service: String,
    #[serde(rename = "occurredAt")]
    pub occurred_at_ms: i64,
}

impl DecisionEvent {
    pub fn new(
        service: impl Into<String>,
        sub: impl Into<String>,
        permission: impl Into<String>,
        decision: impl Into<String>,
        reason: Option<String>,
        policy_version: Option<u64>,
    ) -> Self {
        Self {
            event_type: "DECISION",
            sub: sub.into(),
            permission: permission.into(),
            decision: decision.into(),
            reason,
            policy_version,
            service: service.into(),
            occurred_at_ms: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_millis() as i64)
                .unwrap_or(0),
        }
    }
}

/// Bounded, non-blocking sink. Clone-cheap; share across handlers.
#[derive(Clone)]
pub struct DecisionAuditSink {
    tx: mpsc::Sender<DecisionEvent>,
    dropped: Arc<AtomicU64>,
}

impl DecisionAuditSink {
    /// Spawn the background producer. `capacity` bounds in-flight events; when full,
    /// [`DecisionAuditSink::record`] drops (never blocks the request).
    pub fn spawn(
        brokers: &str,
        topic: impl Into<String>,
        capacity: usize,
    ) -> Result<Self, anyhow::Error> {
        // Platform connection+security; caller-supplied brokers win over the env.
        let settings = {
            let mut s = crate::kafka::KafkaSettings::from_env();
            s.brokers = brokers.to_string();
            s
        };
        let mut client = ClientConfig::new();
        for (k, v) in settings.security_settings() {
            client.set(k, v);
        }
        let producer: FutureProducer = client
            .set("message.timeout.ms", "5000")
            .set("acks", "1")
            .create()?;
        let topic = topic.into();
        let (tx, mut rx) = mpsc::channel::<DecisionEvent>(capacity);

        tokio::spawn(async move {
            while let Some(event) = rx.recv().await {
                let Ok(payload) = serde_json::to_vec(&event) else {
                    continue;
                };
                let key = event.sub.clone();
                let record = FutureRecord::to(&topic).key(&key).payload(&payload);
                if let Err((e, _)) = producer.send(record, Duration::from_secs(5)).await {
                    #[cfg(feature = "observability")]
                    tracing::warn!(error = %e, "decision audit publish failed");
                    let _ = e;
                }
            }
        });

        Ok(Self { tx, dropped: Arc::new(AtomicU64::new(0)) })
    }

    /// Enqueue a decision. Non-blocking; increments the dropped counter if the
    /// channel is full or closed.
    pub fn record(&self, event: DecisionEvent) {
        if self.tx.try_send(event).is_err() {
            self.dropped.fetch_add(1, Ordering::Relaxed);
        }
    }

    /// Number of decisions dropped because the channel was full (for a metric).
    pub fn dropped(&self) -> u64 {
        self.dropped.load(Ordering::Relaxed)
    }
}