//! Change signals for CI jobs and runs. [`CiLive`] keeps one generation counter per job and one per
//! run. A write that changes a job or a run bumps its counter, and every subscriber wakes on the
//! change. The counter carries no data: a subscriber reads the database or the log file again when
//! it wakes. Nothing polls. A signal with no receivers left is removed, so the maps hold only the
//! jobs and runs somebody is watching. The hub is in-process only, like [`NotifyHub`].
//!
//! [`NotifyHub`]: crate::notify_hub::NotifyHub

use std::collections::HashMap;
use std::sync::{Mutex, MutexGuard, PoisonError};

use tokio::sync::watch;

/// The change counters of CI jobs and runs, each keyed by its id.
#[derive(Default)]
pub struct CiLive {
    jobs: Signals,
    runs: Signals,
}

impl CiLive {
    /// Subscribes to changes of job `job_id`. The receiver starts as seen, so a bump after this
    /// call wakes it. Subscribe before the state you read, so a change in between is not lost.
    pub fn subscribe_job(&self, job_id: &str) -> watch::Receiver<u64> {
        self.jobs.subscribe(job_id)
    }

    /// Subscribes to changes of run `run_id`. Same rules as [`CiLive::subscribe_job`].
    // No caller yet: the run status stream that waits on this is the next change.
    #[allow(dead_code)]
    pub fn subscribe_run(&self, run_id: &str) -> watch::Receiver<u64> {
        self.runs.subscribe(run_id)
    }

    /// Wakes the subscribers of job `job_id`: its log grew or its state changed. Does nothing when
    /// nobody is subscribed.
    pub fn bump_job(&self, job_id: &str) {
        self.jobs.bump(job_id);
    }

    /// Wakes the subscribers of run `run_id`: one of its jobs or its own status changed.
    pub fn bump_run(&self, run_id: &str) {
        self.runs.bump(run_id);
    }

    /// Wakes the subscribers of job `job_id` once more and drops its signal. Call it when the job
    /// has reached a terminal state: no change follows, so the signal is not kept. A subscriber
    /// that waits on a closed signal wakes once and then sees it closed.
    pub fn close_job(&self, job_id: &str) {
        self.jobs.close(job_id);
    }
}

/// One keyed set of counters. A key has an entry only while it has a receiver or a bump is due.
#[derive(Default)]
struct Signals {
    counters: Mutex<HashMap<String, watch::Sender<u64>>>,
}

impl Signals {
    fn counters(&self) -> MutexGuard<'_, HashMap<String, watch::Sender<u64>>> {
        self.counters.lock().unwrap_or_else(PoisonError::into_inner)
    }

    fn subscribe(&self, key: &str) -> watch::Receiver<u64> {
        self.counters()
            .entry(key.to_owned())
            .or_insert_with(|| watch::channel(0).0)
            .subscribe()
    }

    /// Counts one change. Without a subscriber there is no entry, so the bump is a no-op. An entry
    /// whose last receiver is gone is removed here.
    fn bump(&self, key: &str) {
        let mut counters = self.counters();
        let Some(tx) = counters.get(key) else {
            return;
        };
        tx.send_modify(|generation| *generation = generation.wrapping_add(1));
        if tx.receiver_count() == 0 {
            counters.remove(key);
        }
    }

    /// Counts one last change and removes the entry. The sender is dropped afterwards, so each
    /// subscriber wakes once more and then finds the signal closed.
    fn close(&self, key: &str) {
        let removed = self.counters().remove(key);
        if let Some(tx) = removed {
            tx.send_modify(|generation| *generation = generation.wrapping_add(1));
        }
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;

    #[tokio::test]
    async fn a_bump_wakes_only_the_subscribers_of_that_job() {
        let live = CiLive::default();
        let mut job = live.subscribe_job("job-1");
        let other = live.subscribe_job("job-2");
        let run = live.subscribe_run("run-1");

        live.bump_job("job-1");

        tokio::time::timeout(Duration::from_secs(1), job.changed())
            .await
            .expect("woken")
            .expect("open");
        assert_eq!(*job.borrow_and_update(), 1);
        assert!(!other.has_changed().expect("open"));
        assert!(!run.has_changed().expect("open"));
    }

    #[tokio::test]
    async fn a_bump_before_subscribing_is_not_seen_and_a_bump_after_is() {
        let live = CiLive::default();
        let early = live.subscribe_run("run-1");
        live.bump_run("run-1");
        let mut late = live.subscribe_run("run-1");
        assert!(early.has_changed().expect("open"));
        assert!(!late.has_changed().expect("open"));
        live.bump_run("run-1");
        assert!(late.has_changed().expect("open"));
        late.changed().await.expect("open");
        assert_eq!(*late.borrow_and_update(), 2);
    }

    #[test]
    fn a_bump_without_subscribers_keeps_no_entry() {
        let live = CiLive::default();
        live.bump_job("job-1");
        live.bump_run("run-1");
        assert!(live.jobs.counters().is_empty());
        assert!(live.runs.counters().is_empty());
    }

    #[test]
    fn a_bump_drops_the_entry_once_its_last_receiver_is_gone() {
        let live = CiLive::default();
        let rx = live.subscribe_job("job-1");
        assert_eq!(live.jobs.counters().len(), 1);
        drop(rx);
        live.bump_job("job-1");
        assert!(live.jobs.counters().is_empty());
    }

    #[tokio::test]
    async fn close_wakes_the_subscriber_once_and_then_ends_the_signal() {
        let live = CiLive::default();
        let mut rx = live.subscribe_job("job-1");
        live.close_job("job-1");
        assert!(live.jobs.counters().is_empty());

        rx.changed().await.expect("the last change is delivered");
        assert_eq!(*rx.borrow_and_update(), 1);
        assert!(rx.changed().await.is_err());
    }
}