//! Packed sequential `SequentialReader` — Mode 1 read-side (ADR-002).
//!
//! Walks a buffer field-by-field according to the compiled
//! [`ReadPlan`], reading length prefixes to
//! determine variable-length data positions. Used at read time when the
//! consumer is parsing an incoming frame.
//!
//! The reader is sequential — it cannot jump to field N without reading
//! fields 0..N-1 first. This is inherent to packed layouts where
//! variable-length fields shift subsequent fields. [`SequentialReader`]
//! uses the [`crate::data_access`] read functions for all typed reads
//! and applies the plan's per-node endianness to every multi-byte value.
//!
//! ## Compiled form (ADR-011)
//!
//! Since ADR-011 the reader walks a pre-resolved read plan rather than
//! re-parsing the BAST typed tree per field: every `$ref` is resolved,
//! every endianness is baked, every union variant is inlined at
//! [`ReadPlan::compile`](crate::read_plan::ReadPlan::compile) time. The
//! reader holds an `Arc<ReadPlan>` (shared with the engine, refcount
//! bump per reader) plus per-reader cursor state only.

use crate::data_access;
use crate::error::AlkTypeError;
use crate::read_plan::{
    CompositePlan, DiscriminatorPlan, FieldPlan, ReadKind, ReadPlan,
};
use crate::schema::{AlkTypeKind, Endian, U32_SIZE};
use std::sync::Arc;
use serde_json::Value;

/// A value read from a field during sequential traversal.
///
/// Composite kinds ([`FieldValue::Struct`], [`FieldValue::Union`],
/// [`FieldValue::Array`]) return layout descriptors; the consumer
/// recurses with a fresh [`SequentialReader`] scoped to the
/// reported byte range.
///
/// **Known asymmetry**: `record` returns [`FieldValue::Bytes`] covering
/// the record's byte range, not a typed `Record { ... }` variant. The
/// consumer recurses into the record's value schema by walking the
/// borrowed slice. Every other composite kind returns a typed
/// descriptor; `Record` is the exception (review #002, N2). A future
/// revision may add a `FieldValue::Record` variant; for v0.1.0 the
/// `Bytes` form is stable.
#[derive(Debug, PartialEq)]
pub enum FieldValue<'a> {
    /// `int8`.
    I8(i8),
    /// `int16`.
    I16(i16),
    /// `int32`.
    I32(i32),
    /// `int64`.
    I64(i64),
    /// `uint8`.
    U8(u8),
    /// `uint16`.
    U16(u16),
    /// `uint32`.
    U32(u32),
    /// `uint64`.
    U64(u64),
    /// `float32`.
    F32(f32),
    /// `float64`.
    F64(f64),
    /// `bool`.
    Bool(bool),
    /// `enum` — `u32` index into the schema's `values` array.
    Enum(u32),
    /// `string` — borrows from the input buffer.
    String(&'a str),
    /// `bytes` — borrows from the input buffer.
    Bytes(&'a [u8]),
    /// `struct` — the consumer recurses with a new [`SequentialReader`]
    /// scoped to `start..end`.
    Struct {
        /// Inclusive start of the nested struct's byte range.
        start: usize,
        /// Exclusive end of the nested struct's byte range.
        end: usize,
    },
    /// `union` — the consumer looks up the variant schema using
    /// `discriminator` and recurses at `variant_start`.
    Union {
        /// Stringified discriminator value (mapping key).
        discriminator: String,
        /// Byte offset where the variant struct begins. Per
        /// discriminator kind:
        ///
        /// - **Byte-offset discriminator**: `union_start +
        ///   disc.offset + disc.size` — an absolute offset honoring the
        ///   schema's declared `disc.offset` displacement (which may be
        ///   nonzero, so the variant can start before or after a naive
        ///   shared-field walk would place it).
        /// - **Field-name discriminator**: the offset after the whole
        ///   `shared` walk (the union's declared `fields`: the
        ///   discriminator field + any shared fields). Under the
        ///   shared-then-variant wire convention (ADR-011 addendum,
        ///   review #006 H3) the variant's own fields begin exactly
        ///   here; the variant does not re-declare shared fields, so
        ///   walking the variant schema at this offset reads its own
        ///   fields only.
        variant_start: usize,
    },
    /// `array` — the consumer iterates `count` elements of
    /// stride `element_stride` starting at `element_start`.
    Array {
        /// Number of elements in the array.
        count: u32,
        /// Byte offset of the first element.
        element_start: usize,
        /// Byte distance between consecutive elements. `0` signals a
        /// variable-length element type — the consumer must walk each
        /// element sequentially.
        ///
        /// **Changed in 0.3.0 (deferred decision 4):** fixed-size struct
        /// (and nested-array) elements now report their true stride
        /// (e.g. `4` for `Point { x: u16, y: u16 }`); 0.2.0 reported `0`
        /// for those because only leaf primitives were stride-recognized.
        /// The `0` was behaviorally observable but semantically wrong —
        /// the consumer could index directly with the true stride.
        element_stride: usize,
    },
}

/// Walks a buffer field-by-field according to a compiled
/// [`ReadPlan`], reading length prefixes to determine variable-length
/// data positions. Used at read time when parsing incoming protocol
/// frames.
///
/// The reader is sequential — it cannot jump to field N without reading
/// fields 0..N-1 first. This is inherent to packed layouts where
/// variable-length fields shift subsequent fields.
///
/// Construct via [`crate::AlkTypeEngine::sequential_reader`] (shares
/// the engine's compiled plan), then drive with
/// [`SequentialReader::read_next`] until it returns `Ok(None)`. Use
/// [`SequentialReader::reset`] to walk the same buffer again, or
/// [`SequentialReader::read_field`] to seek a single field by name
/// (which walks all preceding fields to reach the target).
#[derive(Debug)]
pub struct SequentialReader {
    plan: Arc<ReadPlan>,
    field_index: usize,
    position: usize,
}

impl SequentialReader {
    /// Create a new `SequentialReader` from a compiled [`ReadPlan`].
    ///
    /// Infallible: the plan is already compiled (all `$ref`s resolved,
    /// all endianness baked) — construction just stores the `Arc` and
    /// zeroes the cursor. The fallible part of the old constructor moved
    /// to [`ReadPlan::compile`] (ADR-011).
    pub fn new(plan: Arc<ReadPlan>) -> Self {
        Self {
            plan,
            field_index: 0,
            position: 0,
        }
    }

    /// Read the next field from `buffer` at the current position.
    ///
    /// Returns `Ok(Some((field_name, value)))` and advances the internal
    /// position, or `Ok(None)` when all fields have been read. Variable-
    /// length fields (`string`/`bytes`) consume their 4-byte length prefix
    /// plus the data; composite fields advance past their computed byte
    /// range.
    ///
    /// # Errors
    ///
    /// Propagates [`AlkTypeError::Access`] from the underlying
    /// [`crate::data_access`] reads when `buffer` is too short or
    /// contains invalid data.
    pub fn read_next<'a>(
        &mut self,
        buffer: &'a [u8],
    ) -> Result<Option<(String, FieldValue<'a>)>, AlkTypeError> {
        if self.field_index >= self.plan.fields().len() {
            return Ok(None);
        }
        let name = self.plan.fields()[self.field_index].name().to_string();
        let (value, new_position) =
            self.read_field_at(buffer, self.field_index, self.position)?;
        self.position = new_position;
        self.field_index += 1;
        Ok(Some((name, value)))
    }

    /// Read a specific field by name. This walks through all preceding
    /// fields to reach the target (sequential access is inherent to
    /// packed layouts). Resets the reader first; the cursor is left at
    /// the position just past the target field.
    ///
    /// # Errors
    ///
    /// Returns [`AlkTypeError::Schema`] if `field_path` does not match
    /// any top-level field. Propagates [`AlkTypeError::Access`] for
    /// buffer-too-short or invalid data.
    pub fn read_field<'a>(
        &mut self,
        buffer: &'a [u8],
        field_path: &str,
    ) -> Result<FieldValue<'a>, AlkTypeError> {
        self.reset();
        let target_index = self
            .plan
            .field_index(field_path)
            .ok_or_else(|| AlkTypeError::Schema(format!("field not found in struct: {field_path}")))?;
        let mut position = 0usize;
        let mut target_value: Option<FieldValue<'a>> = None;
        for index in 0..=target_index {
            let (value, new_position) = self.read_field_at(buffer, index, position)?;
            position = new_position;
            if index == target_index {
                target_value = Some(value);
            }
        }
        self.position = position;
        self.field_index = target_index + 1;
        target_value.ok_or_else(|| {
            AlkTypeError::Schema(format!(
                "internal: target field {field_path} not produced by loop"
            ))
        })
    }

    /// Reset the reader to the beginning of the buffer (position 0,
    /// first field).
    pub fn reset(&mut self) {
        self.field_index = 0;
        self.position = 0;
    }

    /// The current byte position in the buffer.
    pub fn position(&self) -> usize {
        self.position
    }

    /// The endianness used by this reader (the plan's root container
    /// default).
    pub fn endian(&self) -> Endian {
        self.plan.endian()
    }

    /// The BAST document the plan was compiled from.
    pub fn schema(&self) -> &Value {
        self.plan.schema()
    }

    /// The compiled read plan this reader walks (shared via `Arc`).
    pub fn plan(&self) -> &Arc<ReadPlan> {
        &self.plan
    }

    fn read_field_at<'a>(
        &self,
        buffer: &'a [u8],
        index: usize,
        offset: usize,
    ) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
        let field = self
            .plan
            .fields()
            .get(index)
            .ok_or_else(|| AlkTypeError::Schema(format!("field index {index} out of range")))?;
        plan_read_field_at(field, buffer, offset, field.name())
    }
}

