obj-core 1.1.0

//! Field extraction: `Document` → encoded index keys.
//!
//! [`extract_index_keys`] is the bridge between a user `Document`
//! and the per-index B-trees of #57's catalog layer. For each
//! declared index it walks the document's field path(s) and hands
//! the resolved [`Dynamic`] values to `encode_index_key` (#55).
//!
//! # postcard is not self-describing
//!
//! The M5 [`Dynamic`] type ships a `from_postcard_bytes` decoder
//! that ONLY accepts the tagged-Dynamic wire format — NOT raw
//! native-postcard payloads. M7 field extraction cannot use that
//! decoder for user documents (which are always native postcard).
//!
//! The workaround is a dedicated serde-driven reflection: a
//! `DynamicSerializer` walks the document's `serde::Serialize`
//! impl and emits a `Dynamic` tree with `Dynamic::Map` for every
//! struct and `Dynamic::Seq` for every sequence. The result is
//! the same shape `Dynamic::get` is built for, so the top-level
//! field-path walk in `extract_index_keys` is one `Dynamic::get`
//! call per `IndexSpec::key_paths` entry.
//!
//! # Path limitations
//!
//! M7 supports **top-level field paths only** — a single field name
//! within the document's struct. Dotted paths (`"address.city"`),
//! array indexing (`"tags[0]"`), and `JSONPath` syntax are out of
//! scope. The `IndexSpec::key_paths` vector is a list of top-level
//! field names; each entry is one `Dynamic::get` lookup. The
//! limitation is documented in `docs/format.md` § Indexes.
//!
//! # Power-of-ten posture
//!
//! - **Rule 1.** No recursion in the extractor itself — the
//!   serializer is the only place where the call graph naturally
//!   reflects the document structure, and it carries an explicit
//!   `MAX_REFLECT_DEPTH` bound that mirrors `MAX_DYNAMIC_DEPTH`.
//! - **Rule 2.** Per-key-path iteration is bounded by the spec's
//!   `key_paths.len()`. `Each`-kind extraction is bounded by
//!   [`MAX_EACH_ENTRIES`] (16 384) — beyond which extraction errors
//!   with [`crate::Error::EachIndexTooLarge`] (added later in #59;
//!   M7 currently uses an `Error::InvalidArgument` placeholder
//!   pending that variant's introduction).
//! - **Rule 4.** `extract_index_keys` is short — per-kind helpers
//!   carry the heavy lifting (one helper per kind).
//! - **Rule 7.** No `unwrap` / `expect` on the production path.
//!   Missing field / wrong type surface as the specific
//!   `Error::IndexField*` variants.

#![forbid(unsafe_code)]

use std::collections::BTreeMap;

use serde::{ser, Serialize};

use crate::codec::{Document, Dynamic};
use crate::error::{Error, Result};
use crate::index::key::{encode_field, encode_index_key, EncodedIndexKey};
use crate::index::spec::{IndexKind, IndexSpec};

/// Maximum depth of the document-reflection walk. Mirrors the
/// `MAX_DYNAMIC_DEPTH` bound in [`crate::codec::dynamic`] —
/// defensive against pathological nested-struct inputs that would
/// otherwise grow `Dynamic` unboundedly.
const MAX_REFLECT_DEPTH: usize = 32;

/// Maximum number of entries a single `Each` extraction may emit.
/// 16 384 is the same ceiling we use for the per-document key set;
/// exceeding it suggests a runaway data shape rather than a real
/// indexable field.
pub const MAX_EACH_ENTRIES: usize = 16_384;

/// Extract the set of encoded index keys for `doc` under `spec`.
///
/// - For `Standard`, `Unique`, `Composite`: returns exactly one
///   [`EncodedIndexKey`].
/// - For `Each`: returns one entry per element of the sequence at
///   the configured path. Empty sequence → empty `Vec` (no index
///   work for this doc on this index).
///
/// `collection` is plumbed through purely so error variants carry
/// the collection name in their context (the catalog reconciler
/// in #57 calls this per-collection).
///
/// # Errors
///
/// - [`Error::IndexFieldMissing`] if the configured path is absent.
/// - [`Error::IndexFieldTypeMismatch`] for `Each` on a non-sequence,
///   `Composite` field on a `Map`, etc.
/// - [`Error::InvalidArgument`] if `Each` would emit more than
///   [`MAX_EACH_ENTRIES`].
/// - Propagates encoding errors from [`encode_index_key`].
pub fn extract_index_keys<T: Document>(
    collection: &str,
    spec: &IndexSpec,
    doc: &T,
) -> Result<Vec<EncodedIndexKey>> {
    // #92: project ONLY the indexed field path(s) out of `doc` instead
    // of materializing the whole document via `to_dynamic`. The
    // projecting serializer walks the top-level struct and emits a
    // `Dynamic` for the named field(s) alone — every other field
    // (notably a large `payload`) is visited by a no-op serializer
    // that allocates nothing. The resolved `Dynamic`s are then handed
    // to the SAME `encode_index_key` the full-doc path used, so the
    // produced `EncodedIndexKey` is byte-identical (see the
    // `project_*_byte_identical_to_full_doc` tests).
    let fields = project_fields(collection, spec, doc)?;
    match spec.kind {
        IndexKind::Standard | IndexKind::Unique => extract_scalar(spec, &fields).map(|k| vec![k]),
        IndexKind::Each => extract_each(collection, spec, &fields),
        IndexKind::Composite => extract_composite(spec, &fields).map(|k| vec![k]),
    }
}

/// Encode the single resolved scalar field. Used by `Standard` and
/// `Unique` kinds. `fields` carries exactly one entry — the value at
/// `spec.key_paths[0]` resolved by [`project_fields`].
fn extract_scalar(spec: &IndexSpec, fields: &[Dynamic]) -> Result<EncodedIndexKey> {
    debug_assert_eq!(fields.len(), 1, "scalar kind projects exactly one field");
    encode_index_key(spec, fields)
}

