//! Central availability monitor for saved servers: one Core task, native TCP connects, no login.
//! `Scheduler` holds the state and the timing (no I/O, tested with explicit instants);
//! `AvailabilityMonitor` runs it. The state stays in memory and is wiped when the vault locks.
//! A result says "the SSH port answers from this network"; it never proves a server is off.

use futures::future::BoxFuture;
use parking_lot::Mutex;
use serde_json::{json, Value};
use std::io::ErrorKind;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::net::TcpStream;
use tokio::sync::Notify;
use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;

pub const DEFAULT_INTERVAL_SECS: u64 = 12;
pub const INTERVALS_SECS: [u64; 6] = [5, 10, 12, 15, 30, 60];
pub const TCP_TIMEOUT: Duration = Duration::from_millis(1500);
pub const JUMP_TIMEOUT: Duration = Duration::from_secs(3);
pub const RETRY_DELAY: Duration = Duration::from_secs(2);
pub const FAILURES_FOR_OFFLINE: u32 = 2;
pub const MAX_CONCURRENT: usize = 6;
const DNS_TIMEOUT: Duration = Duration::from_secs(3);
const MAX_ADDRESSES: usize = 3;
/// Spacing of an immediate round; later rounds spread over the whole interval.
const BURST_SPACING: Duration = Duration::from_millis(120);
const HEARTBEAT: Duration = Duration::from_secs(1);
/// Backstop for profile changes that do not call `changed()`.
const RESYNC_EVERY: Duration = Duration::from_secs(30);
/// A jump between wall clock and monotonic clock this large means the computer slept.
const SLEEP_GAP: Duration = Duration::from_secs(5);

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Target {
    pub id: String,
    pub host: String,
    pub port: u16,
    pub jump_id: Option<String>,
}

