photoncloud-monorepo/iam/crates/iam-audit/src/sink.rs

//! Audit sinks
//!
//! Pluggable destinations for audit events.

use std::path::Path;
use std::sync::Arc;

use async_trait::async_trait;
use tokio::fs::{File, OpenOptions};
use tokio::io::AsyncWriteExt;
use tokio::sync::{Mutex, RwLock};

use crate::event::AuditEvent;

/// Result type for audit operations
pub type Result<T> = std::result::Result<T, AuditError>;

/// Audit error
#[derive(Debug, thiserror::Error)]
pub enum AuditError {
    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),

    #[error("Serialization error: {0}")]
    Serialization(#[from] serde_json::Error),

    #[error("Sink error: {0}")]
    Sink(String),
}

/// Trait for audit event sinks
#[async_trait]
pub trait AuditSink: Send + Sync {
    /// Write an audit event to the sink
    async fn write(&self, event: &AuditEvent) -> Result<()>;

    /// Flush any buffered events
    async fn flush(&self) -> Result<()>;

    /// Close the sink
    async fn close(&self) -> Result<()>;
}

/// File-based audit sink
///
/// Writes audit events as JSON lines to a file.
pub struct FileSink {
    file: Mutex<File>,
    path: String,
}

impl FileSink {
    /// Create a new file sink
    pub async fn new(path: impl AsRef<Path>) -> Result<Self> {
        let path_str = path.as_ref().to_string_lossy().to_string();
        let file = OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .await?;

        Ok(Self {
            file: Mutex::new(file),
            path: path_str,
        })
    }

    /// Get the file path
    pub fn path(&self) -> &str {
        &self.path
    }
}

#[async_trait]
impl AuditSink for FileSink {
    async fn write(&self, event: &AuditEvent) -> Result<()> {
        let json = event.to_json()?;
        let line = format!("{}\n", json);

        let mut file = self.file.lock().await;
        file.write_all(line.as_bytes()).await?;

        Ok(())
    }

    async fn flush(&self) -> Result<()> {
        let mut file = self.file.lock().await;
        file.flush().await?;
        Ok(())
    }

    async fn close(&self) -> Result<()> {
        self.flush().await
    }
}

/// In-memory audit sink (for testing)
pub struct MemorySink {
    events: RwLock<Vec<AuditEvent>>,
    max_events: usize,
}

impl MemorySink {
    /// Create a new memory sink
    pub fn new(max_events: usize) -> Self {
        Self {
            events: RwLock::new(Vec::new()),
            max_events,
        }
    }

    /// Create with default max events (10000)
    pub fn default_capacity() -> Self {
        Self::new(10000)
    }

    /// Get all events
    pub async fn events(&self) -> Vec<AuditEvent> {
        self.events.read().await.clone()
    }

    /// Get event count
    pub async fn count(&self) -> usize {
        self.events.read().await.len()
    }

    /// Clear all events
    pub async fn clear(&self) {
        self.events.write().await.clear();
    }

    /// Find events by principal ID
    pub async fn find_by_principal(&self, principal_id: &str) -> Vec<AuditEvent> {
        self.events
            .read()
            .await
            .iter()
            .filter(|e| e.principal_id.as_deref() == Some(principal_id))
            .cloned()
            .collect()
    }
}

#[async_trait]
impl AuditSink for MemorySink {
    async fn write(&self, event: &AuditEvent) -> Result<()> {
        let mut events = self.events.write().await;

        // If at capacity, remove oldest event
        if events.len() >= self.max_events {
            events.remove(0);
        }

        events.push(event.clone());
        Ok(())
    }

    async fn flush(&self) -> Result<()> {
        Ok(())
    }

    async fn close(&self) -> Result<()> {
        Ok(())
    }
}

/// Multi-sink that writes to multiple sinks
pub struct MultiSink {
    sinks: Vec<Arc<dyn AuditSink>>,
}

impl MultiSink {
    /// Create a new multi-sink
    pub fn new(sinks: Vec<Arc<dyn AuditSink>>) -> Self {
        Self { sinks }
    }

    /// Add a sink
    pub fn add(&mut self, sink: Arc<dyn AuditSink>) {
        self.sinks.push(sink);
    }
}

#[async_trait]
impl AuditSink for MultiSink {
    async fn write(&self, event: &AuditEvent) -> Result<()> {
        for sink in &self.sinks {
            sink.write(event).await?;
        }
        Ok(())
    }

    async fn flush(&self) -> Result<()> {
        for sink in &self.sinks {
            sink.flush().await?;
        }
        Ok(())
    }

    async fn close(&self) -> Result<()> {
        for sink in &self.sinks {
            sink.close().await?;
        }
        Ok(())
    }
}

/// Null sink that discards all events
pub struct NullSink;

#[async_trait]
impl AuditSink for NullSink {
    async fn write(&self, _event: &AuditEvent) -> Result<()> {
        Ok(())
    }

    async fn flush(&self) -> Result<()> {
        Ok(())
    }

    async fn close(&self) -> Result<()> {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[tokio::test]
    async fn test_file_sink() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("audit.jsonl");

        let sink = FileSink::new(&path).await.unwrap();

        let event = AuditEvent::authn_success("alice", "jwt");
        sink.write(&event).await.unwrap();
        sink.flush().await.unwrap();

        // Read back and verify
        let contents = tokio::fs::read_to_string(&path).await.unwrap();
        assert!(contents.contains("alice"));
        assert!(contents.contains("jwt"));
    }

    #[tokio::test]
    async fn test_memory_sink() {
        let sink = MemorySink::new(100);

        let event1 = AuditEvent::authn_success("alice", "jwt");
        let event2 = AuditEvent::authn_success("bob", "mtls");

        sink.write(&event1).await.unwrap();
        sink.write(&event2).await.unwrap();

        assert_eq!(sink.count().await, 2);

        let alice_events = sink.find_by_principal("alice").await;
        assert_eq!(alice_events.len(), 1);
    }

    #[tokio::test]
    async fn test_memory_sink_capacity() {
        let sink = MemorySink::new(2);

        for i in 0..5 {
            let event = AuditEvent::authn_success(&format!("user-{}", i), "jwt");
            sink.write(&event).await.unwrap();
        }

        // Should only have last 2 events
        assert_eq!(sink.count().await, 2);

        let events = sink.events().await;
        assert_eq!(events[0].principal_id, Some("user-3".to_string()));
        assert_eq!(events[1].principal_id, Some("user-4".to_string()));
    }

    #[tokio::test]
    async fn test_multi_sink() {
        let sink1 = Arc::new(MemorySink::new(100));
        let sink2 = Arc::new(MemorySink::new(100));

        let multi = MultiSink::new(vec![sink1.clone(), sink2.clone()]);

        let event = AuditEvent::authn_success("alice", "jwt");
        multi.write(&event).await.unwrap();

        assert_eq!(sink1.count().await, 1);
        assert_eq!(sink2.count().await, 1);
    }

    #[tokio::test]
    async fn test_null_sink() {
        let sink = NullSink;
        let event = AuditEvent::authn_success("alice", "jwt");

        // Should not error
        sink.write(&event).await.unwrap();
        sink.flush().await.unwrap();
        sink.close().await.unwrap();
    }
}