/// Read one planned field at `offset`: dispatch on the field's
/// [`ReadKind`], reading leaves directly and recursing into the
/// compiled composite body. `field_path` is for error attribution.
fn plan_read_field_at<'a>(
    field: &FieldPlan,
    buffer: &'a [u8],
    offset: usize,
    field_path: &str,
) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
    match field.kind() {
        ReadKind::Primitive(kind) => {
            plan_read_primitive(buffer, *kind, field.endian(), field_path, offset)
        }
        ReadKind::Enum => {
            let v = data_access::read_enum(buffer, offset, field_path, field.endian())?;
            Ok((FieldValue::Enum(v), offset + 4))
        }
        ReadKind::Struct => {
            let body = match field.body() {
                Some(CompositePlan::Struct(p)) => p,
                _ => {
                    return Err(AlkTypeError::Schema(format!(
                        "internal: struct field {field_path} has no Struct body"
                    )))
                }
            };
            let size = plan_walk_struct_size(body, buffer, offset, field_path)?;
            let end = checked_end(offset, size, field_path, "struct")?;
            Ok((FieldValue::Struct { start: offset, end }, end))
        }
        ReadKind::Union => {
            let body = field.body().ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: union field {field_path} has no body"
                ))
            })?;
            plan_read_union(body, buffer, offset, field_path, field.endian())
        }
        ReadKind::Array => {
            let body = field.body().ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: array field {field_path} has no body"
                ))
            })?;
            plan_read_array(body, buffer, offset, field_path, field.endian())
        }
        ReadKind::Record => {
            let body = field.body().ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: record field {field_path} has no body"
                ))
            })?;
            plan_read_record(body, buffer, offset, field_path, field.endian())
        }
    }
}
/// Read a primitive leaf value at `offset` and return the value plus
/// the position just past it. `endian` is the field's effective
/// endianness, baked into the plan at compile time.
fn plan_read_primitive<'a>(
    buffer: &'a [u8],
    kind: AlkTypeKind,
    endian: Endian,
    field_path: &str,
    offset: usize,
) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
    match kind {
        AlkTypeKind::Int8 => {
            let v = data_access::read_i8(buffer, offset, field_path)?;
            Ok((FieldValue::I8(v), offset + 1))
        }
        AlkTypeKind::Int16 => {
            let v = data_access::read_i16(buffer, offset, field_path, endian)?;
            Ok((FieldValue::I16(v), offset + 2))
        }
        AlkTypeKind::Int32 => {
            let v = data_access::read_i32(buffer, offset, field_path, endian)?;
            Ok((FieldValue::I32(v), offset + 4))
        }
        AlkTypeKind::Int64 => {
            let v = data_access::read_i64(buffer, offset, field_path, endian)?;
            Ok((FieldValue::I64(v), offset + 8))
        }
        AlkTypeKind::Uint8 => {
            let v = data_access::read_u8(buffer, offset, field_path)?;
            Ok((FieldValue::U8(v), offset + 1))
        }
        AlkTypeKind::Uint16 => {
            let v = data_access::read_u16(buffer, offset, field_path, endian)?;
            Ok((FieldValue::U16(v), offset + 2))
        }
        AlkTypeKind::Uint32 => {
            let v = data_access::read_u32(buffer, offset, field_path, endian)?;
            Ok((FieldValue::U32(v), offset + 4))
        }
        AlkTypeKind::Uint64 => {
            let v = data_access::read_u64(buffer, offset, field_path, endian)?;
            Ok((FieldValue::U64(v), offset + 8))
        }
        AlkTypeKind::Float32 => {
            let v = data_access::read_f32(buffer, offset, field_path, endian)?;
            Ok((FieldValue::F32(v), offset + 4))
        }
        AlkTypeKind::Float64 => {
            let v = data_access::read_f64(buffer, offset, field_path, endian)?;
            Ok((FieldValue::F64(v), offset + 8))
        }
        AlkTypeKind::Boolean => {
            let v = data_access::read_bool(buffer, offset, field_path)?;
            Ok((FieldValue::Bool(v), offset + 1))
        }
        AlkTypeKind::String => {
            let s = data_access::read_string(buffer, offset, field_path, endian)?;
            let total = U32_SIZE
                .checked_add(s.len())
                .ok_or_else(|| AlkTypeError::Access {
                    field_path: field_path.to_string(),
                    reason: "string size 4 + len overflows usize".to_string(),
                })?;
            let end = checked_end(offset, total, field_path, "string")?;
            Ok((FieldValue::String(s), end))
        }
        AlkTypeKind::Bytes => {
            let b = data_access::read_bytes(buffer, offset, field_path, endian)?;
            let total = U32_SIZE
                .checked_add(b.len())
                .ok_or_else(|| AlkTypeError::Access {
                    field_path: field_path.to_string(),
                    reason: "bytes size 4 + len overflows usize".to_string(),
                })?;
            let end = checked_end(offset, total, field_path, "bytes")?;
            Ok((FieldValue::Bytes(b), end))
        }
        other => Err(AlkTypeError::Schema(format!(
            "unsupported primitive kind {other} at {field_path}"
        ))),
    }
}

