//! Live state-feeds — `wai.feed.state` (Phase 3).
//!
//! A state-feed is a non-physics world: a scoreboard, a telemetry
//! stream, a market tape. There is no solver — the "sim" is the
//! identity step. What it standardizes is that the *state* converges:
//! the same signed op log replays to the identical state hash on every
//! sink, regardless of arrival order, exactly as an interactive world
//! does. This is the roadmap's "interactive-worlds with a trivial sim =
//! apply the typed field updates", and it inherits replay-equivalence,
//! canonical ordering, causality, sealing, and the receipt chain from
//! the worlds machinery unchanged.
//!
//! [`FeedState`] (`wai.world.sim.feed0`) is the reference registrant. A
//! node is a record: an ordered vector of self-describing typed cells
//! (`Int` i64, `Fixed` Q32.32, `Text` UTF-8). The cells are
//! self-describing so the canonical state bytes — and therefore the
//! state hash — need no external schema to parse; a declared schema is
//! an edge-level validation layer, not a floor-level dependency.
//!
//! Two resolution disciplines, both deterministic because every op
//! applies in the canonical total order (extension §5.2): `SetCell`
//! (LWW — last canonical writer wins) and `AddInt`/`AddFixed` (SUM —
//! commutative accumulation: a score increment, a packet counter, a
//! traded-volume tick).

use super::fixed::Fx;
use super::graph::WorldState;
use super::op::{mint_node_id, CausalOp};
use super::rng::Xoshiro256pp;
use super::sim::{Resolve, Sim};
use super::{SimContract, WorldError};

/// The reference state-feed sim capability.
pub const CAP_SIM_FEED0: &str = "wai.world.sim.feed0";

/// Mutation kinds registered by feed0. Field-update kinds carry a
/// `field_idx u16` prefix; an op targeting an absent node or an
/// out-of-range / wrong-typed cell is a deterministic no-op (an op set
/// must mean the same thing on every sink — extension §6.4).
pub mod feed_mutation {
    /// Payload = a record body (`n_cells u16 | cells…`). Target MUST
    /// equal `mint_node_id(actor, lamport)`. Resolution: LWW.
    pub const CREATE: u16 = 0x0001;
    /// Empty payload. Resolution: LWW.
    pub const DELETE: u16 = 0x0002;
    /// `field_idx u16 | cell(tag u8 | value)` — replace one cell.
    /// Resolution: LWW.
    pub const SET_CELL: u16 = 0x0010;
    /// `field_idx u16 | i64` — add to an `Int` cell (saturating).
    /// Resolution: SUM.
    pub const ADD_INT: u16 = 0x0011;
    /// `field_idx u16 | i64` (Fx bits) — add to a `Fixed` cell
    /// (saturating). Resolution: SUM.
    pub const ADD_FIXED: u16 = 0x0012;
}

/// One typed field value. Self-describing on the wire so the canonical
/// state bytes parse without a schema.
#[derive(Clone, Debug, PartialEq)]
pub enum Cell {
    /// Integer (scores, counts, volumes, enums-as-int, booleans-as-0/1).
    Int(i64),
    /// `wai.det.fixed64` Q32.32 (gauges, prices, percentages).
    Fixed(Fx),
    /// UTF-8 text (team names, ticker symbols, labels). Hashed as bytes.
    Text(Vec<u8>),
}

/// Cell type tags (the leading byte of a serialized cell).
const TAG_INT: u8 = 0x01;
const TAG_FIXED: u8 = 0x02;
const TAG_TEXT: u8 = 0x03;

impl Cell {
    fn write(&self, out: &mut Vec<u8>) {
        match self {
            Cell::Int(v) => {
                out.push(TAG_INT);
                out.extend_from_slice(&v.to_le_bytes());
            }
            Cell::Fixed(v) => {
                out.push(TAG_FIXED);
                out.extend_from_slice(&v.to_bits().to_le_bytes());
            }
            Cell::Text(s) => {
                out.push(TAG_TEXT);
                out.extend_from_slice(&(s.len() as u16).to_le_bytes());
                out.extend_from_slice(s);
            }
        }
    }
}

