//! aura's first real data source: the C3/C7 ingestion boundary.
//!
//! Transposes [`data_server`]'s Array-of-Structs `M1Parsed` records into aura's
//! Structure-of-Arrays base columns (C7) and normalizes data-server's
//! Unix-millisecond time to aura's canonical epoch-nanosecond [`Timestamp`] at
//! this one boundary (C3). A transposed [`M1Columns`] exposes its close column
//! as the engine source-stream shape (`Vec<(Timestamp, Scalar)>`) the SMA-cross
//! sample strategy consumes.
//!
//! Two ingestion shapes coexist: the **eager** [`load_m1_window`] → [`M1Columns`]
//! transpose (a bounded window materialized to owned SoA columns), and the **lazy
//! streaming** [`M1FieldSource`] — a `Source` that pulls one data-server
//! `Arc<[M1Parsed]>` chunk at a time and decodes each `Scalar` on demand
//! (resident O(one chunk), not O(window length)) for backtests over long
//! horizons. Both normalize ms→epoch-ns through the one [`unix_ms_to_epoch_ns`]
//! seam (C3).
//!
//! This crate is the **data-source ingestion edge**: it links `data-server`
//! (and its transitive `chrono`/`regex`/`zip`) and normalizes it into aura's
//! columns here. Dependencies follow the amended C16 per-case policy (INDEX.md),
//! not a blanket zero-dependency wall.

use aura_core::{Scalar, Timestamp};
use data_server::records::M1Parsed;
use data_server::{DataServer, SymbolChunkIter};
use std::sync::Arc;

/// Normalize data-server's Unix-millisecond time to aura's canonical epoch-ns
/// [`Timestamp`] — the single unit normalization of C3, performed at the one
/// ingestion boundary and nowhere else. The `i64` epoch-ns range covers all
/// market data through year ~2262, well beyond any real file.
pub fn unix_ms_to_epoch_ns(time_ms: i64) -> Timestamp {
    Timestamp(time_ms * 1_000_000)
}

/// One M1 window transposed Array-of-Structs → Structure-of-Arrays (C7): the
/// OHLCV bar as a bundle of base columns, time already normalized to epoch-ns.
/// `volume` is the one `i64` column; the price/spread columns are `f64`. All
/// columns share an index: `ts[i]` is the timestamp of `close[i]`, etc.
#[derive(Clone, Debug, PartialEq)]
pub struct M1Columns {
    pub ts: Vec<Timestamp>,
    pub open: Vec<f64>,
    pub high: Vec<f64>,
    pub low: Vec<f64>,
    pub close: Vec<f64>,
    pub spread: Vec<f64>,
    pub volume: Vec<i64>,
}

impl M1Columns {
    /// Pre-size all columns for `n` bars.
    fn with_capacity(n: usize) -> Self {
        Self {
            ts: Vec::with_capacity(n),
            open: Vec::with_capacity(n),
            high: Vec::with_capacity(n),
            low: Vec::with_capacity(n),
            close: Vec::with_capacity(n),
            spread: Vec::with_capacity(n),
            volume: Vec::with_capacity(n),
        }
    }

    /// Append one chunk of AoS bars, transposing into these SoA columns and
    /// normalizing time ms->epoch-ns at the boundary (C3). Reserves per chunk so
    /// repeated appends grow amortized; the push order is preserved (the
    /// chronological merge order C3 relies on). Lets `load_m1_window` transpose
    /// chunk-by-chunk without an intermediate AoS buffer.
    fn extend_from_bars(&mut self, bars: &[M1Parsed]) {
        let n = bars.len();
        self.ts.reserve(n);
        self.open.reserve(n);
        self.high.reserve(n);
        self.low.reserve(n);
        self.close.reserve(n);
        self.spread.reserve(n);
        self.volume.reserve(n);
        for b in bars {
            self.ts.push(unix_ms_to_epoch_ns(b.time_ms));
            self.open.push(b.open);
            self.high.push(b.high);
            self.low.push(b.low);
            self.close.push(b.close);
            self.spread.push(b.spread);
            self.volume.push(b.volume);
        }
    }