/// Encode one key per element of the resolved sequence field.
fn extract_each(
    collection: &str,
    spec: &IndexSpec,
    fields: &[Dynamic],
) -> Result<Vec<EncodedIndexKey>> {
    debug_assert_eq!(fields.len(), 1, "Each kind projects exactly one field");
    let value = &fields[0];
    let Dynamic::Seq(items) = value else {
        return Err(Error::IndexFieldTypeMismatch {
            collection: collection.to_owned(),
            index: spec.name.clone(),
            path: spec.key_paths[0].clone(),
            expected: "Seq",
            found: dynamic_kind_name(value),
        });
    };
    if items.len() > MAX_EACH_ENTRIES {
        return Err(Error::EachIndexTooLarge {
            collection: collection.to_owned(),
            index: spec.name.clone(),
            len: items.len(),
            max: MAX_EACH_ENTRIES,
        });
    }
    let mut out = Vec::with_capacity(items.len());
    for element in items {
        out.push(encode_field(element)?);
    }
    Ok(out)
}

/// Encode the resolved composite fields into the envelope.
fn extract_composite(spec: &IndexSpec, fields: &[Dynamic]) -> Result<EncodedIndexKey> {
    debug_assert!(fields.len() >= 2, "composite projects ≥ 2 fields");
    encode_index_key(spec, fields)
}

/// Project the `Dynamic` value at each `spec.key_paths` entry out of
/// `doc`, visiting ONLY those fields (every other field is discarded
/// by a no-op serializer — see [`NullSerializer`]). Returns one
/// resolved [`Dynamic`] per path, in `key_paths` order.
///
/// # Errors
///
/// - [`Error::IndexFieldMissing`] if a path is absent from `doc`'s
///   top-level struct (parity with the old `lookup_field` →
///   `Dynamic::get` → `None` path).
/// - [`Error::InvalidArgument`] if `doc` rejects reflection (a
///   non-struct/map top-level, depth overflow, or a `Serialize` impl
///   error) — collapsed from [`DynamicSerError`].
fn project_fields<T: Document>(
    collection: &str,
    spec: &IndexSpec,
    doc: &T,
) -> Result<Vec<Dynamic>> {
    debug_assert!(!spec.key_paths.is_empty(), "spec has ≥ 1 key path");
    let projector = FieldProjector::new(&spec.key_paths);
    let mut resolved = doc.serialize(projector).map_err(Error::from)?;
    let mut out = Vec::with_capacity(spec.key_paths.len());
    for path in &spec.key_paths {
        let value = resolved
            .remove(path)
            .ok_or_else(|| Error::IndexFieldMissing {
                collection: collection.to_owned(),
                index: spec.name.clone(),
                path: path.clone(),
            })?;
        out.push(value);
    }
    Ok(out)
}

/// Diagnostic name of a `Dynamic` variant, used in
/// [`Error::IndexFieldTypeMismatch::found`].
fn dynamic_kind_name(value: &Dynamic) -> &'static str {
    match value {
        Dynamic::Null => "Null",
        Dynamic::Bool(_) => "Bool",
        Dynamic::U64(_) => "U64",
        Dynamic::I64(_) => "I64",
        Dynamic::F64(_) => "F64",
        Dynamic::String(_) => "String",
        Dynamic::Bytes(_) => "Bytes",
        Dynamic::Seq(_) => "Seq",
        Dynamic::Map(_) => "Map",
        Dynamic::Enum { .. } => "Enum",
    }
}

/// Convert a `T: Serialize` into a full `Dynamic` tree by driving
/// serde through the [`DynamicSerializer`].
///
/// This is **not** a postcard round-trip — postcard is not self-
/// describing and cannot reconstruct field names. The serializer
/// emits a `Dynamic::Map` for every struct, keyed by serde's field
/// names; the field values are nested `Dynamic` trees following the
/// same rules recursively (bounded by [`MAX_REFLECT_DEPTH`]).
///
/// # #92 status
///
/// The index-extraction hot path NO LONGER calls this — it projects
/// only the indexed field(s) via [`FieldProjector`] so a large
/// unindexed `payload` is never cloned into a `Dynamic`. The
/// full-document walk is retained (test-gated) as the byte-identity
/// oracle the projecting path is validated against, and as the
/// reference shape for any future full-doc reflection caller
/// (reconcile / migration); `DynamicSerializer` itself remains live
/// on the production path because [`FieldProjector`] reuses it to
/// reflect each matched field's value.
#[cfg(test)]
fn to_dynamic<T: Serialize>(value: &T) -> Result<Dynamic> {
    let ser = DynamicSerializer { depth: 0 };
    value.serialize(ser).map_err(Error::from)
}

// ---------- FieldProjector (#92) ---------------------------------
//
// The field-projecting serializer. Where `DynamicSerializer` builds a
// `Dynamic` for the WHOLE document, `FieldProjector` walks only the
// top-level struct (or map) and emits a `Dynamic` for the field(s)
// named in `wanted`. Every other field is fed to [`NullSerializer`],
// which allocates nothing — in particular it never clones a large
// `payload: Vec<u8>` into a `Dynamic::Bytes`. The named field's value
// IS reflected through the full `DynamicSerializer`, so the resolved
// `Dynamic` is bit-for-bit what the full-doc path would have produced
// for that field (the byte-identity guardrail).

/// Top-level field-projecting `Serializer`. Collects the values of the
/// `wanted` field names into `collected`; discards every other field.
///
/// Only the struct / struct-variant / map top-level shapes collect
/// anything — a top-level scalar / seq / enum has no named fields, so
/// `collected` stays empty and [`project_fields`] surfaces the same
/// [`Error::IndexFieldMissing`] the old `Dynamic::get`-on-a-non-Map
/// path produced (error parity).
struct FieldProjector<'w> {
    wanted: &'w [String],
}

impl<'w> FieldProjector<'w> {
    fn new(wanted: &'w [String]) -> Self {
        Self { wanted }
    }
}

/// The accumulator a `FieldProjector` hands back: the resolved
/// `Dynamic` for each matched field name. A `BTreeMap` (not a `Vec`)
/// so `project_fields` can pull paths out in `key_paths` order and so
/// a duplicate field name (impossible for a real struct) cannot grow
/// the result unboundedly.
type ProjectedFields = BTreeMap<String, Dynamic>;

/// Builder shared by `SerializeStruct` / `SerializeStructVariant` /
/// `SerializeMap` for the projecting walk. Holds the wanted-name set
/// and the fields matched so far.
struct ProjectBuilder<'w> {
    wanted: &'w [String],
    collected: ProjectedFields,
    pending_key: Option<String>,
}

