//! # Collector supervision
//!
//! The daemon runs unattended for weeks, so every way a collector can stop has
//! to be a way it can start again — and when it cannot, that has to reach the
//! bus rather than a log file nobody reads.
//!
//! Two failure shapes need two mechanisms. The git and terminal collectors are
//! function calls on one producer thread: a panic in either would take the
//! thread down while `publish::serve` kept handing subscribers a frozen cache,
//! so each call is wrapped and a panic costs one tick. The hardware collector
//! is a child process that can die on its own, so it gets a respawn loop.
//!
//! Health is published as [`SignalName::CollectorUp`], once per [`Source`], on
//! transitions only — publishing `up` every tick would put three redundant
//! frames per tick on the bus for no reader's benefit.

use std::collections::BTreeMap;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use signal_schema::{Signal, SignalName, Source, Value, SCHEMA_VERSION};

use crate::collectors;
use crate::hub::Hub;
use crate::now_millis;

/// First delay after a collector exits.
pub const BACKOFF_START: Duration = Duration::from_secs(1);
/// Ceiling on the retry delay. There is no retry limit: a transient failure at
/// hour three must not disable a collector until a human notices.
pub const BACKOFF_CAP: Duration = Duration::from_secs(60);
/// A run lasting at least this long is treated as a success, and resets the
/// delay. Without it a child that flaps ratchets the delay to the ceiling and
/// stays there.
pub const BACKOFF_RESET_AFTER: Duration = Duration::from_secs(30);

/// Retry delays for a collector that keeps exiting.
#[derive(Debug, Clone, Copy)]
pub struct Backoff {
    next: Duration,
}

impl Default for Backoff {
    fn default() -> Self {
        Backoff { next: BACKOFF_START }
    }
}

impl Backoff {
    /// Record that a run ended after `ran_for`, and return how long to wait
    /// before the next attempt.
    pub fn record_exit(&mut self, ran_for: Duration) -> Duration {
        if ran_for >= BACKOFF_RESET_AFTER {
            self.next = BACKOFF_START;
        }
        let delay = self.next;
        self.next = (self.next * 2).min(BACKOFF_CAP);
        delay
    }
}

/// Publishes [`SignalName::CollectorUp`] when a collector's state changes.
///
/// Shared between the producer thread and the hardware supervisor, so both
/// report through one place and neither can publish a redundant frame.
#[derive(Clone, Default)]
pub struct Health {
    last: Arc<Mutex<BTreeMap<u8, bool>>>,
}

impl Health {
    pub fn new() -> Health {
        Health::default()
    }

    /// Publish `up` for `source` if it differs from what was last published.
    /// The first call for a source always publishes, so a subscriber that
    /// connects later reads the truth out of the last-value cache.
    pub fn set(&self, hub: &Hub, source: Source, up: bool) {
        let mut last = self.last.lock().unwrap();
        if last.get(&source.to_u8()) == Some(&up) {
            return;
        }
        last.insert(source.to_u8(), up);
        drop(last);
        hub.publish(Signal {
            schema_version: SCHEMA_VERSION,
            ts: now_millis(),
            source,
            name: SignalName::CollectorUp,
            value: Value(if up { 1.0 } else { 0.0 }),
            tag: None,
        });
    }
}

/// Run `f`, catching a panic and reporting it as `source` going down.
///
/// A panic inside a collector used to kill the producer thread outright. The
/// daemon stayed up and the socket kept serving, so a face went on rendering
/// the last values it had — plausible numbers that had stopped being true.
fn guarded<T>(hub: &Hub, health: &Health, source: Source, f: impl FnOnce() -> T) -> Option<T> {
    match catch_unwind(AssertUnwindSafe(f)) {
        Ok(v) => {
            health.set(hub, source, true);
            Some(v)
        }
        Err(_) => {
            eprintln!("signald: {source:?} collector panicked; continuing");
            health.set(hub, source, false);
            None
        }
    }
}

/// The producer loop: poll the git and terminal collectors on a tick and
/// publish into `hub`. Runs for the life of the daemon.
///
/// A panic in either collector costs that collector one tick, not the thread.
/// The terminal collector holds state across ticks, so a panic replaces it with
/// a fresh one — that costs one window of rate data and nothing else.
pub fn run_producer(
    hub: Hub,
    health: Health,
    repos: Vec<PathBuf>,
    spool: PathBuf,
    interval: Duration,
) -> ! {
    let mut terminal = collectors::terminal::Collector::new();
    loop {
        let signals = guarded(&hub, &health, Source::Git, || {
            repos
                .iter()
                .flat_map(|repo| collectors::git::collect(repo))
                .filter(|sig| crate::tag_within_roots(sig, &repos))
                .collect::<Vec<_>>()
        });
        for sig in signals.unwrap_or_default() {
            hub.publish(sig);
        }

        match guarded(&hub, &health, Source::Terminal, || terminal.collect(&spool)) {
            Some(signals) => {
                for sig in signals {
                    hub.publish(sig);
                }
            }
            None => terminal = collectors::terminal::Collector::new(),
        }

        std::thread::sleep(interval);
    }
}

