//! Gated integration test for the Source ingestion seam: a real data-server M1
//! close stream driven LAZILY through the signal-quality sample harness via
//! `M1FieldSource`, plus the residency predicate (peak resident records ≤ one
//! chunk, independent of window length) and the N-field sharing property. Skips
//! where the local Pepperstone data directory is absent.

use std::sync::mpsc;
use std::sync::Arc;

use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, Source, SourceSpec,
    Target,
};
use aura_ingest::{M1Field, M1FieldSource};
use aura_backtest::SimBroker;
use aura_std::{Recorder, Sma, Sub};
use aura_strategy::Bias;
use data_server::loader::CHUNK_SIZE;
use data_server::{DataServer, DEFAULT_DATA_PATH};

const SYMBOL: &str = "AAPL.US";
// A bounded 2006-08 window in inclusive Unix-ms — enough bars to run the e2e
// backtest and the two-field sharing test end-to-end (it need not be
// multi-chunk; the residency test drives the unbounded archive instead).
const FROM_MS: i64 = 1_154_390_400_000;
const TO_MS: i64 = 1_157_068_799_999;

/// Bootstrap the cycle-0007 two-sink signal-quality harness and run it on a
/// single boxed source, folding the recorded equity + exposure into a RunReport.
fn run_sample(window: (Timestamp, Timestamp), source: Box<dyn Source>) -> RunReport {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_ex, rx_ex) = mpsc::channel();
    let f64_recorder_sig = || NodeSchema {
        inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
        output: vec![],
        params: vec![],
    };
    let mut h = Harness::bootstrap(FlatGraph {
        nodes: vec![
            Box::new(Sma::new(2)),
            Box::new(Sma::new(4)),
            Box::new(Sub::new()),
            Box::new(Bias::new(0.5)),
            Box::new(SimBroker::new(0.0001)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)),
        ],
        signatures: vec![
            Sma::builder().schema().clone(),
            Sma::builder().schema().clone(),
            Sub::builder().schema().clone(),
            Bias::builder().schema().clone(),
            SimBroker::builder(0.0001).schema().clone(),
            f64_recorder_sig(),
            f64_recorder_sig(),
        ],
        sources: vec![SourceSpec::raw(ScalarKind::F64, vec![
                Target { node: 0, slot: 0 },
                Target { node: 1, slot: 0 },
                Target { node: 4, slot: 1 },
            ])],
        edges: vec![
            Edge { from: 0, to: 2, slot: 0, from_field: 0 },
            Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            Edge { from: 3, to: 4, slot: 0, from_field: 0 },
            Edge { from: 4, to: 5, slot: 0, from_field: 0 },
            Edge { from: 3, to: 6, slot: 0, from_field: 0 },
        ],
        taps: Vec::new(),
    })
    .expect("valid signal-quality DAG");

    h.run(vec![source]);

    let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
    let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
    let equity = f64_field(&eq_rows, 0);
    let exposure = f64_field(&ex_rows, 0);
    RunReport {
        manifest: RunManifest {
            commit: "streaming-seam-test".to_string(),
            params: vec![
                ("sma_fast".to_string(), Scalar::i64(2)),
                ("sma_slow".to_string(), Scalar::i64(4)),
                ("bias_scale".to_string(), Scalar::f64(0.5)),
            ],
            defaults: vec![],
            window,
            seed: 0,
            broker: "sim-optimal(pip_size=0.0001)".to_string(),
            selection: None,
            instrument: None,
            topology_hash: None,
            project: None,
        },
        metrics: summarize(&equity, &exposure),
    }
}

fn skip_if_no_data(server: &Arc<DataServer>) -> bool {
    if !server.has_symbol(SYMBOL) {
        eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)");
        return true;
    }
    false
}

#[test]
fn streaming_close_source_backtests_end_to_end_deterministically() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if skip_if_no_data(&server) {
        return;
    }
    let open_close = || {
        M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close)
            .expect("AAPL.US has data in the 2006-08 window")
    };

    // window bounds for the manifest: drain a source once to read first/last ts.
    let mut probe = open_close();
    let first = probe.peek().expect("non-empty window");
    let mut last = first;
    while let Some((t, _)) = Source::next(&mut probe) {
        last = t;
    }
    let window = (first, last);

    let r1 = run_sample(window, Box::new(open_close()));
    assert!(r1.metrics.total_pips.is_finite(), "a backtest ran over the streamed window");

    // same window streamed again ⇒ bit-identical report (C1).
    let r2 = run_sample(window, Box::new(open_close()));
    assert_eq!(r1.to_json(), r2.to_json());
}