impl<'w> ProjectBuilder<'w> {
    fn new(wanted: &'w [String]) -> Self {
        Self {
            wanted,
            collected: BTreeMap::new(),
            pending_key: None,
        }
    }

    /// Reflect `value` into a `Dynamic` only if `key` is wanted;
    /// otherwise discard it through [`NullSerializer`] (no allocation).
    fn take_field<T: ?Sized + Serialize>(&mut self, key: &str, value: &T) -> DynRes<()> {
        if self.wanted.iter().any(|w| w == key) {
            let val = value.serialize(DynamicSerializer { depth: 1 })?;
            self.collected.insert(key.to_owned(), val);
        } else {
            value.serialize(NullSerializer)?;
        }
        Ok(())
    }
}

#[allow(clippy::unused_self)] // serde's by-value receiver; mirrors DynamicSerializer
impl<'w> ser::Serializer for FieldProjector<'w> {
    type Ok = ProjectedFields;
    type Error = DynamicSerError;
    type SerializeSeq = ser::Impossible<ProjectedFields, DynamicSerError>;
    type SerializeTuple = ser::Impossible<ProjectedFields, DynamicSerError>;
    type SerializeTupleStruct = ser::Impossible<ProjectedFields, DynamicSerError>;
    type SerializeTupleVariant = ser::Impossible<ProjectedFields, DynamicSerError>;
    type SerializeMap = ProjectBuilder<'w>;
    type SerializeStruct = ProjectBuilder<'w>;
    type SerializeStructVariant = ProjectBuilder<'w>;

    fn serialize_struct(self, _name: &'static str, _len: usize) -> DynRes<ProjectBuilder<'w>> {
        Ok(ProjectBuilder::new(self.wanted))
    }
    fn serialize_struct_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        _len: usize,
    ) -> DynRes<ProjectBuilder<'w>> {
        Ok(ProjectBuilder::new(self.wanted))
    }
    fn serialize_map(self, _len: Option<usize>) -> DynRes<ProjectBuilder<'w>> {
        Ok(ProjectBuilder::new(self.wanted))
    }
    // Transparent wrappers recurse with the same projector so a
    // `#[serde(transparent)]` newtype or an `Option<Struct>` top level
    // still projects the inner struct's fields (parity with
    // `DynamicSerializer`'s pass-through of these shapes).
    fn serialize_newtype_struct<T: ?Sized + Serialize>(
        self,
        _name: &'static str,
        v: &T,
    ) -> DynRes<ProjectedFields> {
        v.serialize(self)
    }
    fn serialize_some<T: ?Sized + Serialize>(self, v: &T) -> DynRes<ProjectedFields> {
        v.serialize(self)
    }
    // Every non-struct top level resolves to "no named fields" — an
    // empty map. `project_fields` turns that into IndexFieldMissing,
    // matching the old `Dynamic::get` on a non-Map returning `None`.
    fn serialize_i128(self, _v: i128) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_u128(self, _v: u128) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_bool(self, _v: bool) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_i8(self, _v: i8) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_i16(self, _v: i16) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_i32(self, _v: i32) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_i64(self, _v: i64) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_u8(self, _v: u8) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_u16(self, _v: u16) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_u32(self, _v: u32) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_u64(self, _v: u64) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_f32(self, _v: f32) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_f64(self, _v: f64) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_char(self, _v: char) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_str(self, _v: &str) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_bytes(self, _v: &[u8]) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_none(self) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_unit(self) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_unit_struct(self, _name: &'static str) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_unit_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
    ) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_newtype_variant<T: ?Sized + Serialize>(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        _v: &T,
    ) -> DynRes<ProjectedFields> {
        Ok(BTreeMap::new())
    }
    fn serialize_seq(self, _len: Option<usize>) -> DynRes<Self::SerializeSeq> {
        Err(seq_unsupported())
    }
    fn serialize_tuple(self, _len: usize) -> DynRes<Self::SerializeTuple> {
        Err(seq_unsupported())
    }
    fn serialize_tuple_struct(
        self,
        _name: &'static str,
        _len: usize,
    ) -> DynRes<Self::SerializeTupleStruct> {
        Err(seq_unsupported())
    }
    fn serialize_tuple_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        _len: usize,
    ) -> DynRes<Self::SerializeTupleVariant> {
        Err(seq_unsupported())
    }
}

/// A top-level tuple / tuple-struct / tuple-variant has positional
/// (not named) fields and so contributes no addressable index field —
/// the same outcome as `Dynamic::get` on a non-Map. Surface it as the
/// reflection-rejected error rather than silently succeeding with no
/// fields, so a mis-shaped spec is not masked.
fn seq_unsupported() -> DynamicSerError {
    DynamicSerError("index extraction: top-level tuple has no named fields to project".to_owned())
}

impl ser::SerializeStruct for ProjectBuilder<'_> {
    type Ok = ProjectedFields;
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(
        &mut self,
        key: &'static str,
        value: &T,
    ) -> DynRes<()> {
        self.take_field(key, value)
    }
    fn end(self) -> DynRes<ProjectedFields> {
        Ok(self.collected)
    }
}

impl ser::SerializeStructVariant for ProjectBuilder<'_> {
    type Ok = ProjectedFields;
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(
        &mut self,
        key: &'static str,
        value: &T,
    ) -> DynRes<()> {
        self.take_field(key, value)
    }
    fn end(self) -> DynRes<ProjectedFields> {
        Ok(self.collected)
    }
}

impl ser::SerializeMap for ProjectBuilder<'_> {
    type Ok = ProjectedFields;
    type Error = DynamicSerError;
    fn serialize_key<T: ?Sized + Serialize>(&mut self, key: &T) -> DynRes<()> {
        // Resolve the key to a String the same way `DynamicSerializer`'s
        // MapBuilder does, so a stringable map key projects identically
        // to a struct field of the same name.
        let key_dyn = key.serialize(DynamicSerializer { depth: 1 })?;
        let key_string = match key_dyn {
            Dynamic::String(s) => s,
            Dynamic::U64(n) => n.to_string(),
            Dynamic::I64(n) => n.to_string(),
            Dynamic::Bool(b) => b.to_string(),
            other => {
                return Err(DynamicSerError(format!(
                    "map key must be stringable (got {other:?})"
                )));
            }
        };
        self.pending_key = Some(key_string);
        Ok(())
    }
    fn serialize_value<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        let key = self
            .pending_key
            .take()
            .ok_or_else(|| DynamicSerError("map value without preceding key".to_owned()))?;
        self.take_field(&key, value)
    }
    fn end(self) -> DynRes<ProjectedFields> {
        Ok(self.collected)
    }
}