/// A feed record: an ordered vector of typed cells. Canonical field
/// order is index order, which is therefore the hash order.
#[derive(Clone, Debug, PartialEq, Default)]
pub struct FeedRecord {
    pub cells: Vec<Cell>,
}

impl FeedRecord {
    /// Record body: `n_cells u16 | cells…`.
    pub fn to_body(&self) -> Vec<u8> {
        let mut out = Vec::new();
        out.extend_from_slice(&(self.cells.len() as u16).to_le_bytes());
        for c in &self.cells {
            c.write(&mut out);
        }
        out
    }

    /// Parse a record body, consuming exactly `bytes` (strict — trailing
    /// bytes are a malformed mutation, not silently ignored).
    pub fn from_body(bytes: &[u8]) -> Result<FeedRecord, WorldError> {
        let mut c = Cur { b: bytes, p: 0 };
        let n = c.u16()? as usize;
        let mut cells = Vec::with_capacity(n);
        for _ in 0..n {
            cells.push(c.cell()?);
        }
        if c.p != bytes.len() {
            return Err(WorldError::BadMutation);
        }
        Ok(FeedRecord { cells })
    }
}

/// `wai.world.sim.feed0` — the identity-step state-feed registrant.
pub struct FeedState;

impl Sim for FeedState {
    type Node = FeedRecord;

    fn capability(&self) -> &'static str {
        CAP_SIM_FEED0
    }

    fn resolve_kind(&self, mutation_kind: u16) -> Option<Resolve> {
        use feed_mutation::*;
        match mutation_kind {
            CREATE | DELETE | SET_CELL => Some(Resolve::Lww),
            ADD_INT | ADD_FIXED => Some(Resolve::Sum),
            _ => None,
        }
    }

    fn apply(&self, world: &mut WorldState<FeedRecord>, op: &CausalOp) -> Result<(), WorldError> {
        use feed_mutation::*;
        match op.mutation_kind {
            CREATE => {
                if op.target != mint_node_id(&op.actor, op.lamport) {
                    return Err(WorldError::BadNodeMint);
                }
                let rec = FeedRecord::from_body(&op.mutation)?;
                world.nodes.insert(op.target, rec);
            }
            DELETE => {
                if !op.mutation.is_empty() {
                    return Err(WorldError::BadMutation);
                }
                world.nodes.remove(&op.target);
            }
            SET_CELL => {
                let mut c = Cur { b: &op.mutation, p: 0 };
                let idx = c.u16()? as usize;
                let cell = c.cell()?;
                if c.p != op.mutation.len() {
                    return Err(WorldError::BadMutation);
                }
                if let Some(rec) = world.nodes.get_mut(&op.target)
                    && let Some(slot) = rec.cells.get_mut(idx)
                {
                    *slot = cell;
                }
            }
            ADD_INT => {
                let (idx, v) = parse_idx_i64(&op.mutation)?;
                if let Some(rec) = world.nodes.get_mut(&op.target)
                    && let Some(Cell::Int(cur)) = rec.cells.get_mut(idx)
                {
                    *cur = cur.saturating_add(v);
                }
            }
            ADD_FIXED => {
                let (idx, bits) = parse_idx_i64(&op.mutation)?;
                if let Some(rec) = world.nodes.get_mut(&op.target)
                    && let Some(Cell::Fixed(cur)) = rec.cells.get_mut(idx)
                {
                    *cur = *cur + Fx::from_bits(bits); // saturating Fx add
                }
            }
            k => return Err(WorldError::UnknownMutationKind(k)),
        }
        Ok(())
    }

    /// A state-feed has no physics — the step is the identity. This is
    /// what makes it a *feed* and not a world; convergence comes wholly
    /// from canonical-ordered `apply`.
    fn step(
        &self,
        _world: &mut WorldState<FeedRecord>,
        _contract: &SimContract,
        _tick: u64,
        _rng: &mut Xoshiro256pp,
    ) {
    }

    fn serialize_node(&self, node: &FeedRecord, out: &mut Vec<u8>) {
        out.extend_from_slice(&node.to_body());
    }

    fn parse_node(&self, bytes: &[u8]) -> Result<FeedRecord, WorldError> {
        FeedRecord::from_body(bytes)
    }
}

