//! # valkey_sink — first Valkey consumer (deploy spec §5.6)
//!
//! Implements the existing [`EventSink`] trait: `XADD` of [`CallEvent`]s as
//! JSON to a capped stream (`XADD MAXLEN ~ 10000`). STRICTLY
//! fire-and-forget: a bounded `mpsc` channel + a single background tokio
//! task does the I/O. Valkey down ⇒ count-and-drop via [`Arc`]`<`
//! [`AtomicU64`]`>` (the same counter `/metrics` reads in Task 6 as
//! `rutster_event_sink_drops_total`). When `RUTSTER_VALKEY_URL` is unset,
//! [`TracingEventSink`] stays in place; `main.rs` makes the swap.
//!
//! ## Why `XADD MAXLEN ~ 10000` (the adopted default)?
//!
//! ~10k events ≈ a few MB of memory under normal op (a [`CallEvent`]
//! serializes to ~200 bytes of JSON; 10k × 200B ≈ 2MB). The forensic
//! window covers days of calls for a small fleet. The cap is `~`
//! (approximate) which lets Valkey trim lazily — lower write-path latency
//! than exact `MAXLEN 10000`. Bumping or dropping the cap is a one-line
//! code change + redeploy; no operator action on Valkey.
//!
//! ## Hot-path discipline (the whole point of this module)
//!
//! `emit()` is called from the media `std::thread` every time a session
//! lifecycle changes. That thread must NEVER await I/O; a single slow
//! Valkey round-trip would push every session past its 20ms outbound
//! deadline simultaneously. So `emit()` does exactly one
//! [`mpsc::Sender::try_send`]. If the bounded queue is full, or the
//! background task has died, the event is counted and dropped.
//!
//! This is the canonical Rust pattern for “observer off the hot path”:
//! [`Arc`] makes the counter shareable across threads, [`AtomicU64`]
//! makes incrementing lock-free, and [`mpsc`] decouples the producer
//! from the consumer.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};

use tokio::sync::mpsc;

use crate::event_sink::{CallEvent, EndReason, EventSink};

/// Stream key on Valkey. Constant for the slice; a future slice makes it
/// configurable when the operator wants a separate stream per tenant
/// (multi-tenancy ADR defers the schema).
pub const STREAM_KEY: &str = "rutster:events";

/// XADD MAXLEN cap. ~10k events ≈ a few MB; reversible code constant
/// (see module docs for the justification).
const MAXLEN_APPROX: usize = 10_000;

/// Bounded queue capacity. 256 is large enough to absorb a burst of
/// lifecycle events from a graceful-shutdown mass-hangup (~64 sessions
/// × 3 events = ~200) without dropping; small enough to surface a sustained
/// Valkey outage quickly via the `/metrics` drop counter.
const QUEUE_CAPACITY: usize = 256;

/// Valkey-backed [`EventSink`]. Cloneable (the inner is all [`Arc`]) so
/// `main.rs` can hand clones to the media thread + the `/metrics` reader.
pub struct ValkeyEventSink {
    /// Bounded channel into the background Valkey publisher. The media
    /// thread calls `try_send` on this — never `send` — so it never waits.
    tx: mpsc::Sender<CallEvent>,
    /// Shared drop counter. Incremented from both `emit()` (queue full /
    /// closed) and the background task (XADD failure). [`Ordering::Relaxed`]
    /// is correct because the counter is standalone; no other memory must
    /// be synchronized with it.
    drops: Arc<AtomicU64>,
    /// Test-only override of the stream key. A [`Mutex`] (not a tokio
    /// mutex) because it must be set synchronously from the integration
    /// test, which is not async at that point.
    stream_key: Arc<Mutex<String>>,
}

impl ValkeyEventSink {
    /// Spawn the background XADD consumer task + return an [`Arc`] handle.
    ///
    /// The caller (`main.rs`) stores this in [`MediaThreadOpts::sink`](
    /// crate::media_thread::MediaThreadOpts), replacing the default
    /// [`TracingEventSink`]. The concrete [`Arc`]`<ValkeyEventSink>`
    /// coerces to [`SharedEventSink`](crate::event_sink::SharedEventSink)
    /// at the assignment site.
    pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> Arc<Self> {
        Self::spawn_with_stream_key(url, tokio_handle, STREAM_KEY.to_string())
    }