// ---------- NullSerializer (#92) ---------------------------------
//
// A `Serializer` that discards everything it visits, allocating
// nothing. Used by `ProjectBuilder` for the document's unwanted
// fields — most importantly the multi-hundred-byte `payload` that the
// old `to_dynamic` path cloned into a `Dynamic::Bytes` (and then
// dropped). Every method returns the unit `Ok = ()`; the composite
// methods return `Self` so nested unwanted structures are likewise
// drained without building a `Dynamic`. The walk is depth-UNBOUNDED
// by design: it allocates no per-node memory, so a deep unwanted
// subtree costs only stack frames inside serde's own bounded derive
// output — never a `Dynamic` node.

/// No-op serializer: visits and discards, allocates nothing.
struct NullSerializer;

#[allow(clippy::unused_self)] // serde's by-value receiver; methods are intentionally no-ops
impl ser::Serializer for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    type SerializeSeq = Self;
    type SerializeTuple = Self;
    type SerializeTupleStruct = Self;
    type SerializeTupleVariant = Self;
    type SerializeMap = Self;
    type SerializeStruct = Self;
    type SerializeStructVariant = Self;

    fn serialize_bool(self, _v: bool) -> DynRes<()> {
        Ok(())
    }
    fn serialize_i8(self, _v: i8) -> DynRes<()> {
        Ok(())
    }
    fn serialize_i16(self, _v: i16) -> DynRes<()> {
        Ok(())
    }
    fn serialize_i32(self, _v: i32) -> DynRes<()> {
        Ok(())
    }
    fn serialize_i64(self, _v: i64) -> DynRes<()> {
        Ok(())
    }
    fn serialize_u8(self, _v: u8) -> DynRes<()> {
        Ok(())
    }
    fn serialize_u16(self, _v: u16) -> DynRes<()> {
        Ok(())
    }
    fn serialize_u32(self, _v: u32) -> DynRes<()> {
        Ok(())
    }
    fn serialize_u64(self, _v: u64) -> DynRes<()> {
        Ok(())
    }
    fn serialize_f32(self, _v: f32) -> DynRes<()> {
        Ok(())
    }
    fn serialize_f64(self, _v: f64) -> DynRes<()> {
        Ok(())
    }
    fn serialize_char(self, _v: char) -> DynRes<()> {
        Ok(())
    }
    fn serialize_str(self, _v: &str) -> DynRes<()> {
        Ok(())
    }
    fn serialize_bytes(self, _v: &[u8]) -> DynRes<()> {
        Ok(())
    }
    fn serialize_none(self) -> DynRes<()> {
        Ok(())
    }
    fn serialize_some<T: ?Sized + Serialize>(self, v: &T) -> DynRes<()> {
        v.serialize(self)
    }
    fn serialize_unit(self) -> DynRes<()> {
        Ok(())
    }
    fn serialize_unit_struct(self, _name: &'static str) -> DynRes<()> {
        Ok(())
    }
    fn serialize_unit_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
    ) -> DynRes<()> {
        Ok(())
    }
    fn serialize_newtype_struct<T: ?Sized + Serialize>(
        self,
        _name: &'static str,
        v: &T,
    ) -> DynRes<()> {
        v.serialize(self)
    }
    fn serialize_newtype_variant<T: ?Sized + Serialize>(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        v: &T,
    ) -> DynRes<()> {
        v.serialize(self)
    }
    fn serialize_seq(self, _len: Option<usize>) -> DynRes<Self> {
        Ok(self)
    }
    fn serialize_tuple(self, _len: usize) -> DynRes<Self> {
        Ok(self)
    }
    fn serialize_tuple_struct(self, _name: &'static str, _len: usize) -> DynRes<Self> {
        Ok(self)
    }
    fn serialize_tuple_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        _len: usize,
    ) -> DynRes<Self> {
        Ok(self)
    }
    fn serialize_map(self, _len: Option<usize>) -> DynRes<Self> {
        Ok(self)
    }
    fn serialize_struct(self, _name: &'static str, _len: usize) -> DynRes<Self> {
        Ok(self)
    }
    fn serialize_struct_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        _len: usize,
    ) -> DynRes<Self> {
        Ok(self)
    }
}

impl ser::SerializeSeq for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    fn serialize_element<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        value.serialize(NullSerializer)
    }
    fn end(self) -> DynRes<()> {
        Ok(())
    }
}

impl ser::SerializeTuple for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    fn serialize_element<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        value.serialize(NullSerializer)
    }
    fn end(self) -> DynRes<()> {
        Ok(())
    }
}

impl ser::SerializeTupleStruct for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        value.serialize(NullSerializer)
    }
    fn end(self) -> DynRes<()> {
        Ok(())
    }
}

impl ser::SerializeTupleVariant for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        value.serialize(NullSerializer)
    }
    fn end(self) -> DynRes<()> {
        Ok(())
    }
}

impl ser::SerializeMap for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    fn serialize_key<T: ?Sized + Serialize>(&mut self, key: &T) -> DynRes<()> {
        key.serialize(NullSerializer)
    }
    fn serialize_value<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        value.serialize(NullSerializer)
    }
    fn end(self) -> DynRes<()> {
        Ok(())
    }
}

impl ser::SerializeStruct for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(
        &mut self,
        _key: &'static str,
        value: &T,
    ) -> DynRes<()> {
        value.serialize(NullSerializer)
    }
    fn end(self) -> DynRes<()> {
        Ok(())
    }
}

impl ser::SerializeStructVariant for NullSerializer {
    type Ok = ();
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(
        &mut self,
        _key: &'static str,
        value: &T,
    ) -> DynRes<()> {
        value.serialize(NullSerializer)
    }
    fn end(self) -> DynRes<()> {
        Ok(())
    }
}

// ---------- DynamicSerializer ------------------------------------

/// Errors surfaced from the [`DynamicSerializer`].
///
/// Lives next to the serializer because every serde method returns
/// a `Result<_, Self::Error>`. The error carries an owned message
/// so a wrapped `serde::ser::Error::custom` does not leak `'static`.
#[derive(Debug)]
struct DynamicSerError(String);