    /// The close column as an engine source stream: each normalized timestamp
    /// zipped with its close as a [`Scalar::f64`]. Ascending in timestamp iff
    /// the bars were (the C3 ingestion precondition `Harness::run` relies on).
    /// This is the price input the SMA-cross sample strategy consumes; a project
    /// that wants another field reads the public columns and zips its own.
    pub fn close_stream(&self) -> Vec<(Timestamp, Scalar)> {
        self.ts
            .iter()
            .zip(&self.close)
            .map(|(&t, &v)| (t, Scalar::f64(v)))
            .collect()
    }
}

/// Transpose data-server's AoS M1 records into aura's SoA columns (C7),
/// normalizing time at the boundary (C3). Pure: identical bars yield identical
/// columns (C1) — it reads no clock and no external state.
pub fn transpose_m1(bars: &[M1Parsed]) -> M1Columns {
    let mut c = M1Columns::with_capacity(bars.len());
    c.extend_from_bars(bars);
    c
}

/// Drain a data-server M1 window into transposed SoA columns. Reads the
/// chronological chunk iterator to exhaustion, transposing each chunk directly
/// into the SoA columns at the boundary (C3 — one merge point; no intermediate
/// AoS buffer, so peak memory is the columns alone, not columns + an AoS copy).
///
/// Each bound is an `Option<i64>` of inclusive Unix-ms (data-server's window
/// contract): `None` means unbounded on that side (data-server skips the
/// coarse `unix_ms_to_year_month` file filter rather than reaching for a
/// sentinel). `from_ms = None` reads from the start of history; `to_ms = None`
/// reads to the end. This threads data-server's own per-bound `Option`
/// contract through unchanged, so "to the end of history" is expressible
/// without an `i64::MAX` sentinel (which panics upstream in chrono's
/// `from_timestamp_millis`).
///
/// The returned `Option` is **file-level**, propagating data-server's own
/// `None`: it is `None` only when no archived file overlaps the window (e.g. a
/// far-future window, or an unknown symbol) — i.e. there is no data source to
/// read from. A window that *does* overlap a loaded file but happens to hold
/// zero bars (a narrow gap inside coverage) returns `Some(M1Columns)` with
/// empty columns, **not** `None`. So `Some`/`None` distinguishes "data source
/// present" from "nothing to read from", not "has bars" from "has none": a
/// consumer testing for "no bars in this window" must check
/// `cols.close.is_empty()`, not the `Option` alone.
pub fn load_m1_window(
    server: &Arc<DataServer>,
    symbol: &str,
    from_ms: Option<i64>,
    to_ms: Option<i64>,
) -> Option<M1Columns> {
    let mut it = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
    // Transpose each chunk straight into the SoA columns — no intermediate AoS
    // buffer (halves peak load memory, drops one full copy). The per-chunk push
    // order is the chronological merge order (C3).
    let mut cols = M1Columns::with_capacity(0);
    // M1Parsed is Copy; &Arc<[M1Parsed]> derefs to &[M1Parsed] in arg position.
    while let Some(chunk) = it.next_chunk() {
        cols.extend_from_bars(&chunk);
    }
    Some(cols)
}

/// Which base column of an M1 bar a source streams (C7: a composite window is a
/// bundle of base columns; one `Source` per consumed field).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum M1Field {
    Open,
    High,
    Low,
    Close,
    Spread,
    Volume,
}

/// Project one M1 bar into `(normalized ts, scalar)` for `field`. Pure: a
/// function of `(field, bar)` only — no iterator, no clock — so it is testable
/// in isolation. Time is normalized ms→epoch-ns at this one seam (C3).
fn decode(field: M1Field, bar: &M1Parsed) -> (Timestamp, Scalar) {
    let value = match field {
        M1Field::Open => Scalar::f64(bar.open),
        M1Field::High => Scalar::f64(bar.high),
        M1Field::Low => Scalar::f64(bar.low),
        M1Field::Close => Scalar::f64(bar.close),
        M1Field::Spread => Scalar::f64(bar.spread),
        M1Field::Volume => Scalar::i64(bar.volume),
    };
    (unix_ms_to_epoch_ns(bar.time_ms), value)
}

