//! Hardware ingest: frames written by the out-of-process collector are read by
//! the same `wire::read_frame` the socket uses and published into the hub, so
//! a subscriber's snapshot carries them like any other collector's signals.

use std::io::Cursor;

use signal_schema::{wire, Signal, SignalName, Source, Tag, Value, SCHEMA_VERSION};
use signald::collectors::hardware;
use signald::hub::Hub;

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

/// The five signals `macos-collector` emits per tick, as it emits them.
fn one_tick() -> Vec<Signal> {
    vec![
        sig(Source::Hardware, SignalName::CpuLoad, 0.25),
        sig(Source::Macos, SignalName::BatteryPct, 80.0),
        sig(Source::Macos, SignalName::Charging, 1.0),
        sig(Source::Hardware, SignalName::BatteryDrawW, 12.5),
        sig(Source::Macos, SignalName::ThermalState, 0.0),
    ]
}

fn frames(signals: &[Signal]) -> Vec<u8> {
    let mut buf = Vec::new();
    for s in signals {
        wire::write_frame(&mut buf, s).unwrap();
    }
    buf
}

#[test]
fn canned_frames_reach_a_subscriber_snapshot() {
    let hub = Hub::new();
    let mut first = one_tick();
    let mut second = one_tick();
    second[0].value = Value(0.75); // cpu_load changes on the second tick
    first.append(&mut second);

    let n = hardware::ingest(&mut Cursor::new(frames(&first)), &hub).unwrap();
    assert_eq!(n, 10, "every frame is published");

    let (snapshot, _rx) = hub.subscribe();
    assert_eq!(snapshot.len(), 5, "one cached value per metric");
    let cpu = snapshot.iter().find(|s| s.name == SignalName::CpuLoad).unwrap();
    assert_eq!(cpu.value, Value(0.75), "keep-latest");
    assert_eq!(cpu.source, Source::Hardware);
    for name in [
        SignalName::BatteryPct,
        SignalName::Charging,
        SignalName::BatteryDrawW,
        SignalName::ThermalState,
    ] {
        assert!(snapshot.iter().any(|s| s.name == name), "{name:?} missing");
    }
}

#[test]
fn frame_that_breaks_the_tag_rule_is_dropped_and_ingest_continues() {
    let hub = Hub::new();
    let good = sig(Source::Hardware, SignalName::CpuLoad, 0.5);
    // CpuLoad never allows a tag. Encode one anyway, bypassing the schema's
    // well-formedness check, as a misbehaving collector could.
    let mut bad = good.clone();
    bad.tag = Some(Tag::repo_path("/not/allowed").unwrap());
    let after = sig(Source::Macos, SignalName::BatteryPct, 42.0);

    let mut buf = frames(&[good]);
    buf.extend(wire::encode(&bad));
    buf.extend(frames(&[after]));

    let n = hardware::ingest(&mut Cursor::new(buf), &hub).unwrap();
    assert_eq!(n, 2, "the tagged frame is dropped, the other two published");

    let snapshot = hub.snapshot();
    assert_eq!(snapshot.len(), 2);
    assert!(
        snapshot.iter().all(|s| s.tag.is_none()),
        "no tagged frame reaches the hub"
    );
    assert!(
        snapshot.iter().any(|s| s.name == SignalName::BatteryPct),
        "ingest kept reading past the rejected frame"
    );
}