impl std::fmt::Display for DynamicSerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for DynamicSerError {}

impl ser::Error for DynamicSerError {
    fn custom<T: std::fmt::Display>(msg: T) -> Self {
        Self(msg.to_string())
    }
}

impl From<DynamicSerError> for Error {
    fn from(_e: DynamicSerError) -> Self {
        // The serializer's errors are programmer-bug paths in the
        // `Document` impl (non-stringable map keys, depth overflow,
        // a user `Serialize` impl returning a custom error). All
        // collapse to a single static `Error::InvalidArgument` so
        // we do not need to grow `Error` with an owned-string
        // variant just for this path. The serializer's `Display`
        // body keeps the per-instance message available for logs
        // if a caller `Debug`-prints the wrapped error.
        Error::InvalidArgument(
            "index extraction: Document Serialize impl rejected reflection \
             (see DynamicSerError for the detail)",
        )
    }
}

/// Serde `Serializer` that converts the visited value into a
/// `Dynamic` tree. Bounded-depth (Rule 1) via the `depth` counter.
struct DynamicSerializer {
    depth: usize,
}

impl DynamicSerializer {
    fn deeper(&self) -> Result<Self> {
        let next = self
            .depth
            .checked_add(1)
            .ok_or(Error::InvalidArgument("index extraction: depth overflow"))?;
        if next >= MAX_REFLECT_DEPTH {
            return Err(Error::InvalidArgument(
                "index extraction: max reflection depth exceeded",
            ));
        }
        Ok(Self { depth: next })
    }
}

/// Local helper for serde plumbing: a `Result<Dynamic,
/// DynamicSerError>` alias.
type DynRes<T = Dynamic> = std::result::Result<T, DynamicSerError>;

#[allow(clippy::cast_possible_truncation)] // i128 → i64 / u128 → u64 narrowing is the encoded form
#[allow(clippy::cast_sign_loss)] // sign loss on the unsigned-only `serialize_u*` is intentional
impl ser::Serializer for DynamicSerializer {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    type SerializeSeq = SeqBuilder;
    type SerializeTuple = SeqBuilder;
    type SerializeTupleStruct = SeqBuilder;
    type SerializeTupleVariant = SeqBuilder;
    type SerializeMap = MapBuilder;
    type SerializeStruct = MapBuilder;
    type SerializeStructVariant = MapBuilder;

    fn serialize_bool(self, v: bool) -> DynRes {
        Ok(Dynamic::Bool(v))
    }
    fn serialize_i8(self, v: i8) -> DynRes {
        Ok(Dynamic::I64(i64::from(v)))
    }
    fn serialize_i16(self, v: i16) -> DynRes {
        Ok(Dynamic::I64(i64::from(v)))
    }
    fn serialize_i32(self, v: i32) -> DynRes {
        Ok(Dynamic::I64(i64::from(v)))
    }
    fn serialize_i64(self, v: i64) -> DynRes {
        Ok(Dynamic::I64(v))
    }
    fn serialize_i128(self, v: i128) -> DynRes {
        Ok(Dynamic::I64(v as i64))
    }
    fn serialize_u8(self, v: u8) -> DynRes {
        Ok(Dynamic::U64(u64::from(v)))
    }
    fn serialize_u16(self, v: u16) -> DynRes {
        Ok(Dynamic::U64(u64::from(v)))
    }
    fn serialize_u32(self, v: u32) -> DynRes {
        Ok(Dynamic::U64(u64::from(v)))
    }
    fn serialize_u64(self, v: u64) -> DynRes {
        Ok(Dynamic::U64(v))
    }
    fn serialize_u128(self, v: u128) -> DynRes {
        Ok(Dynamic::U64(v as u64))
    }
    fn serialize_f32(self, v: f32) -> DynRes {
        Ok(Dynamic::F64(f64::from(v)))
    }
    fn serialize_f64(self, v: f64) -> DynRes {
        Ok(Dynamic::F64(v))
    }
    fn serialize_char(self, v: char) -> DynRes {
        Ok(Dynamic::String(v.to_string()))
    }
    fn serialize_str(self, v: &str) -> DynRes {
        Ok(Dynamic::String(v.to_owned()))
    }
    fn serialize_bytes(self, v: &[u8]) -> DynRes {
        Ok(Dynamic::Bytes(v.to_vec()))
    }
    fn serialize_none(self) -> DynRes {
        Ok(Dynamic::Null)
    }
    fn serialize_some<T: ?Sized + Serialize>(self, v: &T) -> DynRes {
        v.serialize(self)
    }
    fn serialize_unit(self) -> DynRes {
        Ok(Dynamic::Null)
    }
    fn serialize_unit_struct(self, _name: &'static str) -> DynRes {
        Ok(Dynamic::Null)
    }
    fn serialize_unit_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        variant: &'static str,
    ) -> DynRes {
        Ok(Dynamic::String(variant.to_owned()))
    }
    fn serialize_newtype_struct<T: ?Sized + Serialize>(self, _name: &'static str, v: &T) -> DynRes {
        v.serialize(self)
    }
    fn serialize_newtype_variant<T: ?Sized + Serialize>(
        self,
        _name: &'static str,
        _variant_index: u32,
        variant: &'static str,
        v: &T,
    ) -> DynRes {
        let inner = v.serialize(deeper(&self)?)?;
        let mut m = BTreeMap::new();
        m.insert(variant.to_owned(), inner);
        Ok(Dynamic::Map(m))
    }
    fn serialize_seq(self, len: Option<usize>) -> DynRes<SeqBuilder> {
        let cap = len.unwrap_or(0).min(MAX_EACH_ENTRIES);
        Ok(SeqBuilder {
            depth: self.depth,
            items: Vec::with_capacity(cap),
        })
    }
    fn serialize_tuple(self, len: usize) -> DynRes<SeqBuilder> {
        self.serialize_seq(Some(len))
    }
    fn serialize_tuple_struct(self, _name: &'static str, len: usize) -> DynRes<SeqBuilder> {
        self.serialize_seq(Some(len))
    }
    fn serialize_tuple_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        len: usize,
    ) -> DynRes<SeqBuilder> {
        self.serialize_seq(Some(len))
    }
    fn serialize_map(self, _len: Option<usize>) -> DynRes<MapBuilder> {
        Ok(MapBuilder {
            depth: self.depth,
            map: BTreeMap::new(),
            pending_key: None,
        })
    }
    fn serialize_struct(self, _name: &'static str, _len: usize) -> DynRes<MapBuilder> {
        self.serialize_map(None)
    }
    fn serialize_struct_variant(
        self,
        _name: &'static str,
        _variant_index: u32,
        _variant: &'static str,
        _len: usize,
    ) -> DynRes<MapBuilder> {
        self.serialize_map(None)
    }
}