/// A streaming [`Source`](aura_engine::Source) over a data-server M1 window. Holds
/// at most one `Arc` chunk (a zero-copy clone of the cache's chunk) and a cursor;
/// constructs each `Scalar` per-pull; refills via `next_chunk()` when the chunk
/// drains. Resident footprint is O(one chunk), independent of window length.
pub struct M1FieldSource {
    iter: SymbolChunkIter<M1Parsed>,
    chunk: Option<Arc<[M1Parsed]>>,
    pos: usize,
    field: M1Field,
    head: Option<(Timestamp, Scalar)>,
    /// The requested window (inclusive Unix-ms), kept so `bounds()` can report a
    /// producer-supplied extent without streaming (#71). `None` = open-ended.
    from_ms: Option<i64>,
    to_ms: Option<i64>,
}

impl M1FieldSource {
    /// Open over `[from_ms, to_ms]` (inclusive Unix-ms, data-server's contract).
    /// `None` when no archived file overlaps the window (unknown symbol /
    /// far-future window) — propagating `stream_m1_windowed`'s file-level Option.
    /// A window that overlaps a file but holds zero matching bars yields a source
    /// whose first `peek` is `None` (immediately exhausted), not `None` here.
    pub fn open(
        server: &Arc<DataServer>,
        symbol: &str,
        from_ms: Option<i64>,
        to_ms: Option<i64>,
        field: M1Field,
    ) -> Option<Self> {
        let iter = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
        let mut s = Self { iter, chunk: None, pos: 0, field, head: None, from_ms, to_ms };
        s.advance();
        Some(s)
    }

    /// Decode the head at the cursor, refilling chunks as needed. Sets
    /// `self.head = None` at exhaustion.
    fn advance(&mut self) {
        loop {
            if self.chunk.is_none() {
                self.chunk = self.iter.next_chunk();
                self.pos = 0;
                if self.chunk.is_none() {
                    self.head = None;
                    return;
                }
            }
            let chunk = self.chunk.as_ref().unwrap();
            if self.pos < chunk.len() {
                self.head = Some(decode(self.field, &chunk[self.pos]));
                return;
            }
            self.chunk = None; // current chunk drained — loop to refill
        }
    }

    /// Records resident in this source right now: the current chunk's length (or
    /// 0 at exhaustion). The residency probe — bounded by one chunk length by
    /// construction (the source holds at most one `Arc` chunk and no accumulating
    /// field), so it can never grow with window length.
    pub fn resident_records(&self) -> usize {
        self.chunk.as_ref().map_or(0, |c| c.len())
    }
}