/// `offset + size`, checked — the shared end-position computation for
/// every composite walk.
fn checked_end(
    offset: usize,
    size: usize,
    field_path: &str,
    what: &str,
) -> Result<usize, AlkTypeError> {
    offset.checked_add(size).ok_or_else(|| AlkTypeError::Access {
        field_path: field_path.to_string(),
        reason: format!("{what} end {offset} + {size} overflows usize"),
    })
}

/// Walk a struct plan's fields sequentially from `offset`, returning
/// the total byte size. Only advances the cursor (reads length
/// prefixes); does not collect values.
fn plan_walk_struct_size(
    plan: &ReadPlan,
    buffer: &[u8],
    offset: usize,
    field_path: &str,
) -> Result<usize, AlkTypeError> {
    let mut position = offset;
    for field in plan.fields() {
        let sub_path = format!("{field_path}.{}", field.name());
        let (_, new_position) =
            plan_read_kind(field, buffer, position, &sub_path)?;
        if new_position < position {
            return Err(AlkTypeError::Access {
                field_path: sub_path,
                reason: format!("struct field walked backwards: {position} → {new_position}"),
            });
        }
        position = new_position;
    }
    Ok(position - offset)
}

/// Read a value for a plan node used as a TypeRef position (array
/// element, record value, union variant body) — same dispatch as
/// `plan_read_field_at` but over the composite `body` directly.
fn plan_read_kind<'a>(
    field: &FieldPlan,
    buffer: &'a [u8],
    offset: usize,
    field_path: &str,
) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
    match field.kind() {
        ReadKind::Primitive(kind) => {
            plan_read_primitive(buffer, *kind, field.endian(), field_path, offset)
        }
        ReadKind::Enum => {
            let v = data_access::read_enum(buffer, offset, field_path, field.endian())?;
            Ok((FieldValue::Enum(v), offset + 4))
        }
        ReadKind::Struct => {
            let body = match field.body() {
                Some(CompositePlan::Struct(p)) => p,
                _ => {
                    return Err(AlkTypeError::Schema(format!(
                        "internal: struct node {field_path} has no Struct body"
                    )))
                }
            };
            let size = plan_walk_struct_size(body, buffer, offset, field_path)?;
            let end = checked_end(offset, size, field_path, "struct")?;
            Ok((FieldValue::Struct { start: offset, end }, end))
        }
        ReadKind::Union => {
            let body = field.body().ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: union node {field_path} has no body"
                ))
            })?;
            plan_read_union(body, buffer, offset, field_path, field.endian())
        }
        ReadKind::Array => {
            let body = field.body().ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: array node {field_path} has no body"
                ))
            })?;
            plan_read_array(body, buffer, offset, field_path, field.endian())
        }
        ReadKind::Record => {
            let body = field.body().ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: record node {field_path} has no body"
                ))
            })?;
            plan_read_record(body, buffer, offset, field_path, field.endian())
        }
    }
}