/// Construct a deeper [`DynamicSerializer`] for nested entries; the
/// result mirrors `self.depth + 1` and trips
/// [`MAX_REFLECT_DEPTH`].
fn deeper(s: &DynamicSerializer) -> DynRes<DynamicSerializer> {
    s.deeper().map_err(|e| {
        DynamicSerError(match e {
            Error::InvalidArgument(msg) => msg.to_owned(),
            other => other.to_string(),
        })
    })
}

/// Builder for sequence / tuple / tuple-struct / tuple-variant
/// shapes. Each `serialize_element` recurses into a deeper
/// `DynamicSerializer`.
struct SeqBuilder {
    depth: usize,
    items: Vec<Dynamic>,
}

impl ser::SerializeSeq for SeqBuilder {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    fn serialize_element<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        if self.items.len() >= MAX_EACH_ENTRIES {
            return Err(DynamicSerError(
                "index extraction: sequence exceeds MAX_EACH_ENTRIES".to_owned(),
            ));
        }
        let item = value.serialize(DynamicSerializer {
            depth: self.depth + 1,
        })?;
        self.items.push(item);
        Ok(())
    }
    fn end(self) -> DynRes {
        Ok(Dynamic::Seq(self.items))
    }
}

impl ser::SerializeTuple for SeqBuilder {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    fn serialize_element<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        <Self as ser::SerializeSeq>::serialize_element(self, value)
    }
    fn end(self) -> DynRes {
        <Self as ser::SerializeSeq>::end(self)
    }
}

impl ser::SerializeTupleStruct for SeqBuilder {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        <Self as ser::SerializeSeq>::serialize_element(self, value)
    }
    fn end(self) -> DynRes {
        <Self as ser::SerializeSeq>::end(self)
    }
}

impl ser::SerializeTupleVariant for SeqBuilder {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        <Self as ser::SerializeSeq>::serialize_element(self, value)
    }
    fn end(self) -> DynRes {
        <Self as ser::SerializeSeq>::end(self)
    }
}

/// Builder for map / struct / struct-variant shapes. Uses
/// `pending_key` to handle the `serialize_key` / `serialize_value`
/// pairing required by `SerializeMap`.
struct MapBuilder {
    depth: usize,
    map: BTreeMap<String, Dynamic>,
    pending_key: Option<String>,
}

impl MapBuilder {
    fn deeper_serializer(&self) -> DynamicSerializer {
        DynamicSerializer {
            depth: self.depth + 1,
        }
    }
}

impl ser::SerializeMap for MapBuilder {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    fn serialize_key<T: ?Sized + Serialize>(&mut self, key: &T) -> DynRes<()> {
        let key_dyn = key.serialize(self.deeper_serializer())?;
        let key_string = match key_dyn {
            Dynamic::String(s) => s,
            Dynamic::U64(n) => n.to_string(),
            Dynamic::I64(n) => n.to_string(),
            Dynamic::Bool(b) => b.to_string(),
            other => {
                return Err(DynamicSerError(format!(
                    "map key must be stringable (got {other:?})"
                )));
            }
        };
        self.pending_key = Some(key_string);
        Ok(())
    }
    fn serialize_value<T: ?Sized + Serialize>(&mut self, value: &T) -> DynRes<()> {
        let key = self
            .pending_key
            .take()
            .ok_or_else(|| DynamicSerError("map value without preceding key".to_owned()))?;
        let val = value.serialize(self.deeper_serializer())?;
        self.map.insert(key, val);
        Ok(())
    }
    fn end(self) -> DynRes {
        Ok(Dynamic::Map(self.map))
    }
}

impl ser::SerializeStruct for MapBuilder {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(
        &mut self,
        key: &'static str,
        value: &T,
    ) -> DynRes<()> {
        let val = value.serialize(self.deeper_serializer())?;
        self.map.insert(key.to_owned(), val);
        Ok(())
    }
    fn end(self) -> DynRes {
        Ok(Dynamic::Map(self.map))
    }
}