/// Encode a `SetCell` mutation payload.
pub fn set_cell(field_idx: u16, cell: &Cell) -> Vec<u8> {
    let mut out = field_idx.to_le_bytes().to_vec();
    cell.write(&mut out);
    out
}

/// Encode an `AddInt` / `AddFixed` mutation payload (the i64 is the raw
/// addend for `AddInt`, or the Fx bit pattern for `AddFixed`).
pub fn add_payload(field_idx: u16, v: i64) -> Vec<u8> {
    let mut out = field_idx.to_le_bytes().to_vec();
    out.extend_from_slice(&v.to_le_bytes());
    out
}

fn parse_idx_i64(bytes: &[u8]) -> Result<(usize, i64), WorldError> {
    if bytes.len() != 10 {
        return Err(WorldError::BadMutation);
    }
    let idx = u16::from_le_bytes(bytes[0..2].try_into().unwrap()) as usize;
    let v = i64::from_le_bytes(bytes[2..10].try_into().unwrap());
    Ok((idx, v))
}

/// Minimal cursor over a mutation/record byte slice.
struct Cur<'a> {
    b: &'a [u8],
    p: usize,
}

impl<'a> Cur<'a> {
    fn take(&mut self, n: usize) -> Result<&'a [u8], WorldError> {
        let end = self.p.checked_add(n).ok_or(WorldError::Truncated)?;
        if end > self.b.len() {
            return Err(WorldError::Truncated);
        }
        let s = &self.b[self.p..end];
        self.p = end;
        Ok(s)
    }
    fn u16(&mut self) -> Result<u16, WorldError> {
        Ok(u16::from_le_bytes(self.take(2)?.try_into().unwrap()))
    }
    fn cell(&mut self) -> Result<Cell, WorldError> {
        let tag = self.take(1)?[0];
        match tag {
            TAG_INT => Ok(Cell::Int(i64::from_le_bytes(self.take(8)?.try_into().unwrap()))),
            TAG_FIXED => Ok(Cell::Fixed(Fx::from_bits(i64::from_le_bytes(
                self.take(8)?.try_into().unwrap(),
            )))),
            TAG_TEXT => {
                let len = u16::from_le_bytes(self.take(2)?.try_into().unwrap()) as usize;
                let s = self.take(len)?.to_vec();
                Ok(Cell::Text(s))
            }
            _ => Err(WorldError::BadMutation),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::world::container::WorldContainer;
    use crate::world::op::CausalOp;
    use crate::world::replay::{collect_checkpoints, LiveWorld};
    use crate::world::{SimContract, NUMERIC_FIXED64, RESOLVE_LWW};
    use ed25519_dalek::SigningKey;

    fn contract() -> SimContract {
        SimContract {
            tick_rate: 1,
            numeric: NUMERIC_FIXED64.into(),
            rng_alg: "xoshiro256++".into(),
            rng_seed: [0u8; 32],
            sim: CAP_SIM_FEED0.into(),
            resolve_default: RESOLVE_LWW.into(),
            checkpoint_every: 1,
        }
    }

    fn key(b: u8) -> SigningKey {
        SigningKey::from_bytes(&[b; 32])
    }

    fn record_body(cells: &[Cell]) -> Vec<u8> {
        FeedRecord { cells: cells.to_vec() }.to_body()
    }

    fn empty_feed(ops: Vec<CausalOp>) -> WorldContainer {
        let c = contract();
        WorldContainer {
            contract_bytes: c.to_canonical_json(),
            contract: c,
            graph_bytes: WorldState::<FeedRecord>::new().canonical_bytes(&FeedState),
            assets: vec![],
            ops,
            seal: None,
        }
    }

    #[test]
    fn record_body_round_trips() {
        let r = FeedRecord {
            cells: vec![Cell::Int(-7), Cell::Fixed(Fx::from_int(3)), Cell::Text(b"HOME".to_vec())],
        };
        assert_eq!(FeedRecord::from_body(&r.to_body()).unwrap(), r);
        // Trailing byte is rejected (strict parse).
        let mut bad = r.to_body();
        bad.push(0);
        assert!(FeedRecord::from_body(&bad).is_err());
    }

    // A scoreboard: one record { team:Text, score:Int }, score driven by
    // SUM increments. Replaying the sealed log on a fresh sink reproduces
    // the exact state hash — replay-equivalence for a non-physics feed.
    #[test]
    fn scoreboard_sum_replays_exactly() {
        let k = key(1);
        let lamport = 1u64;
        let node = mint_node_id(&k.verifying_key().to_bytes(), lamport);

        let create = CausalOp::sign(
            &k,
            lamport,
            1,
            vec![],
            node,
            feed_mutation::CREATE,
            record_body(&[Cell::Text(b"HOME".to_vec()), Cell::Int(0)]),
        );
        // three +2, +3, +2 increments to field 1 across ticks
        let inc = |lam: u64, tick: u64, by: i64| {
            CausalOp::sign(&k, lam, tick, vec![], node, feed_mutation::ADD_INT, add_payload(1, by))
        };
        let ops = vec![create, inc(2, 1, 2), inc(3, 2, 3), inc(4, 3, 2)];

        let container = empty_feed(ops.clone());
        let a = collect_checkpoints(&FeedState, &container, 3).unwrap();

        // Independent sink, ops fed in shuffled arrival order, converges.
        let mut live = LiveWorld::open(FeedState, empty_feed(vec![])).unwrap();
        for i in [3usize, 1, 0, 2] {
            live.ingest(ops[i].clone()).unwrap();
        }
        let h = live.advance_to(3).unwrap();
        assert_eq!(h, a.last().unwrap().hash);

        // Score really is 0+2+3+2 = 7.
        let rec = live.state().nodes.get(&node).unwrap();
        assert_eq!(rec.cells[1], Cell::Int(7));
        assert_eq!(rec.cells[0], Cell::Text(b"HOME".to_vec()));
    }

    // SET_CELL is LWW: the canonically-last writer at a tick wins,
    // regardless of arrival order.
    #[test]
    fn set_cell_is_last_writer_wins() {
        let k = key(2);
        let node = mint_node_id(&k.verifying_key().to_bytes(), 1);
        let create = CausalOp::sign(&k, 1, 1, vec![], node, feed_mutation::CREATE, record_body(&[Cell::Fixed(Fx::ZERO)]));
        // two sets at the same tick; higher lamport is canonically later.
        let lo = CausalOp::sign(&k, 2, 2, vec![], node, feed_mutation::SET_CELL, set_cell(0, &Cell::Fixed(Fx::from_int(1))));
        let hi = CausalOp::sign(&k, 3, 2, vec![], node, feed_mutation::SET_CELL, set_cell(0, &Cell::Fixed(Fx::from_int(9))));

        let mut live = LiveWorld::open(FeedState, empty_feed(vec![])).unwrap();
        for op in [&create, &hi, &lo] {
            live.ingest(op.clone()).unwrap();
        }
        live.advance_to(2).unwrap();
        assert_eq!(live.state().nodes.get(&node).unwrap().cells[0], Cell::Fixed(Fx::from_int(9)));
    }

    // A wrong-typed / out-of-range field update is a deterministic
    // no-op, not an error — the op set means the same thing everywhere.
    #[test]
    fn typed_update_to_wrong_cell_is_noop() {
        let k = key(3);
        let node = mint_node_id(&k.verifying_key().to_bytes(), 1);
        let create = CausalOp::sign(&k, 1, 1, vec![], node, feed_mutation::CREATE, record_body(&[Cell::Text(b"x".to_vec())]));
        // ADD_INT against a Text cell, and against an absent index.
        let bad_type = CausalOp::sign(&k, 2, 2, vec![], node, feed_mutation::ADD_INT, add_payload(0, 5));
        let bad_idx = CausalOp::sign(&k, 3, 2, vec![], node, feed_mutation::ADD_INT, add_payload(9, 5));

        let mut live = LiveWorld::open(FeedState, empty_feed(vec![])).unwrap();
        for op in [&create, &bad_type, &bad_idx] {
            live.ingest(op.clone()).unwrap();
        }
        live.advance_to(2).unwrap();
        assert_eq!(live.state().nodes.get(&node).unwrap().cells[0], Cell::Text(b"x".to_vec()));
    }
}