/// Read a `union` composite body: read the discriminator, return the
/// variant start offset, and advance past the entire union payload.
///
/// Byte-offset discriminators read the integer at `disc.offset` and
/// start the variant after the discriminator's bytes (`shared: None`).
/// Field-name discriminators walk the union's `shared` sub-plan (the
/// declared `fields`: the discriminator field + any shared fields),
/// read the discriminator field at its index within `shared`, and start
/// the variant after the shared fields. Nested unions (a variant body
/// that is itself `CompositePlan::Union`) recurse naturally.
fn plan_read_union<'a>(
    body: &CompositePlan,
    buffer: &'a [u8],
    offset: usize,
    field_path: &str,
    endian: Endian,
) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
    let (disc, shared, variants) = match body {
        CompositePlan::Union {
            disc,
            shared,
            variants,
        } => (disc, shared, variants),
        _ => {
            return Err(AlkTypeError::Schema(format!(
                "internal: union body at {field_path} is not CompositePlan::Union"
            )))
        }
    };
    match disc {
        DiscriminatorPlan::Byte {
            offset: disc_offset,
            disc_type,
        } => {
            let abs_offset = checked_end(offset, *disc_offset, field_path, "discriminator")?;
            let (disc_value, disc_size) =
                plan_read_byte_discriminator(buffer, abs_offset, field_path, *disc_type, endian)?;
            let key = disc_value.to_string();
            let variant = variants
                .iter()
                .find(|(k, _)| *k == key)
                .map(|(_, v)| v)
                .ok_or_else(|| AlkTypeError::Access {
                    field_path: field_path.to_string(),
                    reason: format!("unknown union discriminator value: {key}"),
                })?;
            let variant_start = checked_end(abs_offset, disc_size, field_path, "variant")?;
            let variant_size = plan_walk_variant_size(variant, buffer, variant_start, field_path, endian)?;
            let end = checked_end(variant_start, variant_size, field_path, "union")?;
            Ok((
                FieldValue::Union {
                    discriminator: key,
                    variant_start,
                },
                end,
            ))
        }
        DiscriminatorPlan::Field { name, field_index } => {
            let shared_plan = shared.as_ref().ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: field-disc union at {field_path} has no shared plan"
                ))
            })?;
            let disc_field = shared_plan.fields().get(*field_index).ok_or_else(|| {
                AlkTypeError::Schema(format!(
                    "internal: union {field_path} discriminator index {field_index} out of range"
                ))
            })?;
            let disc_path = format!("{field_path}.{name}");
            let key =
                plan_discriminator_string_value(disc_field, buffer, offset, &disc_path)?;
            let variant = variants
                .iter()
                .find(|(k, _)| *k == key)
                .map(|(_, v)| v)
                .ok_or_else(|| AlkTypeError::Access {
                    field_path: field_path.to_string(),
                    reason: format!("unknown union discriminator value: {key}"),
                })?;
            // Walk the whole shared plan (discriminator field + any
            // shared fields); the variant starts after all of it.
            let mut position = offset;
            for field in shared_plan.fields() {
                let sub_path = format!("{field_path}.{}", field.name());
                let (_, new_position) = plan_read_kind(field, buffer, position, &sub_path)?;
                if new_position < position {
                    return Err(AlkTypeError::Access {
                        field_path: sub_path,
                        reason: format!(
                            "union shared field walked backwards: {position} → {new_position}"
                        ),
                    });
                }
                position = new_position;
            }
            let variant_size =
                plan_walk_variant_size(variant, buffer, position, field_path, endian)?;
            let end = checked_end(position, variant_size, field_path, "union")?;
            Ok((
                FieldValue::Union {
                    discriminator: key,
                    variant_start: position,
                },
                end,
            ))
        }
    }
}

/// Stringify a field-name discriminator value without re-reading: the
/// discriminator field's kind determines the buffer form. Mirrors the
/// 0.2.0 `discriminator_string_value` over the `FieldValue` the disc
/// read produced.
fn plan_discriminator_string_value(
    disc_field: &FieldPlan,
    buffer: &[u8],
    offset: usize,
    field_path: &str,
) -> Result<String, AlkTypeError> {
    match disc_field.kind() {
        ReadKind::Primitive(AlkTypeKind::String) => {
            let s = data_access::read_string(buffer, offset, field_path, disc_field.endian())?;
            Ok(s.to_string())
        }
        ReadKind::Primitive(AlkTypeKind::Uint8) => {
            Ok(data_access::read_u8(buffer, offset, field_path)?.to_string())
        }
        ReadKind::Primitive(AlkTypeKind::Uint16) => Ok(data_access::read_u16(
            buffer,
            offset,
            field_path,
            disc_field.endian(),
        )?
        .to_string()),
        ReadKind::Primitive(AlkTypeKind::Uint32) => Ok(data_access::read_u32(
            buffer,
            offset,
            field_path,
            disc_field.endian(),
        )?
        .to_string()),
        ReadKind::Enum => Ok(data_access::read_enum(
            buffer,
            offset,
            field_path,
            disc_field.endian(),
        )?
        .to_string()),
        other => Err(AlkTypeError::Schema(format!(
            "union {field_path} has unsupported field discriminator kind: {other:?}"
        ))),
    }
}