impl Target {
    /// Only the routing fields of a profile; the monitor never touches credentials.
    pub fn from_profile(profile: &Value) -> Option<Self> {
        Some(Self {
            id: profile["id"].as_str().filter(|id| !id.is_empty())?.to_string(),
            host: profile["host"].as_str().filter(|host| !host.is_empty())?.to_string(),
            port: u16::try_from(profile["port"].as_u64()?).ok()?,
            jump_id: profile["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(str::to_string),
        })
    }
}

#[derive(Clone, Debug, PartialEq)]
pub enum Outcome {
    Online { ms: f64 },
    Failed { reason: &'static str },
    /// The route could not be tested; says nothing about the server.
    Unverified { reason: &'static str },
}

struct Entry {
    target: Target,
    generation: u64,
    status: &'static str,
    latency_ms: Option<f64>,
    last_success: Option<u64>,
    last_attempt: Option<u64>,
    failures: u32,
    reason: Option<&'static str>,
    next_due: Option<Instant>,
    in_flight: bool,
    /// Set by an immediate round: the next regular check takes this entry's phase in the interval.
    spread: bool,
}

impl Entry {
    fn value(&self) -> Value {
        json!({
            "id": self.target.id,
            "status": self.status,
            "latencyMs": self.latency_ms,
            "lastSuccessAt": self.last_success,
            "lastAttemptAt": self.last_attempt,
            "failures": self.failures,
            "reason": self.reason,
            "via": if self.target.jump_id.is_some() { "jump" } else { "direct" }
        })
    }
}

pub struct Scheduler {
    entries: Vec<Entry>,
    enabled: bool,
    interval: Duration,
    next_generation: u64,
}

impl Default for Scheduler {
    fn default() -> Self {
        Self { entries: Vec::new(), enabled: true, interval: Duration::from_secs(DEFAULT_INTERVAL_SECS), next_generation: 1 }
    }
}

impl Scheduler {
    fn generation(&mut self) -> u64 {
        self.next_generation += 1;
        self.next_generation
    }

    /// Applies the saved profiles. Returns what the UI must hear about: changed entries and removed IDs.
    pub fn sync(&mut self, targets: Vec<Target>, now: Instant) -> (Vec<Value>, Vec<String>) {
        let removed: Vec<String> = self.entries.iter().filter(|e| !targets.iter().any(|t| t.id == e.target.id)).map(|e| e.target.id.clone()).collect();
        self.entries.retain(|e| targets.iter().any(|t| t.id == e.target.id));
        let mut changed = Vec::new();
        let mut delay = Duration::ZERO;
        for target in targets {
            let position = self.entries.iter().position(|e| e.target.id == target.id);
            if position.is_some_and(|i| self.entries[i].target == target) {
                continue;
            }
            let generation = self.generation();
            let enabled = self.enabled;
            let fresh = Entry {
                target,
                generation,
                status: if enabled { "checking" } else { "unknown" },
                latency_ms: None,
                last_success: None,
                last_attempt: None,
                failures: 0,
                reason: None,
                next_due: enabled.then_some(now + delay),
                in_flight: false,
                spread: true,
            };
            delay += BURST_SPACING;
            changed.push(fresh.value());
            match position {
                Some(i) => self.entries[i] = fresh,
                None => self.entries.push(fresh),
            }
        }
        (changed, removed)
    }

    /// Returns true when the state visible to the UI changed.
    pub fn set_policy(&mut self, enabled: bool, interval: Duration, now: Instant) -> bool {
        let was_enabled = self.enabled;
        self.enabled = enabled;
        self.interval = interval;
        if enabled && !was_enabled {
            self.refresh_all(now);
            for entry in &mut self.entries {
                entry.status = "checking";
            }
            return !self.entries.is_empty();
        }
        if !enabled && was_enabled {
            for i in 0..self.entries.len() {
                let generation = self.generation();
                let entry = &mut self.entries[i];
                entry.generation = generation;
                entry.status = "unknown";
                entry.latency_ms = None;
                entry.failures = 0;
                entry.reason = None;
                entry.next_due = None;
                entry.in_flight = false;
            }
            return !self.entries.is_empty();
        }
        // A shorter interval must not wait for the old, longer one.
        for entry in &mut self.entries {
            if let Some(due) = entry.next_due {
                entry.next_due = Some(due.min(now + interval));
            }
        }
        false
    }

    /// Immediate round for every server, spaced so that they do not all start at one moment.
    pub fn refresh_all(&mut self, now: Instant) {
        if !self.enabled {
            return;
        }
        for (i, entry) in self.entries.iter_mut().enumerate() {
            entry.next_due = Some(now + BURST_SPACING * i as u32);
            entry.spread = true;
        }
    }

    /// Takes up to `slots` due checks and marks them as running.
    pub fn take_due(&mut self, now: Instant, slots: usize) -> Vec<(Target, u64)> {
        let mut due = Vec::new();
        for entry in &mut self.entries {
            if due.len() == slots {
                break;
            }
            if !entry.in_flight && entry.next_due.is_some_and(|at| at <= now) {
                entry.in_flight = true;
                entry.next_due = None;
                due.push((entry.target.clone(), entry.generation));
            }
        }
        due
    }

    /// Records a finished check. A result for an older address or a removed server is dropped.
    pub fn apply(&mut self, id: &str, generation: u64, outcome: Outcome, now: Instant, wall_ms: u64) -> Option<Value> {
        let enabled = self.enabled;
        let (interval, count) = (self.interval, self.entries.len().max(1) as u32);
        let index = self.entries.iter().position(|e| e.target.id == id && e.generation == generation)?;
        let entry = &mut self.entries[index];
        entry.in_flight = false;
        entry.last_attempt = Some(wall_ms);
        let mut retry = false;
        match outcome {
            Outcome::Online { ms } => {
                entry.status = "online";
                entry.latency_ms = Some(ms);
                entry.last_success = Some(wall_ms);
                entry.failures = 0;
                entry.reason = None;
            }
            Outcome::Failed { reason } => {
                entry.failures += 1;
                entry.reason = Some(reason);
                if entry.failures >= FAILURES_FOR_OFFLINE {
                    entry.status = "offline";
                    entry.latency_ms = None;
                } else {
                    retry = true;
                }
            }
            Outcome::Unverified { reason } => {
                entry.status = "unverified";
                entry.latency_ms = None;
                entry.failures = 0;
                entry.reason = Some(reason);
            }
        }
        entry.next_due = enabled.then(|| {
            if retry {
                return now + RETRY_DELAY;
            }
            let phase = if entry.spread { interval * index as u32 / count } else { Duration::ZERO };
            entry.spread = false;
            now + interval + phase
        });
        Some(entry.value())
    }

    pub fn next_due(&self) -> Option<Instant> {
        self.entries.iter().filter(|e| !e.in_flight).filter_map(|e| e.next_due).min()
    }

    pub fn snapshot(&self) -> Vec<Value> {
        self.entries.iter().map(Entry::value).collect()
    }

    pub fn clear(&mut self) {
        self.entries.clear();
    }
}

fn failure_reason(error: &std::io::Error) -> &'static str {
    match error.kind() {
        ErrorKind::ConnectionRefused => "refused",
        ErrorKind::TimedOut => "timeout",
        ErrorKind::HostUnreachable | ErrorKind::NetworkUnreachable | ErrorKind::AddrNotAvailable => "unreachable",
        ErrorKind::PermissionDenied => "denied",
        _ => "error",
    }
}

/// TCP connect only: the socket closes right after the handshake, before any SSH traffic.
pub async fn probe_tcp(host: &str, port: u16, timeout: Duration) -> Outcome {
    let addresses: Vec<_> = match tokio::time::timeout(DNS_TIMEOUT, tokio::net::lookup_host((host, port))).await {
        Ok(Ok(found)) => found.take(MAX_ADDRESSES).collect(),
        _ => return Outcome::Failed { reason: "dns" },
    };
    let mut reason = "dns";
    for address in addresses {
        let start = Instant::now();
        match tokio::time::timeout(timeout, TcpStream::connect(address)).await {
            Ok(Ok(_stream)) => return Outcome::Online { ms: (start.elapsed().as_secs_f64() * 10_000.0).round() / 10.0 },
            Ok(Err(error)) => reason = failure_reason(&error),
            Err(_) => reason = "timeout",
        }
    }
    Outcome::Failed { reason }
}

/// What the monitor needs from the rest of the Core.
pub trait MonitorHost: Send + Sync + 'static {
    /// False while the vault is locked or no main window exists.
    fn available(&self) -> bool;
    fn settings(&self) -> (bool, Duration);
    fn targets(&self) -> Option<Vec<Target>>;
    fn publish(&self, event: Value);
    /// Checks a target through an SSH session that is already open to its jump host.
    fn probe_jump(&self, jump_id: &str, target: &Target) -> BoxFuture<'static, Outcome>;
}

#[derive(Default)]
pub struct AvailabilityMonitor {
    state: Mutex<Scheduler>,
    wake: Notify,
    resync: AtomicBool,
    refresh: AtomicBool,
    stop: CancellationToken,
}

fn wall_ms() -> u64 {
    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0)
}

