//! Test-only `SessionStream` implementation for integration tests.
//!
//! Records every received `BusEvent` into an in-memory buffer; can be
//! configured to fail the next N events to exercise health drift paths.

#![cfg(any(test, feature = "test-utils"))]

use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU32, Ordering};

use async_trait::async_trait;
use chrono::Utc;

use dirigent_protocol::streaming::{
    BusEvent, SessionStream, StreamError, StreamKind, StreamOutcome, StreamScope, StreamSummary,
};

/// In-memory `SessionStream` used in integration tests.
pub struct MockStream {
    scope: StreamScope,
    name: String,
    pub received: Arc<Mutex<Vec<BusEvent>>>,
    pub fail_remaining: Arc<AtomicU32>,
    pub shutdown_called: Arc<Mutex<bool>>,
}

impl MockStream {
    pub fn new(name: impl Into<String>, scope: StreamScope) -> Arc<Self> {
        Arc::new(Self {
            scope,
            name: name.into(),
            received: Arc::new(Mutex::new(Vec::new())),
            fail_remaining: Arc::new(AtomicU32::new(0)),
            shutdown_called: Arc::new(Mutex::new(false)),
        })
    }

    /// Configure the next N `on_event` calls to return `StreamOutcome::Failed`.
    pub fn fail_next(&self, n: u32) {
        self.fail_remaining.store(n, Ordering::Relaxed);
    }

    pub fn received_count(&self) -> usize {
        self.received.lock().unwrap().len()
    }

    pub fn was_shutdown(&self) -> bool {
        *self.shutdown_called.lock().unwrap()
    }
}

#[async_trait]
impl SessionStream for MockStream {
    fn summary(&self) -> StreamSummary {
        StreamSummary {
            name: self.name.clone(),
            kind: StreamKind::Custom,
            target: "mock".into(),
            active_since: Utc::now(),
        }
    }

    fn scope(&self) -> StreamScope {
        self.scope.clone()
    }

    async fn on_event(&self, event: &BusEvent) -> StreamOutcome {
        if self.fail_remaining.load(Ordering::Relaxed) > 0 {
            self.fail_remaining.fetch_sub(1, Ordering::Relaxed);
            return StreamOutcome::Failed(StreamError::Rejected("mock fail".into()));
        }
        self.received.lock().unwrap().push(event.clone());
        StreamOutcome::Ok
    }

    async fn shutdown(&self) {
        *self.shutdown_called.lock().unwrap() = true;
    }
}