/// Read a byte-offset discriminator integer and return its value (as a
/// `u32`) plus its byte size.
fn plan_read_byte_discriminator(
    buffer: &[u8],
    offset: usize,
    field_path: &str,
    disc_type: AlkTypeKind,
    endian: Endian,
) -> Result<(u32, usize), AlkTypeError> {
    match disc_type {
        AlkTypeKind::Uint8 => {
            let v = data_access::read_u8(buffer, offset, field_path)?;
            Ok((u32::from(v), 1))
        }
        AlkTypeKind::Uint16 => {
            let v = data_access::read_u16(buffer, offset, field_path, endian)?;
            Ok((u32::from(v), 2))
        }
        AlkTypeKind::Uint32 => {
            let v = data_access::read_u32(buffer, offset, field_path, endian)?;
            Ok((v, 4))
        }
        other => Err(AlkTypeError::Schema(format!(
            "unsupported byte discriminator type: {other}"
        ))),
    }
}

/// Resolve a compiled variant body to its byte size starting at
/// `variant_start`. Struct variants walk their field list; union
/// variants (nested unions) read their discriminator and advance.
fn plan_walk_variant_size(
    variant: &CompositePlan,
    buffer: &[u8],
    variant_start: usize,
    field_path: &str,
    endian: Endian,
) -> Result<usize, AlkTypeError> {
    match variant {
        CompositePlan::Struct(p) => plan_walk_struct_size(p, buffer, variant_start, field_path),
        CompositePlan::Union { .. } => {
            let (_, end) =
                plan_read_union(variant, buffer, variant_start, field_path, endian)?;
            Ok(end - variant_start)
        }
        other => Err(AlkTypeError::Schema(format!(
            "internal: union variant at {field_path} compiled to a non-struct/union body: {other:?}"
        ))),
    }
}

/// Read an `array` composite body: fixed-size elements index by their
/// compiled stride; variable-length elements (stride `0`) are walked
/// sequentially.
fn plan_read_array<'a>(
    body: &CompositePlan,
    buffer: &'a [u8],
    offset: usize,
    field_path: &str,
    endian: Endian,
) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
    let (element, count, element_stride) = match body {
        CompositePlan::Array {
            element,
            count,
            element_stride,
        } => (element, *count, *element_stride),
        _ => {
            return Err(AlkTypeError::Schema(format!(
                "internal: array body at {field_path} is not CompositePlan::Array"
            )))
        }
    };
    let count_u32 = u32::try_from(count).map_err(|_| AlkTypeError::Access {
        field_path: field_path.to_string(),
        reason: format!("array count {count} overflows u32"),
    })?;

    let total = if element_stride == 0 {
        plan_walk_variable_array_size(element, buffer, offset, count, endian, field_path)?
    } else {
        count.checked_mul(element_stride).ok_or_else(|| AlkTypeError::Access {
            field_path: field_path.to_string(),
            reason: format!("array size {count} × stride {element_stride} overflows usize"),
        })?
    };

    let end = checked_end(offset, total, field_path, "array")?;

    Ok((
        FieldValue::Array {
            count: count_u32,
            element_start: offset,
            element_stride,
        },
        end,
    ))
}

/// Walk `count` variable-length array elements starting at `offset` and
/// return the total byte size of the element data.
fn plan_walk_variable_array_size(
    element: &CompositePlan,
    buffer: &[u8],
    start: usize,
    count: usize,
    endian: Endian,
    field_path: &str,
) -> Result<usize, AlkTypeError> {
    let mut position = start;
    for i in 0..count {
        let element_path = format!("{field_path}[{i}]");
        let (_, new_position) =
            plan_read_composite(element, buffer, position, &element_path, endian)?;
        if new_position < position {
            return Err(AlkTypeError::Access {
                field_path: element_path,
                reason: format!("array element walked backwards: {position} → {new_position}"),
            });
        }
        if new_position == position {
            return Err(AlkTypeError::Access {
                field_path: element_path,
                reason: format!(
                    "array element {i} consumed 0 bytes; a zero-size element makes the \
                     declared count unbounded on the wire"
                ),
            });
        }
        position = new_position;
    }
    Ok(position - start)
}

/// Read a composite body used as a TypeRef (array element, record
/// value). Structs walk their size; unions/arrays/records recurse.
fn plan_read_composite<'a>(
    body: &CompositePlan,
    buffer: &'a [u8],
    offset: usize,
    field_path: &str,
    endian: Endian,
) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
    match body {
        CompositePlan::Struct(plan) => {
            let size = plan_walk_struct_size(plan, buffer, offset, field_path)?;
            let end = checked_end(offset, size, field_path, "struct")?;
            Ok((FieldValue::Struct { start: offset, end }, end))
        }
        CompositePlan::Union { .. } => plan_read_union(body, buffer, offset, field_path, endian),
        CompositePlan::Array { .. } => plan_read_array(body, buffer, offset, field_path, endian),
        CompositePlan::Record { .. } => plan_read_record(body, buffer, offset, field_path, endian),
    }
}