    /// Internal constructor used by both [`Self::spawn`] and the test
    /// helper. Factored out so tests can inject a unique stream key
    /// without duplicating the channel / task wiring.
    fn spawn_with_stream_key(
        url: url::Url,
        tokio_handle: tokio::runtime::Handle,
        initial_key: String,
    ) -> Arc<Self> {
        let (tx, rx) = mpsc::channel(QUEUE_CAPACITY);
        let drops = Arc::new(AtomicU64::new(0));
        let stream_key = Arc::new(Mutex::new(initial_key));

        let drops_for_task = drops.clone();
        let stream_key_for_task = stream_key.clone();
        tokio_handle.spawn(background_consumer(
            url,
            rx,
            drops_for_task,
            stream_key_for_task,
        ));

        Arc::new(Self {
            tx,
            drops,
            stream_key,
        })
    }

    /// Override the stream key for integration tests. Each test gets a
    /// unique key so parallel runs do not collide.
    ///
    /// Public (not `#[cfg(test)]`) because Cargo integration tests link
    /// against the library as a regular crate and cannot see `cfg(test)`
    /// items.
    pub fn set_stream_key_for_test(&self, key: &str) {
        // Poisoned mutex recovery: even if another test panicked while
        // holding the lock, the inner String is still valid.
        let mut guard = match self.stream_key.lock() {
            Ok(g) => g,
            Err(poisoned) => poisoned.into_inner(),
        };
        *guard = key.to_string();
    }

    /// Spawn a sink whose consumer deliberately drops every event. Used by
    /// `tests/valkey_sink_lifecycle.rs` to prove the drop counter is
    /// incremented when the consumer is dead/shedding.
    ///
    /// Public (not `#[cfg(test)]`) so the integration test can call it; see
    /// [`Self::set_stream_key_for_test`] for why.
    pub fn spawn_for_test_with_dead_consumer(
        _url: url::Url,
        tokio_handle: tokio::runtime::Handle,
    ) -> Arc<Self> {
        let (tx, mut rx) = mpsc::channel(QUEUE_CAPACITY);
        let drops = Arc::new(AtomicU64::new(0));
        let drops_for_task = drops.clone();

        tokio_handle.spawn(async move {
            while let Some(_event) = rx.recv().await {
                drops_for_task.fetch_add(1, Ordering::Relaxed);
            }
        });

        Arc::new(Self {
            tx,
            drops,
            stream_key: Arc::new(Mutex::new(STREAM_KEY.to_string())),
        })
    }
}

impl EventSink for ValkeyEventSink {
    /// One `try_send` — that's the entire hot-path surface.
    ///
    /// If the bounded queue is full or the background task has exited,
    /// we increment the drop counter and move on. No await, no I/O, no
    /// allocation beyond the event already owned by the caller.
    fn emit(&self, event: CallEvent) {
        if let Err(e) = self.tx.try_send(event) {
            // Full = backpressure from a slow/dead Valkey. Closed = the
            // background consumer exited. Both are "dropped event".
            let reason = match e {
                mpsc::error::TrySendError::Full(_) => "queue full",
                mpsc::error::TrySendError::Closed(_) => "consumer closed",
            };
            self.drops.fetch_add(1, Ordering::Relaxed);
            tracing::warn!(
                target: "rutster::events",
                reason,
                "ValkeyEventSink: dropped event on emit"
            );
        }
    }

    /// Override the default trait method so `/metrics` can surface the
    /// drop count as `rutster_event_sink_drops_total`.
    fn drops(&self) -> Option<u64> {
        Some(self.drops.load(Ordering::Relaxed))
    }
}