/// Keep the hardware collector running, restarting it with backoff.
///
/// The child used to be spawned once and never again: if it died the daemon
/// stayed up, so launchd's `KeepAlive` never fired and hardware signals stopped
/// for good.
pub fn supervise_hardware(hub: Hub, health: Health, program: PathBuf, interval_ms: u64) -> ! {
    let mut backoff = Backoff::default();
    loop {
        let started = Instant::now();
        let up = || report_hardware_up(&hub, &health);
        match collectors::hardware::run(&program, interval_ms, &hub, up) {
            Ok(n) => eprintln!("signald: hardware collector ended after {n} frame(s)"),
            Err(e) => eprintln!("signald: hardware collector failed: {e}"),
        }
        health.set(&hub, Source::Hardware, false);
        health.set(&hub, Source::Macos, false);
        let delay = backoff.record_exit(started.elapsed());
        eprintln!("signald: restarting hardware collector in {}s", delay.as_secs());
        std::thread::sleep(delay);
    }
}

/// Report a collector that was never configured, once.
///
/// Absent is not the same as failed: there is nothing to retry, and spinning on
/// a binary that does not exist is not resilience.
pub fn report_unconfigured(hub: &Hub, health: &Health, program: &str) {
    eprintln!("signald: {program} not found; hardware signals disabled");
    health.set(hub, Source::Hardware, false);
    health.set(hub, Source::Macos, false);
}

/// Mark the hardware collector up. Called once the child is producing frames.
pub fn report_hardware_up(hub: &Hub, health: &Health) {
    health.set(hub, Source::Hardware, true);
    health.set(hub, Source::Macos, true);
}

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

    #[test]
    fn backoff_doubles_to_the_cap() {
        let mut b = Backoff::default();
        let brief = Duration::from_secs(1);
        assert_eq!(b.record_exit(brief), Duration::from_secs(1));
        assert_eq!(b.record_exit(brief), Duration::from_secs(2));
        assert_eq!(b.record_exit(brief), Duration::from_secs(4));
        for _ in 0..10 {
            b.record_exit(brief);
        }
        assert_eq!(b.record_exit(brief), BACKOFF_CAP, "delay is capped");
    }

    #[test]
    fn a_run_that_survives_resets_the_delay() {
        let mut b = Backoff::default();
        let brief = Duration::from_secs(1);
        for _ in 0..5 {
            b.record_exit(brief);
        }
        assert_eq!(
            b.record_exit(BACKOFF_RESET_AFTER),
            BACKOFF_START,
            "a child that ran long enough starts over rather than staying at the ceiling"
        );
    }

    #[test]
    fn health_publishes_only_on_change() {
        let hub = Hub::new();
        let health = Health::new();
        health.set(&hub, Source::Git, true);
        health.set(&hub, Source::Git, true);
        health.set(&hub, Source::Git, true);
        assert_eq!(hub.snapshot().len(), 1, "no redundant frames");

        health.set(&hub, Source::Git, false);
        let snap = hub.snapshot();
        assert_eq!(snap.len(), 1, "same name and source is keep-latest");
        assert_eq!(snap[0].value, Value(0.0));
    }

    #[test]
    fn health_is_per_source() {
        let hub = Hub::new();
        let health = Health::new();
        health.set(&hub, Source::Git, true);
        health.set(&hub, Source::Hardware, false);
        assert_eq!(hub.snapshot().len(), 2, "one entry per source");
    }

    #[test]
    fn a_panicking_collector_reports_down_and_does_not_unwind() {
        let hub = Hub::new();
        let health = Health::new();
        let out = guarded(&hub, &health, Source::Terminal, || panic!("collector broke"));
        assert!(out.is_none());
        let snap = hub.snapshot();
        assert_eq!(snap[0].value, Value(0.0), "terminal reported down");
        assert_eq!(snap[0].source, Source::Terminal);

        let out = guarded(&hub, &health, Source::Terminal, || 7);
        assert_eq!(out, Some(7), "the next tick still runs");
        assert_eq!(hub.snapshot()[0].value, Value(1.0), "and reports up again");
    }
}