//! # The signal hub — last-value cache + live fan-out
//!
//! The in-memory heart of the publish side. Collectors [`Hub::publish`] signals;
//! subscribers [`Hub::subscribe`] and receive the current value of every cached
//! metric immediately, then a live stream of updates.
//!
//! Subscription semantics: *last-value-cache + live stream.* On
//! subscribe, the caller is handed the latest value of each `name` (keyed by
//! name + audited tag), so a renderer paints correct state at once; thereafter
//! it receives updates as they are published. Channels are unbounded, so a slow
//! subscriber never blocks a publisher.
//!
//! When a `Hub` is built [`Hub::with_history`], every published signal is also
//! persisted to the SQLite history store, so live and persisted paths share one
//! write point.

use std::collections::BTreeMap;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, Mutex};

use signal_schema::Signal;

use crate::history::History;

/// Cache key: metric name discriminant, source discriminant, and optional
/// audited tag. Distinct per-repo signals coexist, and so do two signals that
/// share a name but come from different collectors — `SignalName::CollectorUp`
/// is published once per `Source`, and without `source` in the key those
/// would overwrite each other. A new value for the same key replaces the old
/// one (keep-latest).
type Key = (u8, u8, Option<String>);

/// A cloneable handle to the shared hub. Clones share one cache + subscriber
/// set behind a mutex.
#[derive(Clone)]
pub struct Hub {
    inner: Arc<Mutex<Inner>>,
}

struct Inner {
    cache: BTreeMap<Key, Signal>,
    subs: Vec<Sender<Signal>>,
    history: Option<History>,
}

impl Hub {
    /// A hub with no history (live-only). Useful for tests and for a renderer
    /// that never persists.
    pub fn new() -> Hub {
        Hub::build(None)
    }

    /// A hub that also persists every published signal to `history`.
    pub fn with_history(history: History) -> Hub {
        Hub::build(Some(history))
    }

    fn build(history: Option<History>) -> Hub {
        Hub {
            inner: Arc::new(Mutex::new(Inner {
                cache: BTreeMap::new(),
                subs: Vec::new(),
                history,
            })),
        }
    }

    /// Publish one signal: persist it (if a history is attached), update the
    /// last-value cache, and fan it out to every live subscriber. Dead
    /// subscribers (whose receiver was dropped) are pruned here.
    pub fn publish(&self, s: Signal) {
        let mut guard = self.inner.lock().unwrap();
        let Inner {
            cache,
            subs,
            history,
        } = &mut *guard;

        if let Some(h) = history {
            // A history write failing must not stop the live stream.
            let _ = h.record(&s);
        }
        let key = (
            s.name.to_u8(),
            s.source.to_u8(),
            s.tag.as_ref().map(|t| t.as_str().to_string()),
        );
        cache.insert(key, s.clone());
        subs.retain(|tx| tx.send(s.clone()).is_ok());
    }

    /// Subscribe: return the current last-value snapshot plus a receiver that
    /// yields every signal published after this call.
    pub fn subscribe(&self) -> (Vec<Signal>, Receiver<Signal>) {
        let mut guard = self.inner.lock().unwrap();
        let snapshot = guard.cache.values().cloned().collect();
        let (tx, rx) = channel();
        guard.subs.push(tx);
        (snapshot, rx)
    }

    /// The current last-value snapshot without subscribing.
    pub fn snapshot(&self) -> Vec<Signal> {
        self.inner.lock().unwrap().cache.values().cloned().collect()
    }
}

impl Default for Hub {
    fn default() -> Self {
        Hub::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use signal_schema::{SignalName, Source, Value, SCHEMA_VERSION};

    fn sig(name: SignalName, value: f64) -> Signal {
        Signal {
            schema_version: SCHEMA_VERSION,
            ts: 1,
            source: Source::Terminal,
            name,
            value: Value(value),
            tag: None,
        }
    }

    #[test]
    fn subscriber_gets_cached_value_then_live_update() {
        let hub = Hub::new();
        // A value published before anyone subscribes must still be seen (cache).
        hub.publish(sig(SignalName::KeysPerMin, 10.0));

        let (snapshot, rx) = hub.subscribe();
        assert_eq!(snapshot.len(), 1);
        assert_eq!(snapshot[0].value, Value(10.0));

        // A change after subscribing arrives live.
        hub.publish(sig(SignalName::KeysPerMin, 25.0));
        let update = rx.recv().expect("live update");
        assert_eq!(update.name, SignalName::KeysPerMin);
        assert_eq!(update.value, Value(25.0));
    }

    #[test]
    fn cache_keeps_latest_per_name_and_source() {
        let hub = Hub::new();
        hub.publish(sig(SignalName::KeysPerMin, 1.0));
        hub.publish(sig(SignalName::KeysPerMin, 2.0));
        let snap = hub.snapshot();
        assert_eq!(snap.len(), 1, "same name and source collapses to keep-latest");
        assert_eq!(snap[0].value, Value(2.0));
    }

    /// `CollectorUp` is published once per source, so the cache must not treat
    /// the name alone as the identity. Without `source` in the key the second
    /// publish would evict the first and a subscriber would see one collector's
    /// health standing in for all three.
    #[test]
    fn same_name_from_two_sources_coexists() {
        let hub = Hub::new();
        for (source, value) in [(Source::Git, 1.0), (Source::Hardware, 0.0)] {
            hub.publish(Signal {
                source,
                ..sig(SignalName::CollectorUp, value)
            });
        }
        let snap = hub.snapshot();
        assert_eq!(snap.len(), 2, "one entry per source");
        let git = snap.iter().find(|s| s.source == Source::Git).expect("git health");
        let hw = snap.iter().find(|s| s.source == Source::Hardware).expect("hardware health");
        assert_eq!(git.value, Value(1.0));
        assert_eq!(hw.value, Value(0.0));
    }
}