/// Background Valkey publisher. Runs on the tokio runtime for the life
/// of the [`ValkeyEventSink`]; exits cleanly when the channel closes.
async fn background_consumer(
    url: url::Url,
    mut rx: mpsc::Receiver<CallEvent>,
    drops: Arc<AtomicU64>,
    stream_key: Arc<Mutex<String>>,
) {
    let client = match redis::Client::open(url.as_str()) {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(
                error = %e,
                "ValkeyEventSink: failed to open Redis client; consumer exiting"
            );
            return;
        }
    };
    let mut conn = match client.get_multiplexed_async_connection().await {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(
                error = %e,
                "ValkeyEventSink: failed to connect to Valkey; consumer exiting"
            );
            return;
        }
    };

    while let Some(event) = rx.recv().await {
        // Clone the key so we don't hold the mutex guard across await.
        // If the mutex is poisoned, recover the inner String — it is
        // still valid — and keep publishing. This is an observer path;
        // panicking here would not help the media thread.
        let key = match stream_key.lock() {
            Ok(guard) => guard.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        };
        let payload = serialize_event(&event);

        // CRITICAL API: redis 0.27's xadd_maxlen takes a StreamMaxlen enum,
        // NOT a bool. `StreamMaxlen::Approx(N)` emits `MAXLEN ~ N`.
        let result: redis::RedisResult<()> = redis::AsyncCommands::xadd_maxlen(
            &mut conn,
            &key,
            redis::streams::StreamMaxlen::Approx(MAXLEN_APPROX),
            "*",
            &[("event", payload)],
        )
        .await;

        if let Err(e) = result {
            tracing::warn!(
                target: "rutster::events",
                error = %e,
                stream_key = %key,
                "ValkeyEventSink: XADD failed; dropping event"
            );
            drops.fetch_add(1, Ordering::Relaxed);
        }
    }

    tracing::info!(target: "rutster::events", "ValkeyEventSink: background consumer exiting");
}

/// Serialize a lifecycle event to a compact JSON payload for the stream.
///
/// We build the JSON manually with `serde_json::json!` because
/// [`SystemTime`] is not `Serialize` in the standard library, and the
/// public `CallEvent` type intentionally lives in `event_sink.rs` without
/// a serde dependency. The payload shape is stable: `event`, `id`, and
/// timestamps.
fn serialize_event(event: &CallEvent) -> String {
    let json = match event {
        CallEvent::SessionRegistered { id, at } => serde_json::json!({
            "event": "SessionRegistered",
            "id": id.to_string(),
            "at": system_time_secs(*at),
        }),
        CallEvent::SessionConnected { id, at } => serde_json::json!({
            "event": "SessionConnected",
            "id": id.to_string(),
            "at": system_time_secs(*at),
        }),
        CallEvent::SessionEnded {
            id,
            started_at,
            ended_at,
            reason,
            tap_metrics,
        } => serde_json::json!({
            "event": "SessionEnded",
            "id": id.to_string(),
            "started_at": system_time_secs(*started_at),
            "ended_at": system_time_secs(*ended_at),
            "reason": reason_label(*reason),
            "tap_metrics": tap_metrics,
        }),
    };
    json.to_string()
}

/// Return the seconds since the Unix epoch as an `f64`. Times before the
/// epoch clamp to 0.0 (not a real runtime concern for "now" timestamps).
fn system_time_secs(t: SystemTime) -> f64 {
    t.duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs_f64()
}

/// Human-readable, stable label for the [`EndReason`] variant.
fn reason_label(reason: EndReason) -> &'static str {
    match reason {
        EndReason::Deleted => "deleted",
        EndReason::Closed => "closed",
        EndReason::Shutdown => "shutdown",
    }
}

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

    #[test]
    fn serialize_event_produces_expected_shape() {
        let id = rutster_call_model::ChannelId::new();
        let at = SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
        let event = CallEvent::SessionRegistered { id, at };
        let s = serialize_event(&event);
        assert!(s.contains("SessionRegistered"));
        assert!(s.contains(&id.to_string()));
    }

    #[test]
    fn reason_label_exhaustive() {
        // Exhaustiveness check: if EndReason gains a variant, this test
        // stops compiling until we add its label.
        let _ = match EndReason::Deleted {
            EndReason::Deleted => reason_label(EndReason::Deleted),
            EndReason::Closed => reason_label(EndReason::Closed),
            EndReason::Shutdown => reason_label(EndReason::Shutdown),
        };
    }
}