impl ser::SerializeStructVariant for MapBuilder {
    type Ok = Dynamic;
    type Error = DynamicSerError;
    fn serialize_field<T: ?Sized + Serialize>(
        &mut self,
        key: &'static str,
        value: &T,
    ) -> DynRes<()> {
        <Self as ser::SerializeStruct>::serialize_field(self, key, value)
    }
    fn end(self) -> DynRes {
        <Self as ser::SerializeStruct>::end(self)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde::{Deserialize, Serialize};

    #[derive(Debug, Serialize, Deserialize)]
    struct Customer {
        email: String,
        score: i64,
        tags: Vec<String>,
    }

    impl Document for Customer {
        const COLLECTION: &'static str = "customers";
        const VERSION: u32 = 1;
    }

    #[derive(Debug, Serialize, Deserialize)]
    struct Order {
        customer_id: u64,
        placed_at: u64,
        amount_cents: i64,
    }

    impl Document for Order {
        const COLLECTION: &'static str = "orders";
        const VERSION: u32 = 1;
    }

    #[test]
    fn dynamic_reflection_of_simple_struct() {
        let c = Customer {
            email: "ada@example.com".to_owned(),
            score: 42,
            tags: vec!["alpha".to_owned(), "beta".to_owned()],
        };
        let d = to_dynamic(&c).expect("reflect");
        let Dynamic::Map(map) = &d else {
            panic!("expected Map, got {d:?}");
        };
        assert_eq!(
            map.get("email"),
            Some(&Dynamic::String("ada@example.com".to_owned())),
        );
        assert_eq!(map.get("score"), Some(&Dynamic::I64(42)));
        match map.get("tags") {
            Some(Dynamic::Seq(items)) => assert_eq!(items.len(), 2),
            other => panic!("expected Seq, got {other:?}"),
        }
    }

    #[test]
    fn standard_extract_returns_one_key() {
        let c = Customer {
            email: "ada@example.com".to_owned(),
            score: 7,
            tags: vec![],
        };
        let spec = IndexSpec::standard("by_email", "email").expect("spec");
        let keys = extract_index_keys(Customer::COLLECTION, &spec, &c).expect("extract");
        assert_eq!(keys.len(), 1);
        // Encoding should be the order-preserving String form —
        // double-check by encoding the same Dynamic directly.
        let expected = encode_field(&Dynamic::String("ada@example.com".to_owned())).expect("enc");
        assert_eq!(keys[0], expected);
    }

    #[test]
    fn unique_extract_returns_one_key() {
        let c = Customer {
            email: "u@e.com".to_owned(),
            score: 1,
            tags: vec![],
        };
        let spec = IndexSpec::unique("by_email", "email").expect("spec");
        let keys = extract_index_keys(Customer::COLLECTION, &spec, &c).expect("extract");
        assert_eq!(keys.len(), 1);
    }

    #[test]
    fn each_extract_returns_n_keys() {
        let c = Customer {
            email: "x".to_owned(),
            score: 0,
            tags: vec!["a".to_owned(), "b".to_owned(), "c".to_owned()],
        };
        let spec = IndexSpec::each("by_tag", "tags").expect("spec");
        let keys = extract_index_keys(Customer::COLLECTION, &spec, &c).expect("extract");
        assert_eq!(keys.len(), 3);
        // Each entry is the encoded String of one element.
        let want_a = encode_field(&Dynamic::String("a".to_owned())).expect("enc");
        assert_eq!(keys[0], want_a);
    }

    #[test]
    fn each_extract_on_empty_seq_returns_empty_vec() {
        let c = Customer {
            email: "x".to_owned(),
            score: 0,
            tags: vec![],
        };
        let spec = IndexSpec::each("by_tag", "tags").expect("spec");
        let keys = extract_index_keys(Customer::COLLECTION, &spec, &c).expect("extract");
        assert!(keys.is_empty());
    }

    #[test]
    fn composite_extract_returns_one_envelope_key() {
        let o = Order {
            customer_id: 7,
            placed_at: 12_345,
            amount_cents: 100,
        };
        let spec = IndexSpec::composite("by_ct", &["customer_id", "placed_at"]).expect("spec");
        let keys = extract_index_keys(Order::COLLECTION, &spec, &o).expect("extract");
        assert_eq!(keys.len(), 1);
        // Envelope tag must be the documented value.
        assert_eq!(keys[0].as_bytes()[0], crate::index::key::COMPOSITE_TAG);
    }

    #[test]
    fn missing_field_path_errors() {
        let c = Customer {
            email: "x".to_owned(),
            score: 0,
            tags: vec![],
        };
        let spec = IndexSpec::standard("by_nope", "nope").expect("spec");
        let err = extract_index_keys(Customer::COLLECTION, &spec, &c).expect_err("missing");
        match err {
            Error::IndexFieldMissing {
                collection,
                index,
                path,
            } => {
                assert_eq!(collection, "customers");
                assert_eq!(index, "by_nope");
                assert_eq!(path, "nope");
            }
            other => panic!("expected IndexFieldMissing, got {other:?}"),
        }
    }

    #[test]
    fn each_on_non_seq_field_errors() {
        let c = Customer {
            email: "x".to_owned(),
            score: 99,
            tags: vec![],
        };
        // `score` is an i64, not a Vec — Each must reject.
        let spec = IndexSpec::each("by_score", "score").expect("spec");
        let err = extract_index_keys(Customer::COLLECTION, &spec, &c).expect_err("type");
        match err {
            Error::IndexFieldTypeMismatch {
                collection,
                index,
                path,
                expected,
                found,
            } => {
                assert_eq!(collection, "customers");
                assert_eq!(index, "by_score");
                assert_eq!(path, "score");
                assert_eq!(expected, "Seq");
                assert_eq!(found, "I64");
            }
            other => panic!("expected IndexFieldTypeMismatch, got {other:?}"),
        }
    }

    #[test]
    fn composite_decode_round_trip_matches_direct_encoding() {
        let o = Order {
            customer_id: 7,
            placed_at: 12_345,
            amount_cents: 100,
        };
        let spec = IndexSpec::composite("by_ct", &["customer_id", "placed_at"]).expect("spec");
        let extracted = extract_index_keys(Order::COLLECTION, &spec, &o).expect("extract");
        // Manually build the same composite by reflecting + encoding.
        let direct = encode_index_key(
            &spec,
            &[Dynamic::U64(o.customer_id), Dynamic::U64(o.placed_at)],
        )
        .expect("direct");
        assert_eq!(extracted[0], direct);
    }

    // ---------- #92 byte-identity guardrail ----------------------
    //
    // The projecting path (`extract_index_keys`, which now routes
    // Standard / Unique / Composite / Each through `FieldProjector`)
    // MUST yield an `EncodedIndexKey` byte-identical to the legacy
    // full-document path (`to_dynamic` + `lookup_field` + encode). The
    // on-disk index-key format must not move. `legacy_extract_*`
    // reproduces the pre-#92 logic verbatim using the still-present
    // `to_dynamic`, then every shape asserts equality against the new
    // path.

    /// Legacy scalar/each extraction: full-doc reflect → `get` →
    /// encode, exactly as the code did before #92.
    fn legacy_extract<T: Document>(
        collection: &str,
        spec: &IndexSpec,
        doc: &T,
    ) -> Result<Vec<EncodedIndexKey>> {
        let dynamic = to_dynamic(doc)?;
        let lookup = |path: &str| -> Result<Dynamic> {
            dynamic
                .get(path)
                .cloned()
                .ok_or_else(|| Error::IndexFieldMissing {
                    collection: collection.to_owned(),
                    index: spec.name.clone(),
                    path: path.to_owned(),
                })
        };
        match spec.kind {
            IndexKind::Standard | IndexKind::Unique => {
                let v = lookup(&spec.key_paths[0])?;
                Ok(vec![encode_index_key(spec, std::slice::from_ref(&v))?])
            }
            IndexKind::Each => {
                let v = lookup(&spec.key_paths[0])?;
                let Dynamic::Seq(items) = v else {
                    panic!("legacy each on non-seq");
                };
                items.iter().map(encode_field).collect()
            }
            IndexKind::Composite => {
                let mut fields = Vec::new();
                for p in &spec.key_paths {
                    fields.push(lookup(p)?);
                }
                Ok(vec![encode_index_key(spec, &fields)?])
            }
        }
    }

    /// Every indexable field shape, in a single doc, so one struct
    /// exercises u64 / i64 (sign split) / String / bool / Option
    /// (Some + None) / newtype / enum / f64 / bytes for Standard,
    /// Unique, and Composite.
    #[derive(Debug, Serialize, Deserialize)]
    struct Shapes {
        u: u64,
        i_pos: i64,
        i_neg: i64,
        i_zero: i64,
        s: String,
        flag: bool,
        opt_some: Option<u64>,
        opt_none: Option<u64>,
        nt: Newtype,
        en: Color,
        f: f64,
        b: Bytes,
        payload: Vec<u8>, // the unindexed "big" field #92 must skip
    }

    #[derive(Debug, Serialize, Deserialize)]
    struct Newtype(i64);

    #[derive(Debug, Serialize, Deserialize)]
    enum Color {
        Red,
        Green,
    }

    /// A bytes-valued field. Hand-written `Serialize` so it drives
    /// `serialize_bytes` (→ `Dynamic::Bytes`) rather than a per-element
    /// seq — the shape the encoder's `TAG_BYTES` arm consumes.
    #[derive(Debug)]
    struct Bytes(Vec<u8>);

    impl Serialize for Bytes {
        fn serialize<S: serde::Serializer>(&self, ser: S) -> std::result::Result<S::Ok, S::Error> {
            ser.serialize_bytes(&self.0)
        }
    }

    impl<'de> Deserialize<'de> for Bytes {
        fn deserialize<D: serde::Deserializer<'de>>(de: D) -> std::result::Result<Self, D::Error> {
            // Only `Serialize` is exercised by extraction; a permissive
            // byte-buf deserialize keeps `Shapes: DeserializeOwned`
            // (the `Document` bound) satisfied.
            let v = <Vec<u8>>::deserialize(de)?;
            Ok(Bytes(v))
        }
    }