/// Measures the **aura source ring** residency — `M1FieldSource::resident_records()`,
/// the single chunk the source holds — NOT whole-process RSS. The ring is bounded by
/// one chunk at every window length (the assert below). The data-server `FileCache`
/// beneath retains the loaded window's parsed chunks for the pass (the replay-many
/// sharing model), so process RSS grows O(records-touched); that gap is by design,
/// not a leak, and is tracked as #95.
#[test]
fn residency_is_bounded_by_one_chunk_independent_of_window_length() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if skip_if_no_data(&server) {
        return;
    }
    // The whole archive (unbounded both ends), not the narrow e2e window: the
    // non-vacuity assertion below needs > CHUNK_SIZE bars, and early AAPL.US M1
    // coverage is sparse (a single month can hold far fewer). Driving the full
    // span is also the strongest form of the "residency independent of window
    // length" property — the longest window the local data offers.
    let mut src = M1FieldSource::open(&server, SYMBOL, None, None, M1Field::Close)
        .expect("AAPL.US has archived data");

    // #95 adjacent friction resolved: residency is probeable through the trait —
    // a `&dyn Source` reads it without downcasting to the concrete M1FieldSource.
    assert!(
        (&src as &dyn Source).resident_records().is_some(),
        "M1FieldSource residency must be probeable via &dyn Source (no downcast)",
    );

    // Drive the source the way `run` does (peek to pick, next to pop), sampling
    // resident records at every pull.
    let mut peak = 0usize;
    let mut total = 0usize;
    while src.peek().is_some() {
        peak = peak.max(src.resident_records().expect("M1FieldSource reports residency"));
        Source::next(&mut src);
        total += 1;
    }

    // multi-chunk window: more records than a single chunk holds, so the bound
    // below is non-vacuous (a O(window) source would have peaked at `total`).
    assert!(total > CHUNK_SIZE, "window must span multiple chunks (got {total} records)");
    // O(one chunk) source ring, NOT O(window): the per-pull ceiling never exceeds
    // one chunk, regardless of how many chunks the window spans (whole-process RSS
    // is a separate quantity — see this test's doc comment and #95).
    assert!(peak <= CHUNK_SIZE, "resident records {peak} exceeded one chunk ({CHUNK_SIZE})");
    assert!(peak > 0, "a non-empty window resided at least one record");
}

#[test]
fn two_field_sources_share_one_window() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if skip_if_no_data(&server) {
        return;
    }
    // close + volume over the same window: each is an independent Source pulling
    // its field; both stream to exhaustion (the N-sources-share-the-ts-axis shape).
    let mut close = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close)
        .expect("close source");
    let mut volume = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Volume)
        .expect("volume source");

    let mut n_close = 0usize;
    while let Some((_, v)) = Source::next(&mut close) {
        assert_eq!(v.kind(), ScalarKind::F64, "close is an f64 column");
        n_close += 1;
    }
    let mut n_volume = 0usize;
    while let Some((_, v)) = Source::next(&mut volume) {
        assert_eq!(v.kind(), ScalarKind::I64, "volume is an i64 column");
        n_volume += 1;
    }
    assert_eq!(n_close, n_volume, "both fields share the same ts axis ⇒ same count");
    assert!(n_close > 0);
}

#[test]
fn field_source_bounds_are_the_requested_window_normalized() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if skip_if_no_data(&server) {
        return;
    }
    let src = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close)
        .expect("AAPL.US has data in the 2006-08 window");
    // bounds report the *requested* window normalized to epoch-ns, without
    // streaming a single bar (producer-supplied window; #71 firewall).
    assert_eq!(
        Source::bounds(&src),
        Some((
            aura_ingest::unix_ms_to_epoch_ns(FROM_MS),
            aura_ingest::unix_ms_to_epoch_ns(TO_MS),
        ))
    );
    // an open-ended source (no window) has no known extent (Non-goals: archive
    // extent query is deferred).
    let open_ended = M1FieldSource::open(&server, SYMBOL, None, None, M1Field::Close)
        .expect("AAPL.US archive exists");
    assert_eq!(Source::bounds(&open_ended), None);
}