//! `AnyColumn` — the type-erased edge: the four concrete `Column<T>` behind one
//! enum. The kind check is paid at the edge (wiring time); the hot loop grabs
//! the concrete column once (`as_*_mut`) and pushes monomorphically (C7).

use crate::column::Column;
use crate::error::KindMismatch;
use crate::scalar::{Scalar, ScalarKind, Timestamp};
use crate::cell::Cell;

/// A type-erased edge over the four scalar columns (C7).
pub enum AnyColumn {
    I64(Column<i64>),
    F64(Column<f64>),
    Bool(Column<bool>),
    Ts(Column<Timestamp>),
}

impl AnyColumn {
    /// Create a type-erased column of `kind`, sized to `lookback` (C8).
    pub fn with_capacity(kind: ScalarKind, lookback: usize) -> Self {
        match kind {
            ScalarKind::I64 => AnyColumn::I64(Column::with_capacity(lookback)),
            ScalarKind::F64 => AnyColumn::F64(Column::with_capacity(lookback)),
            ScalarKind::Bool => AnyColumn::Bool(Column::with_capacity(lookback)),
            ScalarKind::Timestamp => AnyColumn::Ts(Column::with_capacity(lookback)),
        }
    }

    pub fn kind(&self) -> ScalarKind {
        match self {
            AnyColumn::I64(_) => ScalarKind::I64,
            AnyColumn::F64(_) => ScalarKind::F64,
            AnyColumn::Bool(_) => ScalarKind::Bool,
            AnyColumn::Ts(_) => ScalarKind::Timestamp,
        }
    }

    pub fn len(&self) -> usize {
        match self {
            AnyColumn::I64(c) => c.len(),
            AnyColumn::F64(c) => c.len(),
            AnyColumn::Bool(c) => c.len(),
            AnyColumn::Ts(c) => c.len(),
        }
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    pub fn run_count(&self) -> u64 {
        match self {
            AnyColumn::I64(c) => c.run_count(),
            AnyColumn::F64(c) => c.run_count(),
            AnyColumn::Bool(c) => c.run_count(),
            AnyColumn::Ts(c) => c.run_count(),
        }
    }

    /// Edge-time typed push. `Err(KindMismatch)` if `v`'s kind != this column's
    /// kind; the column is left untouched (C7 guard). The hot path bypasses this
    /// by taking the concrete `Column<T>` once via `as_*_mut`.
    pub fn push(&mut self, v: Scalar) -> Result<(), KindMismatch> {
        match (self, v.kind()) {
            (AnyColumn::I64(c), ScalarKind::I64) => {
                c.push(v.as_i64());
                Ok(())
            }
            (AnyColumn::F64(c), ScalarKind::F64) => {
                c.push(v.as_f64());
                Ok(())
            }
            (AnyColumn::Bool(c), ScalarKind::Bool) => {
                c.push(v.as_bool());
                Ok(())
            }
            (AnyColumn::Ts(c), ScalarKind::Timestamp) => {
                c.push(v.as_ts());
                Ok(())
            }
            (this, got) => Err(KindMismatch { expected: this.kind(), got }),
        }
    }

    /// Branch-free, infallible hot-path push of a bare [`Cell`] into this column's
    /// concrete `Column<T>`, reinterpreting the cell by *this* column's kind. The
    /// inter-node forward uses this: the edge's `from_field`→slot kind match is
    /// verified once at bootstrap (C7), so no per-value kind check is needed here
    /// — unlike [`push`](Self::push), which keeps the kind guard for the
    /// ingestion boundary.
    pub fn push_cell(&mut self, c: Cell) {
        match self {
            AnyColumn::I64(col) => col.push(c.i64()),
            AnyColumn::F64(col) => col.push(c.f64()),
            AnyColumn::Bool(col) => col.push(c.bool()),
            AnyColumn::Ts(col) => col.push(c.ts()),
        }
    }

    /// Read one type-erased value (0 = newest). `None` if cold / out of range.
    pub fn get(&self, k: usize) -> Option<Scalar> {
        match self {
            AnyColumn::I64(c) => c.get(k).map(Scalar::i64),
            AnyColumn::F64(c) => c.get(k).map(Scalar::f64),
            AnyColumn::Bool(c) => c.get(k).map(Scalar::bool),
            AnyColumn::Ts(c) => c.get(k).map(Scalar::ts),
        }
    }

    /// Concrete-column accessor for the monomorphic hot path; `Some` only for
    /// the matching kind.
    pub fn as_i64_mut(&mut self) -> Option<&mut Column<i64>> {
        match self {
            AnyColumn::I64(c) => Some(c),
            _ => None,
        }
    }

    pub fn as_f64_mut(&mut self) -> Option<&mut Column<f64>> {
        match self {
            AnyColumn::F64(c) => Some(c),
            _ => None,
        }
    }

    pub fn as_bool_mut(&mut self) -> Option<&mut Column<bool>> {
        match self {
            AnyColumn::Bool(c) => Some(c),
            _ => None,
        }
    }

    pub fn as_ts_mut(&mut self) -> Option<&mut Column<Timestamp>> {
        match self {
            AnyColumn::Ts(c) => Some(c),
            _ => None,
        }
    }

    /// Read-side concrete-column accessor (the symmetric partner of
    /// `as_*_mut`); `Some` only for the matching kind. `Ctx` uses these to hand
    /// a typed window from a type-erased edge.
    pub fn as_i64(&self) -> Option<&Column<i64>> {
        match self {
            AnyColumn::I64(c) => Some(c),
            _ => None,
        }
    }

    pub fn as_f64(&self) -> Option<&Column<f64>> {
        match self {
            AnyColumn::F64(c) => Some(c),
            _ => None,
        }
    }

    pub fn as_bool(&self) -> Option<&Column<bool>> {
        match self {
            AnyColumn::Bool(c) => Some(c),
            _ => None,
        }
    }

    pub fn as_ts(&self) -> Option<&Column<Timestamp>> {
        match self {
            AnyColumn::Ts(c) => Some(c),
            _ => None,
        }
    }
}

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