    impl Document for Shapes {
        const COLLECTION: &'static str = "shapes";
        const VERSION: u32 = 1;
    }

    fn sample_shapes() -> Shapes {
        Shapes {
            u: 7,
            i_pos: 42,
            i_neg: -42,
            i_zero: 0,
            s: "hello".to_owned(),
            flag: true,
            opt_some: Some(99),
            opt_none: None,
            nt: Newtype(-1),
            en: Color::Green,
            f: -1.5,
            b: Bytes(vec![0x00, 0x01, 0xFF]),
            payload: vec![0xAB; 480],
        }
    }

    /// Assert the projecting path is byte-identical to the legacy
    /// full-doc path for `spec` against `doc`.
    fn assert_byte_identical<T: Document>(spec: &IndexSpec, doc: &T) {
        let new = extract_index_keys(T::COLLECTION, spec, doc).expect("project extract");
        let old = legacy_extract(T::COLLECTION, spec, doc).expect("legacy extract");
        assert_eq!(
            new, old,
            "byte mismatch for spec={spec:?}: projecting path diverged from full-doc path"
        );
    }

    #[test]
    fn project_standard_unique_byte_identical_to_full_doc() {
        let doc = sample_shapes();
        // One Standard + one Unique spec per scalar field shape.
        for field in [
            "u", "i_pos", "i_neg", "i_zero", "s", "flag", "opt_some", "opt_none", "nt", "en", "f",
            "b",
        ] {
            let std = IndexSpec::standard("ix", field).expect("standard spec");
            assert_byte_identical(&std, &doc);
            let uniq = IndexSpec::unique("ix", field).expect("unique spec");
            assert_byte_identical(&uniq, &doc);
        }
    }

    #[test]
    fn project_composite_byte_identical_to_full_doc() {
        let doc = sample_shapes();
        // Multi-field composites spanning the variant zoo.
        let cases: &[&[&str]] = &[
            &["u", "i_neg"],
            &["s", "flag"],
            &["i_pos", "i_zero", "u"],
            &["opt_some", "opt_none"],
            &["nt", "en", "f"],
            &["b", "u", "s"],
        ];
        for paths in cases {
            let spec = IndexSpec::composite("ix", paths).expect("composite spec");
            assert_byte_identical(&spec, &doc);
        }
    }

    #[test]
    fn project_each_byte_identical_to_full_doc() {
        // `Each` over a Vec must still produce the per-element keys
        // identical to the full-doc path.
        let c = Customer {
            email: "x".to_owned(),
            score: 0,
            tags: vec!["a".to_owned(), "bb".to_owned(), "ccc".to_owned()],
        };
        let spec = IndexSpec::each("by_tag", "tags").expect("each spec");
        assert_byte_identical(&spec, &c);
        // Empty seq → empty key set on both paths.
        let empty = Customer {
            email: "x".to_owned(),
            score: 0,
            tags: vec![],
        };
        assert_byte_identical(&spec, &empty);
    }

    #[test]
    fn project_missing_field_errors_identically() {
        let doc = sample_shapes();
        let spec = IndexSpec::standard("by_nope", "nope").expect("spec");
        let new = extract_index_keys(Shapes::COLLECTION, &spec, &doc).expect_err("missing");
        let old = legacy_extract(Shapes::COLLECTION, &spec, &doc).expect_err("legacy missing");
        // Same variant + same fields.
        assert!(matches!(new, Error::IndexFieldMissing { .. }));
        assert_eq!(format!("{new:?}"), format!("{old:?}"));
    }

    #[test]
    fn project_each_on_non_seq_errors_identically() {
        let doc = sample_shapes();
        // `u` is a u64, not a Vec — Each must reject with a type
        // mismatch, identical to the full-doc path's `found`.
        let spec = IndexSpec::each("by_u", "u").expect("spec");
        let new = extract_index_keys(Shapes::COLLECTION, &spec, &doc).expect_err("type");
        match new {
            Error::IndexFieldTypeMismatch {
                ref path,
                expected,
                found,
                ..
            } => {
                assert_eq!(path, "u");
                assert_eq!(expected, "Seq");
                assert_eq!(found, "U64");
            }
            other => panic!("expected IndexFieldTypeMismatch, got {other:?}"),
        }
    }
}