impl aura_engine::Source for M1FieldSource {
    fn peek(&self) -> Option<Timestamp> {
        self.head.map(|(t, _)| t)
    }
    fn next(&mut self) -> Option<(Timestamp, Scalar)> {
        let item = self.head?;
        self.pos += 1;
        self.advance();
        Some(item)
    }
    fn bounds(&self) -> Option<(Timestamp, Timestamp)> {
        // the producer-supplied window: the requested [from_ms, to_ms] normalized
        // to epoch-ns, known without streaming a bar (#71). Open-ended (either
        // bound absent) -> no known extent (Non-goals: archive-extent query).
        match (self.from_ms, self.to_ms) {
            (Some(from), Some(to)) => Some((unix_ms_to_epoch_ns(from), unix_ms_to_epoch_ns(to))),
            _ => None,
        }
    }
}

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

    #[test]
    fn unix_ms_to_epoch_ns_scales_ms_to_ns() {
        // data-server's own epoch fixture: 2017-03-01 00:00 UTC.
        assert_eq!(
            unix_ms_to_epoch_ns(1_488_326_400_000),
            Timestamp(1_488_326_400_000_000_000)
        );
        // epoch maps to epoch.
        assert_eq!(unix_ms_to_epoch_ns(0), Timestamp(0));
    }

    /// A hand-built M1 bar with distinct per-field values so a transpose test
    /// can tell the columns apart.
    fn full_bar(time_ms: i64) -> M1Parsed {
        M1Parsed {
            time_ms,
            open: 1.0,
            high: 2.0,
            low: 0.5,
            close: 1.5,
            spread: 0.1,
            volume: 100,
        }
    }

    #[test]
    fn transpose_m1_maps_every_field_to_its_column() {
        let bars = [full_bar(1_000), full_bar(2_000)];
        let c = transpose_m1(&bars);
        // ts normalized ms -> ns; each column equal length to the input.
        assert_eq!(c.ts, vec![Timestamp(1_000_000_000), Timestamp(2_000_000_000)]);
        assert_eq!(c.open, vec![1.0, 1.0]);
        assert_eq!(c.high, vec![2.0, 2.0]);
        assert_eq!(c.low, vec![0.5, 0.5]);
        assert_eq!(c.close, vec![1.5, 1.5]);
        assert_eq!(c.spread, vec![0.1, 0.1]);
        // volume is the one i64 column.
        assert_eq!(c.volume, vec![100_i64, 100_i64]);
    }

    #[test]
    fn transpose_m1_is_pure() {
        let bars = [full_bar(1_000), full_bar(2_000), full_bar(3_000)];
        assert_eq!(transpose_m1(&bars), transpose_m1(&bars));
    }

    #[test]
    fn transpose_m1_empty_input_yields_empty_columns() {
        let c = transpose_m1(&[]);
        assert!(c.ts.is_empty());
        assert!(c.close.is_empty());
        assert!(c.volume.is_empty());
    }

    #[test]
    fn chunked_accumulation_equals_single_transpose() {
        // `load_m1_window` now transposes chunk-by-chunk into the SoA columns
        // (no intermediate AoS buffer). That must be byte-identical to one
        // transpose over the concatenation — the path it replaced. This pins the
        // shared `extend_from_bars` logic without needing a live DataServer.
        let all = [full_bar(1_000), full_bar(2_000), full_bar(3_000)];
        let single = transpose_m1(&all);
        let mut chunked = M1Columns::with_capacity(0);
        chunked.extend_from_bars(&all[0..2]); // first chunk
        chunked.extend_from_bars(&all[2..3]); // second chunk
        assert_eq!(chunked, single);
    }

    #[test]
    fn close_stream_zips_ts_with_close_in_order() {
        // distinct close per bar so order is observable; ts ascending.
        let bars = [
            M1Parsed { time_ms: 1, open: 0.0, high: 0.0, low: 0.0, close: 10.0, spread: 0.0, volume: 0 },
            M1Parsed { time_ms: 2, open: 0.0, high: 0.0, low: 0.0, close: 20.0, spread: 0.0, volume: 0 },
            M1Parsed { time_ms: 3, open: 0.0, high: 0.0, low: 0.0, close: 30.0, spread: 0.0, volume: 0 },
        ];
        let stream = transpose_m1(&bars).close_stream();
        assert_eq!(
            stream,
            vec![
                (Timestamp(1_000_000), Scalar::f64(10.0)),
                (Timestamp(2_000_000), Scalar::f64(20.0)),
                (Timestamp(3_000_000), Scalar::f64(30.0)),
            ]
        );
        // ascending in timestamp (the merge precondition).
        assert!(stream.windows(2).all(|w| w[0].0 < w[1].0));
    }

    #[test]
    fn close_stream_empty_on_empty_columns() {
        assert!(transpose_m1(&[]).close_stream().is_empty());
    }

    #[test]
    fn decode_projects_each_field_to_its_scalar_kind() {
        // full_bar: open 1.0, high 2.0, low 0.5, close 1.5, spread 0.1, volume 100.
        let bar = full_bar(1_000);
        assert_eq!(decode(M1Field::Open, &bar),   (Timestamp(1_000_000_000), Scalar::f64(1.0)));
        assert_eq!(decode(M1Field::High, &bar),   (Timestamp(1_000_000_000), Scalar::f64(2.0)));
        assert_eq!(decode(M1Field::Low, &bar),    (Timestamp(1_000_000_000), Scalar::f64(0.5)));
        assert_eq!(decode(M1Field::Close, &bar),  (Timestamp(1_000_000_000), Scalar::f64(1.5)));
        assert_eq!(decode(M1Field::Spread, &bar), (Timestamp(1_000_000_000), Scalar::f64(0.1)));
        // volume is the one i64 column.
        assert_eq!(decode(M1Field::Volume, &bar), (Timestamp(1_000_000_000), Scalar::i64(100)));
    }
}