    #[test]
    fn same_kind_push_round_trips() {
        let mut e = AnyColumn::with_capacity(ScalarKind::F64, 4);
        e.push(Scalar::f64(1.5)).unwrap();
        e.push(Scalar::f64(2.5)).unwrap();
        assert_eq!(e.kind(), ScalarKind::F64);
        assert_eq!(e.len(), 2);
        assert_eq!(e.run_count(), 2);
        assert_eq!(e.get(0), Some(Scalar::f64(2.5))); // newest
        assert_eq!(e.get(1), Some(Scalar::f64(1.5)));
        assert_eq!(e.get(2), None);
    }

    #[test]
    fn push_cell_round_trips_by_column_kind() {
        // push_cell reinterprets the bare cell by the column's own kind; reading
        // back through `get` reconstructs the self-describing Scalar.
        let mut f = AnyColumn::with_capacity(ScalarKind::F64, 2);
        f.push_cell(Cell::from_f64(1.5));
        assert_eq!(f.get(0), Some(Scalar::f64(1.5)));
        assert_eq!(f.run_count(), 1);

        let mut i = AnyColumn::with_capacity(ScalarKind::I64, 2);
        i.push_cell(Cell::from_i64(42));
        assert_eq!(i.get(0), Some(Scalar::i64(42)));

        let mut b = AnyColumn::with_capacity(ScalarKind::Bool, 2);
        b.push_cell(Cell::from_bool(true));
        assert_eq!(b.get(0), Some(Scalar::bool(true)));

        let mut t = AnyColumn::with_capacity(ScalarKind::Timestamp, 2);
        t.push_cell(Cell::from_ts(Timestamp(7)));
        assert_eq!(t.get(0), Some(Scalar::ts(Timestamp(7))));
    }

    #[test]
    fn wrong_kind_push_is_rejected() {
        let mut edge = AnyColumn::with_capacity(ScalarKind::I64, 8);
        let err = edge.push(Scalar::f64(1.5)).unwrap_err(); // f64 into an i64 edge
        assert_eq!(err, KindMismatch { expected: ScalarKind::I64, got: ScalarKind::F64 });
        assert_eq!(edge.len(), 0); // nothing was stored
        assert_eq!(edge.run_count(), 0); // and the counter did not move
    }

    #[test]
    fn concrete_accessor_matches_only_its_kind() {
        let mut e = AnyColumn::with_capacity(ScalarKind::F64, 2);
        assert!(e.as_f64_mut().is_some());
        assert!(e.as_i64_mut().is_none());
        assert!(e.as_bool_mut().is_none());
        assert!(e.as_ts_mut().is_none());
        // hot-path push through the concrete column, monomorphic:
        e.as_f64_mut().unwrap().push(3.0);
        assert_eq!(e.get(0), Some(Scalar::f64(3.0)));
        assert_eq!(e.run_count(), 1);
    }

    #[test]
    fn read_accessor_matches_only_its_kind() {
        let f = AnyColumn::with_capacity(ScalarKind::F64, 2);
        assert!(f.as_f64().is_some());
        assert!(f.as_i64().is_none());
        assert!(f.as_bool().is_none());
        assert!(f.as_ts().is_none());

        let i = AnyColumn::with_capacity(ScalarKind::I64, 2);
        assert!(i.as_i64().is_some());
        assert!(i.as_f64().is_none());
    }

    #[test]
    fn timestamp_edge_round_trips() {
        let mut e = AnyColumn::with_capacity(ScalarKind::Timestamp, 2);
        e.push(Scalar::ts(Timestamp(1_700_000_000_000_000_000))).unwrap();
        assert_eq!(e.get(0), Some(Scalar::ts(Timestamp(1_700_000_000_000_000_000))));
    }
}