/// Read a `record` composite body: `[count: u32]` followed by `count`
/// entries of `[key_len: u32][key_bytes][value]`. Returns the record's
/// byte range as `FieldValue::Bytes` (the one composite kind that does
/// not return a typed descriptor — see the `FieldValue` enum doc).
fn plan_read_record<'a>(
    body: &CompositePlan,
    buffer: &'a [u8],
    offset: usize,
    field_path: &str,
    endian: Endian,
) -> Result<(FieldValue<'a>, usize), AlkTypeError> {
    let value_body = match body {
        CompositePlan::Record { value } => value,
        _ => {
            return Err(AlkTypeError::Schema(format!(
                "internal: record body at {field_path} is not CompositePlan::Record"
            )))
        }
    };
    let count = data_access::read_u32(buffer, offset, field_path, endian)?;
    let count_usize = count as usize;
    let mut position = offset + U32_SIZE;
    for i in 0..count_usize {
        let key_path = format!("{field_path}[{i}].key");
        let key = data_access::read_string(buffer, position, &key_path, endian)?;
        let key_len = U32_SIZE
            .checked_add(key.len())
            .ok_or_else(|| AlkTypeError::Access {
                field_path: key_path.clone(),
                reason: "key size 4 + len overflows usize".to_string(),
            })?;
        position = checked_end(position, key_len, &key_path, "key")?;
        let value_path = format!("{field_path}[{i}].value");
        let (_, new_position) =
            plan_read_composite(value_body, buffer, position, &value_path, endian)?;
        position = new_position;
    }
    Ok((FieldValue::Bytes(&buffer[offset..position]), position))
}


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

    const LE: Endian = Endian::Little;
    const BE: Endian = Endian::Big;

    fn write_u32(buf: &mut [u8], offset: usize, value: u32, endian: Endian) {
        let bytes = match endian {
            Endian::Little => value.to_le_bytes(),
            Endian::Big => value.to_be_bytes(),
        };
        buf[offset..offset + 4].copy_from_slice(&bytes);
    }

    fn write_string(buf: &mut [u8], offset: usize, value: &str, endian: Endian) -> usize {
        let bytes = value.as_bytes();
        let total = 4 + bytes.len();
        write_u32(buf, offset, bytes.len() as u32, endian);
        buf[offset + 4..offset + 4 + bytes.len()].copy_from_slice(bytes);
        total
    }

    fn reader(root: &Value, name: &str) -> SequentialReader {
        let plan = ReadPlan::compile(root, name).expect("plan");
        SequentialReader::new(Arc::new(plan))
    }

    #[test]
    fn reads_fixed_size_fields_in_sequence() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "endian": "little",
                    "fields": [
                        { "name": "a", "kind": "uint8" },
                        { "name": "b", "kind": "uint32" },
                        { "name": "c", "kind": "uint16" }
                    ]
                }
            }
        });
        let mut buf = vec![0u8; 16];
        buf[0] = 42;
        write_u32(&mut buf, 1, 0x01020304, LE);
        buf[5..7].copy_from_slice(&1000u16.to_le_bytes());

        let mut reader = reader(&root, "S");
        assert_eq!(reader.position(), 0);

        let (name, value) = reader.read_next(&buf).unwrap().expect("field 0");
        assert_eq!(name, "a");
        assert_eq!(value, FieldValue::U8(42));
        assert_eq!(reader.position(), 1);

        let (name, value) = reader.read_next(&buf).unwrap().expect("field 1");
        assert_eq!(name, "b");
        assert_eq!(value, FieldValue::U32(0x01020304));
        assert_eq!(reader.position(), 5);

        let (name, value) = reader.read_next(&buf).unwrap().expect("field 2");
        assert_eq!(name, "c");
        assert_eq!(value, FieldValue::U16(1000));
        assert_eq!(reader.position(), 7);

        assert!(reader.read_next(&buf).unwrap().is_none());
    }

    #[test]
    fn reads_variable_length_string_with_length_prefix() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        { "name": "id",   "kind": "uint8" },
                        { "name": "name", "kind": "string" },
                        { "name": "tail", "kind": "uint8" }
                    ]
                }
            }
        });
        let mut buf = vec![0u8; 32];
        buf[0] = 7;
        let written = write_string(&mut buf, 1, "hello", LE);
        let after = 1 + written;
        buf[after] = 99;

        let mut reader = reader(&root, "S");
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "id");
        assert_eq!(value, FieldValue::U8(7));
        assert_eq!(reader.position(), 1);

        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "name");
        assert_eq!(value, FieldValue::String("hello"));
        assert_eq!(reader.position(), after);

        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "tail");
        assert_eq!(value, FieldValue::U8(99));
        assert_eq!(reader.position(), after + 1);

        assert!(reader.read_next(&buf).unwrap().is_none());
    }

    #[test]
    fn reads_bytes_field() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [ { "name": "blob", "kind": "bytes" } ]
                }
            }
        });
        let mut buf = vec![0u8; 16];
        let payload = [0xAA, 0xBB, 0xCC];
        write_u32(&mut buf, 0, 3, LE);
        buf[4..7].copy_from_slice(&payload);

        let mut reader = reader(&root, "S");
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "blob");
        assert_eq!(value, FieldValue::Bytes(&payload[..]));
        assert_eq!(reader.position(), 7);
    }

    #[test]
    fn respects_big_endian() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "endian": "big",
                    "fields": [ { "name": "id", "kind": "uint32" } ]
                }
            }
        });
        let mut buf = vec![0u8; 8];
        write_u32(&mut buf, 0, 0x01020304, BE);
        let mut reader = reader(&root, "S");
        let (_, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(value, FieldValue::U32(0x01020304));
        assert_eq!(reader.endian(), BE);
    }

    #[test]
    fn honors_field_level_endian_override() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "endian": "big",
                    "fields": [
                        { "name": "crc", "kind": "uint32", "endian": "little" }
                    ]
                }
            }
        });
        let buf = [0x04u8, 0x03, 0x02, 0x01];
        let mut reader = reader(&root, "S");
        let (_, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(value, FieldValue::U32(0x01020304));
    }

    #[test]
    fn reset_rewinds_cursor() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        { "name": "a", "kind": "uint8" },
                        { "name": "b", "kind": "uint8" }
                    ]
                }
            }
        });
        let buf = [10u8, 20u8];
        let mut reader = reader(&root, "S");
        let _ = reader.read_next(&buf).unwrap().unwrap();
        let _ = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(reader.position(), 2);
        reader.reset();
        assert_eq!(reader.position(), 0);
        assert_eq!(reader.field_index, 0);
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "a");
        assert_eq!(value, FieldValue::U8(10));
    }

    #[test]
    fn read_field_walks_preceding_fields() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        { "name": "a", "kind": "uint8" },
                        { "name": "b", "kind": "uint32" },
                        { "name": "c", "kind": "uint8" }
                    ]
                }
            }
        });
        let mut buf = vec![0u8; 16];
        buf[0] = 1;
        write_u32(&mut buf, 1, 0xDEADBEEF, LE);
        buf[5] = 9;

        let mut reader = reader(&root, "S");
        let value = reader.read_field(&buf, "c").unwrap();
        assert_eq!(value, FieldValue::U8(9));
        assert_eq!(reader.position(), 6);

        reader.reset();
        let value = reader.read_field(&buf, "b").unwrap();
        assert_eq!(value, FieldValue::U32(0xDEADBEEF));
    }

    #[test]
    fn read_field_unknown_returns_schema_error() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [ { "name": "a", "kind": "uint8" } ]
                }
            }
        });
        let buf = [0u8; 4];
        let mut reader = reader(&root, "S");
        let err = reader.read_field(&buf, "missing").unwrap_err();
        assert!(matches!(err, AlkTypeError::Schema(_)), "got {err:?}");
    }

    #[test]
    fn buffer_too_short_returns_access_error() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [ { "name": "id", "kind": "uint32" } ]
                }
            }
        });
        let buf = [0u8; 2];
        let mut reader = reader(&root, "S");
        let err = reader.read_next(&buf).unwrap_err();
        assert!(matches!(err, AlkTypeError::Access { .. }), "got {err:?}");
    }

    #[test]
    fn rejects_non_struct_top_level() {
        let root = json!({
            "$defs": {
                "U": { "kind": "uint32" }
            }
        });
        let err = ReadPlan::compile(&root, "U").unwrap_err();
        assert!(matches!(err, AlkTypeError::Schema(_)), "got {err:?}");
    }

    #[test]
    fn reads_all_fixed_size_kinds() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        { "name": "i8",  "kind": "int8" },
                        { "name": "i16", "kind": "int16" },
                        { "name": "i32", "kind": "int32" },
                        { "name": "i64", "kind": "int64" },
                        { "name": "u8",  "kind": "uint8" },
                        { "name": "u16", "kind": "uint16" },
                        { "name": "u32", "kind": "uint32" },
                        { "name": "u64", "kind": "uint64" },
                        { "name": "f32", "kind": "float32" },
                        { "name": "f64", "kind": "float64" },
                        { "name": "b",   "kind": "bool" },
                        { "name": "e",   "kind": { "$ref": "#/$defs/E" } }
                    ]
                },
                "E": { "kind": "enum", "values": ["A", "B"] }
            }
        });
        let mut buf = vec![0u8; 80];
        buf[0] = 0x80;
        buf[1..3].copy_from_slice(&(-1i16).to_le_bytes());
        buf[3..7].copy_from_slice(&(-5i32).to_le_bytes());
        buf[7..15].copy_from_slice(&(-9i64).to_le_bytes());
        buf[15] = 200;
        buf[16..18].copy_from_slice(&0xBEEFu16.to_le_bytes());
        buf[18..22].copy_from_slice(&0xDEADBEEFu32.to_le_bytes());
        buf[22..30].copy_from_slice(&0x0102030405060708u64.to_le_bytes());
        buf[30..34].copy_from_slice(&std::f32::consts::PI.to_le_bytes());
        buf[34..42].copy_from_slice(&std::f64::consts::PI.to_le_bytes());
        buf[42] = 0x01;
        buf[43..47].copy_from_slice(&7u32.to_le_bytes());

        let mut reader = reader(&root, "S");
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::I8(-128));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::I16(-1));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::I32(-5));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::I64(-9));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::U8(200));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::U16(0xBEEF));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::U32(0xDEADBEEF));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::U64(0x0102030405060708));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert!(matches!(v, FieldValue::F32(x) if (x - std::f32::consts::PI).abs() < 1e-6));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert!(matches!(v, FieldValue::F64(x) if (x - std::f64::consts::PI).abs() < 1e-12));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::Bool(true));
        let (_, v) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(v, FieldValue::Enum(7));
        assert!(reader.read_next(&buf).unwrap().is_none());
    }

    #[test]
    fn nested_struct_reports_byte_range() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        {
                            "name": "inner",
                            "kind": {
                                "kind": "struct",
                                "fields": [
                                    { "name": "x", "kind": "uint8" },
                                    { "name": "y", "kind": "uint16" }
                                ]
                            }
                        },
                        { "name": "tail", "kind": "uint8" }
                    ]
                }
            }
        });
        let mut buf = vec![0u8; 16];
        buf[0] = 1;
        buf[1..3].copy_from_slice(&0x0203u16.to_le_bytes());
        buf[3] = 9;

        let mut reader = reader(&root, "S");
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "inner");
        match value {
            FieldValue::Struct { start, end } => {
                assert_eq!(start, 0);
                assert_eq!(end, 3);
            }
            other => panic!("expected Struct, got {other:?}"),
        }
        assert_eq!(reader.position(), 3);

        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "tail");
        assert_eq!(value, FieldValue::U8(9));
        assert_eq!(reader.position(), 4);
    }

    #[test]
    fn byte_discriminator_union_reads_value() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        {
                            "name": "packet",
                            "kind": { "$ref": "#/$defs/Packet" }
                        }
                    ]
                },
                "Packet": {
                    "kind": "union",
                    "discriminator": { "kind": "byte", "offset": 0, "type": "uint8" },
                    "mapping": {
                        "5": { "$ref": "#/$defs/Read" }
                    }
                },
                "Read": {
                    "kind": "struct",
                    "fields": [ { "name": "x", "kind": "uint8" } ]
                }
            }
        });
        let mut buf = vec![0u8; 8];
        buf[0] = 5;
        buf[1] = 42;

        let mut reader = reader(&root, "S");
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "packet");
        match value {
            FieldValue::Union {
                discriminator,
                variant_start,
            } => {
                assert_eq!(discriminator, "5");
                assert_eq!(variant_start, 1);
            }
            other => panic!("expected Union, got {other:?}"),
        }
        assert_eq!(reader.position(), 2);
    }

    #[test]
    fn field_discriminator_union_reads_value() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        {
                            "name": "event",
                            "kind": { "$ref": "#/$defs/Event" }
                        }
                    ]
                },
                "Event": {
                    "kind": "union",
                    "discriminator": { "kind": "field", "name": "type" },
                    "fields": [ { "name": "type", "kind": "string" } ],
                    "mapping": {
                        "read": { "$ref": "#/$defs/Read" }
                    }
                },
                "Read": {
                    "kind": "struct",
                    "fields": [ { "name": "payload", "kind": "uint8" } ]
                }
            }
        });
        let mut buf = vec![0u8; 32];
        let written = write_string(&mut buf, 0, "read", LE);
        let after = written;
        buf[after] = 7;

        let mut reader = reader(&root, "S");
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "event");
        match value {
            FieldValue::Union {
                discriminator,
                variant_start,
            } => {
                assert_eq!(discriminator, "read");
                assert_eq!(variant_start, after);
            }
            other => panic!("expected Union, got {other:?}"),
        }
        assert_eq!(reader.position(), after + 1);
    }

    #[test]
    fn fixed_count_array_reads_count_inline() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        {
                            "name": "vals",
                            "kind": { "kind": "array", "element": "uint8", "count": 3 }
                        }
                    ]
                }
            }
        });
        let buf = [1u8, 2, 3];

        let mut reader = reader(&root, "S");
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "vals");
        match value {
            FieldValue::Array {
                count,
                element_start,
                element_stride,
            } => {
                assert_eq!(count, 3);
                assert_eq!(element_start, 0);
                assert_eq!(element_stride, 1);
            }
            other => panic!("expected Array, got {other:?}"),
        }
        assert_eq!(reader.position(), 3);
    }

    #[test]
    fn variable_length_element_array_walks_sequentially() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        {
                            "name": "items",
                            "kind": { "kind": "array", "element": "string", "count": 2 }
                        }
                    ]
                }
            }
        });
        let mut buf = vec![0u8; 64];
        let mut pos = 0;
        pos += write_string(&mut buf, pos, "ab", LE);
        pos += write_string(&mut buf, pos, "cdef", LE);

        let mut reader = reader(&root, "S");
        let (name, value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "items");
        match value {
            FieldValue::Array {
                count,
                element_start,
                element_stride,
            } => {
                assert_eq!(count, 2);
                assert_eq!(element_start, 0);
                assert_eq!(element_stride, 0);
            }
            other => panic!("expected Array, got {other:?}"),
        }
        assert_eq!(reader.position(), pos);
    }

    #[test]
    fn record_field_walks_entries() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        {
                            "name": "counts",
                            "kind": { "kind": "record", "values": "uint32" }
                        }
                    ]
                }
            }
        });
        let mut buf = vec![0u8; 64];
        write_u32(&mut buf, 0, 2, LE);
        let mut pos = 4;
        pos += write_string(&mut buf, pos, "a", LE);
        buf[pos..pos + 4].copy_from_slice(&1u32.to_le_bytes());
        pos += 4;
        pos += write_string(&mut buf, pos, "bb", LE);
        buf[pos..pos + 4].copy_from_slice(&2u32.to_le_bytes());
        pos += 4;

        let mut reader = reader(&root, "S");
        let (name, _value) = reader.read_next(&buf).unwrap().unwrap();
        assert_eq!(name, "counts");
        assert_eq!(reader.position(), pos);
    }

    #[test]
    fn union_unknown_discriminator_returns_access_error() {
        let root = json!({
            "$defs": {
                "S": {
                    "kind": "struct",
                    "fields": [
                        {
                            "name": "packet",
                            "kind": { "$ref": "#/$defs/Packet" }
                        }
                    ]
                },
                "Packet": {
                    "kind": "union",
                    "discriminator": { "kind": "byte", "offset": 0, "type": "uint8" },
                    "mapping": {
                        "5": { "$ref": "#/$defs/Read" }
                    }
                },
                "Read": {
                    "kind": "struct",
                    "fields": [ { "name": "x", "kind": "uint8" } ]
                }
            }
        });
        let buf = [99u8, 0];
        let mut reader = reader(&root, "S");
        let err = reader.read_next(&buf).unwrap_err();
        assert!(matches!(err, AlkTypeError::Access { .. }), "got {err:?}");
    }

    #[test]
    fn rejects_schema_with_missing_root() {
        let root = json!({ "$defs": { "Other": { "kind": "struct", "fields": [] } } });
        let err = ReadPlan::compile(&root, "Missing").unwrap_err();
        assert!(matches!(err, AlkTypeError::Schema(_)), "got {err:?}");
    }
}