impl AvailabilityMonitor {
    pub fn snapshot(&self) -> Vec<Value> {
        self.state.lock().snapshot()
    }

    /// Profiles, settings or the vault state changed.
    pub fn changed(&self) {
        self.resync.store(true, Ordering::SeqCst);
        self.wake.notify_one();
    }

    pub fn refresh(&self) {
        self.refresh.store(true, Ordering::SeqCst);
        self.wake.notify_one();
    }

    /// Ends the task; dropping its JoinSet aborts running checks and closes their sockets.
    pub fn stop(&self) {
        self.stop.cancel();
    }

    pub async fn run(self: Arc<Self>, host: Arc<dyn MonitorHost>) {
        let mut checks: JoinSet<(String, u64, Outcome)> = JoinSet::new();
        let mut active = false;
        let mut last_sync = Instant::now();
        let mut clocks = (Instant::now(), SystemTime::now());
        loop {
            let now = Instant::now();
            let slept = {
                let (mono, wall) = (now.duration_since(clocks.0), clocks.1.elapsed().unwrap_or_default());
                clocks = (now, SystemTime::now());
                wall > mono + SLEEP_GAP || mono > HEARTBEAT + SLEEP_GAP
            };
            if !host.available() {
                if active {
                    active = false;
                    checks.abort_all();
                    self.state.lock().clear();
                    host.publish(json!({ "reset": true, "servers": [] }));
                }
            } else {
                let first = !active;
                active = true;
                let (enabled, interval) = host.settings();
                if self.state.lock().set_policy(enabled, interval, now) {
                    host.publish(json!({ "reset": true, "servers": self.snapshot() }));
                }
                if first || self.resync.swap(false, Ordering::SeqCst) || now.duration_since(last_sync) >= RESYNC_EVERY {
                    last_sync = now;
                    if let Some(targets) = host.targets() {
                        let (changed, removed) = self.state.lock().sync(targets, now);
                        if !changed.is_empty() || !removed.is_empty() {
                            host.publish(json!({ "servers": changed, "removed": removed }));
                        }
                    }
                }
                if self.refresh.swap(false, Ordering::SeqCst) || slept {
                    self.state.lock().refresh_all(now);
                }
                let slots = MAX_CONCURRENT.saturating_sub(checks.len());
                for (target, generation) in self.state.lock().take_due(now, slots) {
                    let id = target.id.clone();
                    let via_jump = target.jump_id.clone().map(|jump_id| host.probe_jump(&jump_id, &target));
                    checks.spawn(async move {
                        let outcome = match via_jump {
                            Some(probe) => probe.await,
                            None => probe_tcp(&target.host, target.port, TCP_TIMEOUT).await,
                        };
                        (id, generation, outcome)
                    });
                }
            }
            let wait = self.state.lock().next_due().map_or(HEARTBEAT, |at| at.saturating_duration_since(Instant::now()).clamp(Duration::from_millis(5), HEARTBEAT));
            tokio::select! {
                _ = self.stop.cancelled() => break,
                _ = self.wake.notified() => {}
                _ = tokio::time::sleep(wait) => {}
                Some(done) = checks.join_next(), if !checks.is_empty() => {
                    if let Ok((id, generation, outcome)) = done {
                        let updated = self.state.lock().apply(&id, generation, outcome, Instant::now(), wall_ms());
                        if let Some(updated) = updated {
                            host.publish(json!({ "servers": [updated] }));
                        }
                    }
                }
            }
        }
    }
}

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

    fn target(id: &str, port: u16) -> Target {
        Target { id: id.into(), host: "server.test".into(), port, jump_id: None }
    }
    const FAIL: Outcome = Outcome::Failed { reason: "timeout" };

    fn one(now: Instant) -> Scheduler {
        let mut scheduler = Scheduler::default();
        scheduler.sync(vec![target("a", 22)], now);
        scheduler
    }
    fn run(scheduler: &mut Scheduler, now: Instant, outcome: Outcome) -> Value {
        let (due, generation) = scheduler.take_due(now, 6).remove(0);
        scheduler.apply(&due.id, generation, outcome, now, 1000).unwrap()
    }

    #[test]
    fn new_server_is_checked_at_once() {
        let now = Instant::now();
        let mut scheduler = one(now);
        assert_eq!(scheduler.snapshot()[0]["status"], "checking");
        assert_eq!(scheduler.take_due(now, 6).len(), 1);
        assert!(scheduler.take_due(now, 6).is_empty(), "a running check is not started twice");
    }

    #[test]
    fn online_offline_online_needs_two_failures_two_seconds_apart() {
        let now = Instant::now();
        let mut scheduler = one(now);
        assert_eq!(run(&mut scheduler, now, Outcome::Online { ms: 2.5 })["status"], "online");
        let at = now + Duration::from_secs(12);
        let first = run(&mut scheduler, at, FAIL);
        assert_eq!((first["status"].as_str(), first["failures"].as_u64()), (Some("online"), Some(1)));
        assert_eq!(scheduler.next_due(), Some(at + RETRY_DELAY));
        let second = run(&mut scheduler, at + RETRY_DELAY, FAIL);
        assert_eq!(second["status"], "offline");
        assert_eq!(second["latencyMs"], Value::Null);
        assert_eq!(second["reason"], "timeout");
        assert_eq!(second["lastSuccessAt"], 1000);
        assert_eq!(scheduler.next_due(), Some(at + RETRY_DELAY + Duration::from_secs(12)), "offline waits for the normal interval");
        let back = run(&mut scheduler, at + Duration::from_secs(40), Outcome::Online { ms: 3.0 });
        assert_eq!((back["status"].as_str(), back["failures"].as_u64(), back["reason"].is_null()), (Some("online"), Some(0), true));
    }

    #[test]
    fn first_ever_failure_stays_checking_until_the_retry() {
        let now = Instant::now();
        let mut scheduler = one(now);
        assert_eq!(run(&mut scheduler, now, FAIL)["status"], "checking");
        assert_eq!(run(&mut scheduler, now + RETRY_DELAY, FAIL)["status"], "offline");
    }

    #[test]
    fn concurrency_is_limited_and_rounds_spread_over_the_interval() {
        let now = Instant::now();
        let mut scheduler = Scheduler::default();
        scheduler.sync((0..12).map(|i| target(&format!("s{i}"), 22)).collect(), now);
        let late = now + Duration::from_secs(5);
        let first = scheduler.take_due(late, MAX_CONCURRENT);
        assert_eq!(first.len(), MAX_CONCURRENT);
        for (t, generation) in first {
            scheduler.apply(&t.id, generation, Outcome::Online { ms: 1.0 }, late, 1).unwrap();
        }
        let second = scheduler.take_due(late, MAX_CONCURRENT);
        assert_eq!(second.len(), MAX_CONCURRENT);
        for (t, generation) in second {
            scheduler.apply(&t.id, generation, Outcome::Online { ms: 1.0 }, late, 1).unwrap();
        }
        let mut times: Vec<Instant> = scheduler.entries.iter().map(|e| e.next_due.unwrap()).collect();
        times.dedup();
        assert_eq!(times.len(), 12, "no two servers share a start");
        assert_eq!(times[1] - times[0], Duration::from_secs(1), "12 servers in 12 s: one per second");
    }

    #[test]
    fn immediate_round_is_spaced() {
        let now = Instant::now();
        let mut scheduler = Scheduler::default();
        scheduler.sync(vec![target("a", 22), target("b", 22)], now);
        assert_eq!(scheduler.take_due(now, 6).len(), 1);
        assert_eq!(scheduler.take_due(now + BURST_SPACING, 6).len(), 1);
    }

    #[test]
    fn changed_address_restarts_and_drops_the_old_result() {
        let now = Instant::now();
        let mut scheduler = one(now);
        let (_, old) = scheduler.take_due(now, 6).remove(0);
        let (changed, removed) = scheduler.sync(vec![target("a", 2222)], now);
        assert_eq!((changed.len(), removed.len()), (1, 0));
        assert_eq!(changed[0]["status"], "checking");
        assert!(scheduler.apply("a", old, Outcome::Online { ms: 1.0 }, now, 1).is_none(), "a result from the old port must not count");
        let (due, _) = scheduler.take_due(now, 6).remove(0);
        assert_eq!(due.port, 2222);
        assert_eq!(scheduler.sync(vec![target("a", 2222)], now).0.len(), 0, "unchanged data does not restart");
    }

    #[test]
    fn removed_server_leaves_no_state() {
        let now = Instant::now();
        let mut scheduler = one(now);
        let (_, generation) = scheduler.take_due(now, 6).remove(0);
        let (changed, removed) = scheduler.sync(vec![], now);
        assert!(changed.is_empty());
        assert_eq!(removed, vec!["a".to_string()]);
        assert!(scheduler.snapshot().is_empty());
        assert!(scheduler.apply("a", generation, FAIL, now, 1).is_none());
        assert_eq!(scheduler.next_due(), None);
    }

    #[test]
    fn disabling_stops_checks_and_marks_unknown() {
        let now = Instant::now();
        let mut scheduler = one(now);
        let (_, generation) = scheduler.take_due(now, 6).remove(0);
        assert!(scheduler.set_policy(false, Duration::from_secs(12), now));
        assert_eq!(scheduler.snapshot()[0]["status"], "unknown");
        assert!(scheduler.apply("a", generation, Outcome::Online { ms: 1.0 }, now, 1).is_none());
        assert_eq!(scheduler.next_due(), None);
        scheduler.refresh_all(now);
        assert!(scheduler.take_due(now, 6).is_empty(), "no checks while disabled");
        assert!(scheduler.set_policy(true, Duration::from_secs(12), now));
        assert_eq!(scheduler.snapshot()[0]["status"], "checking");
        assert_eq!(scheduler.take_due(now, 6).len(), 1);
    }

    #[test]
    fn shorter_interval_takes_effect_at_once() {
        let now = Instant::now();
        let mut scheduler = one(now);
        scheduler.set_policy(true, Duration::from_secs(60), now);
        run(&mut scheduler, now, Outcome::Online { ms: 1.0 });
        scheduler.set_policy(true, Duration::from_secs(5), now);
        assert!(scheduler.next_due().unwrap() <= now + Duration::from_secs(5));
    }

    #[test]
    fn unverified_is_never_online_and_hides_old_latency() {
        let now = Instant::now();
        let mut scheduler = one(now);
        run(&mut scheduler, now, Outcome::Online { ms: 4.0 });
        let value = run(&mut scheduler, now + Duration::from_secs(12), Outcome::Unverified { reason: "noJumpSession" });
        assert_eq!(value["status"], "unverified");
        assert_eq!(value["latencyMs"], Value::Null);
        assert_eq!(value["reason"], "noJumpSession");
    }

    #[test]
    fn profile_fields_become_targets_without_secrets() {
        let parsed = Target::from_profile(&json!({ "id": "x", "host": "2001:db8::1", "port": 2222, "jumpProfileId": "j", "password": "not-copied" })).unwrap();
        assert_eq!(parsed, Target { id: "x".into(), host: "2001:db8::1".into(), port: 2222, jump_id: Some("j".into()) });
        assert!(Target::from_profile(&json!({ "id": "x", "port": 22 })).is_none());
    }

    #[tokio::test]
    async fn tcp_probe_measures_a_listener_and_reports_refusal() {
        for bind in ["127.0.0.1", "::1"] {
            let Ok(listener) = tokio::net::TcpListener::bind((bind, 0)).await else { continue };
            let port = listener.local_addr().unwrap().port();
            assert!(matches!(probe_tcp(bind, port, TCP_TIMEOUT).await, Outcome::Online { .. }), "{bind}");
            // Port 1: a freed ephemeral port could be taken by a test running in parallel.
            // Windows retries a refused connection for about 2 s, so there it ends as a timeout.
            let closed = probe_tcp(bind, 1, TCP_TIMEOUT).await;
            if cfg!(windows) { assert!(matches!(closed, Outcome::Failed { .. }), "{bind}: {closed:?}"); } else { assert_eq!(closed, Outcome::Failed { reason: "refused" }, "{bind}"); }
        }
        let listener = tokio::net::TcpListener::bind(("localhost", 0)).await.unwrap();
        let port = listener.local_addr().unwrap().port();
        assert!(matches!(probe_tcp("localhost", port, TCP_TIMEOUT).await, Outcome::Online { .. }), "DNS name, non-standard port");
        assert_eq!(probe_tcp("no-such-host.invalid", 22, TCP_TIMEOUT).await, Outcome::Failed { reason: "dns" });
    }

    struct TestHost {
        unlocked: AtomicBool,
        targets: Mutex<Vec<Target>>,
        events: Mutex<Vec<Value>>,
        probes: Mutex<u32>,
    }
    impl MonitorHost for TestHost {
        fn available(&self) -> bool { self.unlocked.load(Ordering::SeqCst) }
        fn settings(&self) -> (bool, Duration) { (true, Duration::from_millis(300)) }
        fn targets(&self) -> Option<Vec<Target>> { Some(self.targets.lock().clone()) }
        fn publish(&self, event: Value) { self.events.lock().push(event); }
        fn probe_jump(&self, _jump_id: &str, _target: &Target) -> BoxFuture<'static, Outcome> {
            *self.probes.lock() += 1;
            Box::pin(async { Outcome::Unverified { reason: "noJumpSession" } })
        }
    }

    #[tokio::test]
    async fn monitor_runs_all_servers_locks_clean_and_stops() {
        let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
        let open = listener.local_addr().unwrap().port();
        let host = Arc::new(TestHost {
            unlocked: AtomicBool::new(true),
            targets: Mutex::new(vec![
                Target { id: "up".into(), host: "127.0.0.1".into(), port: open, jump_id: None },
                Target { id: "down".into(), host: "127.0.0.1".into(), port: 1, jump_id: None },
                Target { id: "behind".into(), host: "10.255.255.1".into(), port: 22, jump_id: Some("up".into()) },
            ]),
            events: Mutex::new(Vec::new()),
            probes: Mutex::new(0),
        });
        let monitor = Arc::new(AvailabilityMonitor::default());
        let task = tokio::spawn(monitor.clone().run(host.clone()));
        let status = |id: &str| monitor.snapshot().iter().find(|e| e["id"] == id).map(|e| e["status"].as_str().unwrap().to_string());
        tokio::time::sleep(Duration::from_millis(600)).await;
        assert_eq!(status("up").as_deref(), Some("online"));
        assert_eq!(status("behind").as_deref(), Some("unverified"), "a jump-only server gets no direct result");
        tokio::time::timeout(Duration::from_secs(5), async { while status("down").as_deref() != Some("offline") { tokio::time::sleep(Duration::from_millis(50)).await; } }).await.unwrap();

        // Removal and a new port are picked up at once.
        host.targets.lock().retain(|t| t.id != "behind");
        host.targets.lock()[1].port = open;
        monitor.changed();
        tokio::time::timeout(Duration::from_secs(2), async { while status("down").as_deref() != Some("online") { tokio::time::sleep(Duration::from_millis(20)).await; } }).await.unwrap();
        assert_eq!(status("behind"), None);

        // Lock wipes everything; unlock checks at once.
        host.unlocked.store(false, Ordering::SeqCst);
        monitor.changed();
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(monitor.snapshot().is_empty());
        assert_eq!(host.events.lock().last().unwrap()["reset"], true);
        host.unlocked.store(true, Ordering::SeqCst);
        monitor.changed();
        tokio::time::timeout(Duration::from_secs(2), async { while status("up").as_deref() != Some("online") { tokio::time::sleep(Duration::from_millis(20)).await; } }).await.unwrap();

        monitor.stop();
        tokio::time::timeout(Duration::from_secs(1), task).await.expect("task ends on stop").unwrap();
        assert!(*host.probes.lock() >= 1);
    }
}