armdb 0.11.2

//! Secondary-index projection carrier and sink (T8).

use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;

use crate::Key;
use crate::counter::{CounterSink, CounterTargets, PendingDelta};
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, TypedWriteHook, WriteHook};
use crate::unique::{ClaimSink, UniqueTargets};
#[cfg(all(feature = "typed-tree", feature = "armour"))]
use crate::{EmitSink, QueueTargets};

/// Defaults: 4096 rows / 1 MiB encoded payload per `project` call.
pub const DEFAULT_MAX_PROJECTED_ROWS: usize = 4096;
pub const DEFAULT_MAX_PROJECTED_BYTES: usize = 1024 * 1024;

/// Fault injection for the Var reconcile scan — see
/// [`Db::debug_fail_reconcile_value_read`](crate::armour::Db::debug_fail_reconcile_value_read).
#[cfg(any(test, feature = "test-faults"))]
pub(crate) static FAIL_RECONCILE_VALUE_READ: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// Simulate an unreadable stored value in a Var reconcile scan.
///
/// A real I/O failure there is not reachable from a test — a truncated data file
/// is caught by hint validation and never leaves a dangling location — but the
/// branch it feeds is a data-loss path: the scan drives `delete_orphans`, so
/// treating an unreadable row as absent deletes its live index entries. Compiles
/// away entirely without `test-faults` (a non-default feature).
#[inline(always)]
pub(crate) fn reconcile_value_read_fault() -> DbResult<()> {
    #[cfg(any(test, feature = "test-faults"))]
    if FAIL_RECONCILE_VALUE_READ.load(std::sync::atomic::Ordering::SeqCst) {
        return Err(DbError::Internal("injected reconcile value-read failure"));
    }
    Ok(())
}

/// Fault injection for `Collection::flush` (T9 Task 7) — lets a test simulate
/// an unclean session (a collection flush that fails during `Db::close`)
/// without `mem::forget`-ing a live `Db`, which would leak `db.lock` and break
/// the reopen (`Db::drop` is a clean shutdown).
#[cfg(any(test, feature = "test-faults"))]
pub(crate) static FAIL_COLLECTION_FLUSH: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// Simulate a failed `Collection::flush` on every typed family. Compiles away
/// entirely without `test-faults` (a non-default feature).
#[inline(always)]
pub(crate) fn collection_flush_fault() -> DbResult<()> {
    #[cfg(any(test, feature = "test-faults"))]
    if FAIL_COLLECTION_FLUSH.load(std::sync::atomic::Ordering::SeqCst) {
        return Err(DbError::Internal("injected collection flush failure"));
    }
    Ok(())
}

pub(crate) type ValueToBytesFn = dyn Fn(&dyn Any) -> Option<Vec<u8>> + Send + Sync;

#[derive(Clone)]
pub(crate) struct ErasedTarget {
    pub(crate) name: &'static str,
    pub(crate) inner: Arc<dyn ProjectableIndex>,
    participant: Arc<dyn crate::txn::participant::TxParticipant>,
    pub(crate) value_to_bytes: Arc<ValueToBytesFn>,
}

/// Static participant set for one source declaration.
#[derive(Clone)]
pub struct IndexTargets {
    handles: Vec<ErasedTarget>,
}

impl Default for IndexTargets {
    fn default() -> Self {
        Self::new()
    }
}

impl IndexTargets {
    pub fn new() -> Self {
        Self {
            handles: Vec::new(),
        }
    }

    #[allow(private_bounds)]
    pub fn with_index<I>(self, index: &Arc<I>) -> Self
    where
        I: ProjectableIndexTx + IndexKeyed,
    {
        let name = index.collection_name();
        let participant = Arc::clone(index).as_tx_participant();
        let value_to_bytes = {
            let idx = Arc::clone(index);
            Arc::new(move |v: &dyn Any| -> Option<Vec<u8>> {
                v.downcast_ref::<I::V>()
                    .and_then(|v| idx.encode_value(v).ok())
            })
        };
        Self {
            handles: {
                let mut handles = self.handles;
                handles.push(ErasedTarget {
                    name,
                    inner: Arc::clone(index) as Arc<dyn ProjectableIndex>,
                    participant,
                    value_to_bytes,
                });
                handles
            },
        }
    }

    pub(crate) fn erased_targets(&self) -> &[ErasedTarget] {
        &self.handles
    }

    pub fn len(&self) -> usize {
        self.handles.len()
    }

    pub fn is_empty(&self) -> bool {
        self.handles.is_empty()
    }

    pub(crate) fn contains(&self, name: &str) -> bool {
        self.handles.iter().any(|h| h.name == name)
    }

    pub(crate) fn resolve_target(&self, name: &str) -> Option<Arc<dyn ProjectableIndex>> {
        self.handles
            .iter()
            .find(|h| h.name == name)
            .map(|h| Arc::clone(&h.inner))
    }

    /// Resolve a declared index target to its transaction participant (T5 admission).
    pub(crate) fn resolve_tx_participant(
        &self,
        name: &str,
    ) -> Option<Arc<dyn crate::txn::participant::TxParticipant>> {
        self.handles
            .iter()
            .find(|h| h.name == name)
            .map(|h| Arc::clone(&h.participant))
    }
}

/// One row in a deferred projection plan.
#[derive(Debug, Clone)]
pub enum ProjectionOp {
    Put(Vec<u8>),
    Delete,
}

#[derive(Clone)]
struct PendingRow {
    target: Arc<dyn ProjectableIndex>,
    key: Vec<u8>,
    op: ProjectionOp,
}

/// Encoded index rows and counter deltas computed before apply. Index rows are
/// safe to retry idempotently; counter deltas are **not** — re-applying a
/// delta that already landed double-counts. `apply_or_incomplete`'s returned
/// remainder always excludes whatever already applied, for both halves.
#[derive(Clone, Default)]
pub struct PendingProjection {
    rows: Vec<PendingRow>,
    deltas: Vec<PendingDelta>,
    /// The declared counter targets `deltas` resolve against — carried on the
    /// plan itself (not borrowed from the hook) so a plan returned inside
    /// `DbError::ProjectionIncomplete` can still be applied later, outside
    /// the collection's own scope.
    counter_targets: CounterTargets,
}

impl std::fmt::Debug for PendingProjection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PendingProjection")
            .field("rows", &self.rows.len())
            .field("deltas", &self.deltas.len())
            .finish()
    }
}

impl PendingProjection {
    /// Attach this source's counter half: the deltas already diffed by
    /// [`diff_counters`]/[`diff_counters_bytes`], plus the [`CounterTargets`]
    /// they resolve against (see the field's doc comment for why the targets
    /// travel with the plan rather than being borrowed from the hook).
    pub(crate) fn set_deltas(
        &mut self,
        deltas: Vec<PendingDelta>,
        counter_targets: CounterTargets,
    ) {
        self.deltas = deltas;
        self.counter_targets = counter_targets;
    }

    /// Apply every row and delta via `put_bytes`/`delete_bytes`/`apply_delta`
    /// only (no re-project, no re-diff) — the documented recovery entry point
    /// for a `DbError::ProjectionIncomplete` plan (see `docs/schema-metadata.md`
    /// and `docs/breaking-changes/26-08-07-armdb-multi-secondary-index.md` §6).
    ///
    /// Delegates to [`Self::apply_or_incomplete`] on a clone: rows are
    /// idempotent to retry, but a counter delta is **not** — re-running this
    /// same `apply` on the *whole* plan after a partial failure would
    /// double-count whatever already landed. Cloning means a caller who gets
    /// `DbError::ProjectionIncomplete(rest)` back can retry `rest.apply(&db)`
    /// (or `rest` again) and it will only touch what's actually still
    /// outstanding — the remainder contract holds all the way out to this
    /// public entry point, not just internally.
    #[cfg(feature = "armour")]
    pub fn apply(&self, db: &crate::armour::Db) -> DbResult<()> {
        let _ = db;
        self.clone().apply_or_incomplete()
    }

    /// Apply rows, then deltas — in that fixed order, matching `diff`'s own
    /// row-then-delta plan construction. On failure, the returned remainder
    /// excludes whatever already landed: index rows are idempotent to retry,
    /// but a counter delta re-applied would double-count, so both halves must
    /// report an accurate "what's left" plan, not just the rows.
    pub(crate) fn apply_or_incomplete(mut self) -> DbResult<()> {
        for (applied, row) in self.rows.iter().enumerate() {
            let res = match &row.op {
                ProjectionOp::Put(value) => row.target.put_bytes(&row.key, value),
                ProjectionOp::Delete => row.target.delete_bytes(&row.key),
            };
            if res.is_err() {
                self.rows.drain(..applied);
                return Err(DbError::ProjectionIncomplete(self));
            }
        }
        self.rows.clear();

        for (applied, d) in self.deltas.iter().enumerate() {
            // `None` is reserved: today the `OnMissingTarget::Skip` decision
            // lives inside `resolve`, and a non-`one_of` group never probes
            // `contains`, so this arm is unreachable — kept a no-op rather
            // than an `unwrap`.
            let target = match self.counter_targets.resolve_group(d.group, &d.key) {
                Ok(Some(target)) => target,
                Ok(None) => continue,
                Err(e) => {
                    self.deltas.drain(..applied);
                    tracing::error!(error = %e, "counter delta target resolution failed");
                    return Err(DbError::ProjectionIncomplete(self));
                }
            };
            if let Err(e) = target.apply_delta(&d.key, d.delta) {
                self.deltas.drain(..applied);
                tracing::error!(error = %e, "counter delta failed");
                return Err(DbError::ProjectionIncomplete(self));
            }
        }
        Ok(())
    }

    /// Append another plan's rows and deltas (atomic/TTL multi-event
    /// aggregation). Every merged plan in one aggregation comes from the same
    /// source collection (`replay_owned_events` calls `family.
    /// apply_projection_for_owned` once per event on the same `family`), so
    /// `counter_targets` is identical across them — take the incoming one
    /// whenever it is non-empty rather than requiring the caller to special-case
    /// the first merge.
    pub(crate) fn extend(&mut self, other: PendingProjection) {
        self.rows.extend(other.rows);
        self.deltas.extend(other.deltas);
        if !other.counter_targets.is_empty() {
            self.counter_targets = other.counter_targets;
        }
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.rows.is_empty() && self.deltas.is_empty()
    }

    /// Convert a deferred plan into transaction derived writes (T5 two-round commit).
    pub(crate) fn into_tx_derived(
        self,
        targets: &IndexTargets,
    ) -> DbResult<Vec<crate::txn::participant::TxDerivedWrite>> {
        let mut out = Vec::with_capacity(self.rows.len());
        for row in self.rows {
            let name = row.target.collection_name();
            let participant = targets
                .resolve_tx_participant(name)
                .ok_or(DbError::UndeclaredIndexTarget { index: name })?;
            out.push(crate::txn::participant::TxDerivedWrite {
                participant,
                key: row.key,
                value: match row.op {
                    ProjectionOp::Put(v) => Some(v),
                    ProjectionOp::Delete => None,
                },
            });
        }
        Ok(out)
    }

    pub(crate) fn row_count(&self) -> usize {
        self.rows.len()
    }

    pub(crate) fn delta_count(&self) -> usize {
        self.deltas.len()
    }
}
type ProjectedSet = HashMap<(&'static str, Vec<u8>), Vec<u8>>;

fn project_snapshot<K: Key, T, H: TypedWriteHook<K, T>>(
    hook: &H,
    targets: &IndexTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    snapshot: Option<&T>,
) -> DbResult<ProjectedSet> {
    let mut set = ProjectedSet::new();
    if let Some(value) = snapshot {
        let mut sink = ProjectionSink::new(targets, max_rows, max_bytes);
        hook.project(key, value, &mut sink)?;
        for row in sink.rows() {
            set.insert((row.target, row.key.clone()), row.value.clone());
        }
    }
    Ok(set)
}

/// Shared diff used by facade, `atomic()`, TTL, and (later) transactions.
pub(crate) fn diff_projection<K: Key, T, H: TypedWriteHook<K, T>>(
    hook: &H,
    targets: &IndexTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&T>,
    new: Option<&T>,
) -> DbResult<PendingProjection> {
    if !H::PROJECTS {
        return Ok(PendingProjection::default());
    }
    let set_a = project_snapshot(hook, targets, max_rows, max_bytes, key, old)?;
    let set_b = project_snapshot(hook, targets, max_rows, max_bytes, key, new)?;

    let mut rows = Vec::new();

    for (target_name, key_bytes) in set_a.keys() {
        if !set_b.contains_key(&(*target_name, key_bytes.clone())) {
            let target = targets
                .resolve_target(target_name)
                .ok_or(DbError::UndeclaredIndexTarget { index: target_name })?;
            rows.push(PendingRow {
                target,
                key: key_bytes.clone(),
                op: ProjectionOp::Delete,
            });
        }
    }

    for ((target_name, key_bytes), new_value) in &set_b {
        let needs_put = match set_a.get(&(*target_name, key_bytes.clone())) {
            None => true,
            Some(old_value) => {
                let target = targets
                    .resolve_target(target_name)
                    .ok_or(DbError::UndeclaredIndexTarget { index: target_name })?;
                !target.values_equal(old_value, new_value)?
            }
        };
        if needs_put {
            let target = targets
                .resolve_target(target_name)
                .ok_or(DbError::UndeclaredIndexTarget { index: target_name })?;
            rows.push(PendingRow {
                target,
                key: key_bytes.clone(),
                op: ProjectionOp::Put(new_value.clone()),
            });
        }
    }

    Ok(PendingProjection {
        rows,
        ..PendingProjection::default()
    })
}

fn project_snapshot_bytes<K: Key, H: WriteHook<K>>(
    hook: &H,
    targets: &IndexTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    snapshot: Option<&[u8]>,
) -> DbResult<ProjectedSet> {
    let mut set = ProjectedSet::new();
    if let Some(value) = snapshot {
        let mut sink = ProjectionSink::new(targets, max_rows, max_bytes);
        hook.project(key, None, Some(value), &mut sink)?;
        for row in sink.rows() {
            set.insert((row.target, row.key.clone()), row.value.clone());
        }
    }
    Ok(set)
}

/// Rebuild index rows for one projecting source from a full scan (migration reconciliation).
///
/// Adds missing rows and rewrites mismatches. When `delete_orphans` is true, removes
/// index keys not emitted by `build_expected` for each declared target.
///
/// **Open-world hazard:** migration and post-migration open paths pass
/// `delete_orphans: true`, so each declared target is treated as complete for **this
/// source only**. On a shared index, reconciling one `Projected` source while another
/// producer still maintains rows will delete that producer's index entries. Safe only
/// when every producer of each target is already on `Projected`, or use
/// [`schema::repair`](crate::schema::repair) with
/// [`SchemaRegistry::complete_index`](crate::schema::SchemaRegistry::complete_index) and
/// explicit [`RepairOptions::delete_orphans`](crate::schema::RepairOptions::delete_orphans)
/// for phased adoption. See `docs/breaking-changes/26-08-07-armdb-multi-secondary-index.md`
/// §6.
///
/// **Never reaches a unique namespace (T10).** This function only ever walks
/// `targets: &IndexTargets`, and a [`crate::unique::UniqueIndex`] is
/// deliberately not a [`ProjectableIndex`] — `IndexTargets::with_index`
/// cannot even accept one — so a source carrying both an index half
/// (`Projected::new`) and a claim half (`Projected::and_unique`) only ever
/// has its index half reconciled here; the claim half is invisible to this
/// function by construction. That matters because this function's
/// `delete_orphans: true` open-world hazard (previous paragraph) would be
/// outright destructive on a namespace fed by more than one source — see
/// `schema::unique_repair`'s module doc for the dedicated pass a namespace
/// gets instead, and `carrier_reconcile_never_touches_a_namespace` in
/// `armdb/tests/unique_validate_repair.rs` for the regression test pinning
/// this exclusion end to end.
pub(crate) fn reconcile_carrier_projections(
    targets: &IndexTargets,
    mut build_expected: impl FnMut(&mut dyn FnMut(&str, &[u8], &[u8]) -> DbResult<()>) -> DbResult<()>,
    delete_orphans: bool,
) -> DbResult<()> {
    use std::collections::HashMap;

    let mut expected: HashMap<(String, Vec<u8>), Vec<u8>> = HashMap::new();
    build_expected(&mut |target, key, value| {
        expected.insert((target.to_string(), key.to_vec()), value.to_vec());
        Ok(())
    })?;

    for handle in targets.erased_targets() {
        let target_name = handle.name;
        let mut actual: HashMap<Vec<u8>, Vec<u8>> = HashMap::new();
        handle.inner.for_each_bytes(&mut |k, v| {
            actual.insert(k.to_vec(), v.to_vec());
            Ok(())
        })?;

        for ((t, k), exp_v) in &expected {
            if t != target_name {
                continue;
            }
            match actual.get(k) {
                None => handle.inner.put_bytes(k, exp_v)?,
                Some(act_v) if !handle.inner.values_equal(act_v, exp_v)? => {
                    handle.inner.put_bytes(k, exp_v)?;
                }
                _ => {}
            }
        }

        if delete_orphans {
            for k in actual.keys() {
                if !expected.contains_key(&(target_name.to_string(), k.clone())) {
                    handle.inner.delete_bytes(k)?;
                }
            }
        }
        // Targets may buffer Bitcask writes; flush before the caller's
        // `projections_dirty` clear so a crash cannot lose repaired rows.
        handle.inner.durable_flush()?;
    }
    Ok(())
}

/// Shared diff for Const/Zero/Var families (`WriteHook` byte snapshots).
pub(crate) fn diff_projection_bytes<K: Key, H: WriteHook<K>>(
    hook: &H,
    targets: &IndexTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&[u8]>,
    new: Option<&[u8]>,
) -> DbResult<PendingProjection> {
    if !H::PROJECTS {
        return Ok(PendingProjection::default());
    }
    let set_a = project_snapshot_bytes(hook, targets, max_rows, max_bytes, key, old)?;
    let set_b = project_snapshot_bytes(hook, targets, max_rows, max_bytes, key, new)?;

    let mut rows = Vec::new();

    for (target_name, key_bytes) in set_a.keys() {
        if !set_b.contains_key(&(*target_name, key_bytes.clone())) {
            let target = targets
                .resolve_target(target_name)
                .ok_or(DbError::UndeclaredIndexTarget { index: target_name })?;
            rows.push(PendingRow {
                target,
                key: key_bytes.clone(),
                op: ProjectionOp::Delete,
            });
        }
    }

    for ((target_name, key_bytes), new_value) in &set_b {
        let needs_put = match set_a.get(&(*target_name, key_bytes.clone())) {
            None => true,
            Some(old_value) => {
                let target = targets
                    .resolve_target(target_name)
                    .ok_or(DbError::UndeclaredIndexTarget { index: target_name })?;
                !target.values_equal(old_value, new_value)?
            }
        };
        if needs_put {
            let target = targets
                .resolve_target(target_name)
                .ok_or(DbError::UndeclaredIndexTarget { index: target_name })?;
            rows.push(PendingRow {
                target,
                key: key_bytes.clone(),
                op: ProjectionOp::Put(new_value.clone()),
            });
        }
    }

    Ok(PendingProjection {
        rows,
        ..PendingProjection::default()
    })
}

/// `atomic()` teardown: projection → sync (optional) → hook replay.
pub(crate) fn finish_atomic_teardown<F, R>(
    family: &F,
    shard_id: usize,
    result: DbResult<R>,
    needs_sync: bool,
    events: &[F::Owned],
) -> DbResult<R>
where
    F: crate::ctx::Family,
{
    if let Err(e) = replay_owned_events(family, events, needs_sync, shard_id) {
        return match result {
            // `atomic()` has no rollback: the writes the closure made before it
            // errored have landed, so their projections still matter. Only one
            // error can be returned, and the caller asked for the closure's —
            // log the teardown failure rather than dropping it silently.
            Err(op) => {
                tracing::error!(
                    teardown = %e,
                    operation = %op,
                    "projection/sync failed during atomic teardown but the closure's own \
                     error is returned; the index may have drifted — verify with \
                     `schema::validate`"
                );
                Err(op)
            }
            Ok(_) => Err(e),
        };
    }
    result
}

/// Replay buffered `atomic()` / TTL events: projection → sync → hooks.
pub(crate) fn replay_owned_events<F: crate::ctx::Family>(
    family: &F,
    events: &[F::Owned],
    needs_sync: bool,
    shard_id: usize,
) -> DbResult<()> {
    // Merge every `ProjectionIncomplete` plan so a multi-write `atomic()` /
    // TTL batch can be re-applied in full — applying only the first event's plan
    // would leave the rest of the index drifted, since the source mutations are
    // already committed by the time projections run.
    let mut incomplete = PendingProjection::default();
    let mut hard_err = None;
    for ev in events {
        match family.apply_projection_for_owned(ev) {
            Ok(()) => {}
            Err(DbError::ProjectionIncomplete(plan)) => incomplete.extend(plan),
            Err(e) => {
                if hard_err.is_none() {
                    hard_err = Some(e);
                }
            }
        }
    }
    // A hard failure (budget exceeded, undeclared target) produced *no* plan, so
    // its event's rows are unrecoverable from anything we could return. It has
    // to outrank the plan: `ProjectionIncomplete` promises that applying the
    // plan restores consistency, and here it would not.
    let proj_err = match (hard_err, incomplete.is_empty()) {
        (Some(e), true) => Some(e),
        (Some(e), false) => {
            tracing::error!(
                error = %e,
                dropped_rows = incomplete.row_count(),
                dropped_deltas = incomplete.delta_count(),
                "projection failed with no recoverable plan; discarding the pending \
                 rows/deltas of the other events in this batch — repair the index with \
                 `schema::repair`"
            );
            Some(e)
        }
        (None, false) => Some(DbError::ProjectionIncomplete(incomplete)),
        (None, true) => None,
    };
    let mut sync_err = None;
    if needs_sync && let Err(e) = family.sync(shard_id) {
        sync_err = Some(e);
    }
    for ev in events {
        family.fire_owned(ev);
    }
    if let (Some(sync_e), Some(proj_e)) = (&sync_err, &proj_err) {
        tracing::error!(
            sync = %sync_e,
            projection = %proj_e,
            "sync error outranked projection error on atomic/TTL teardown"
        );
    }
    sync_err.or(proj_err).map(Err).unwrap_or(Ok(()))
}

pub(crate) mod sealed {
    /// Sealed: only this crate's collection families (and the `#[doc(hidden)]`
    /// test wrappers) can be projection targets. `IndexTargets::with_index`
    /// requires the crate-private `ProjectableIndexTx`, so an outside impl was
    /// never usable anyway — sealing makes that explicit and keeps adding a
    /// method from being a breaking change.
    pub trait Sealed {}
}

/// Crate-public seam every index family implements (Bitcask + Fixed Zero*, Typed*, VarTyped*).
/// Lives in core so `Projected` does not depend on `feature = "schema"`.
pub trait ProjectableIndex: sealed::Sealed + Send + Sync + 'static {
    fn collection_name(&self) -> &'static str;
    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()>;
    fn delete_bytes(&self, key: &[u8]) -> DbResult<()>;
    /// bytes-equal → Ok(true); else decode both and `PartialEq`.
    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool>;
    /// Full scan of live index rows for reconciliation and validation.
    #[allow(clippy::type_complexity)]
    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()>;
    /// Make everything written to this collection durable — write buffers plus
    /// fsync on Bitcask, `fdatasync` on FixedStore.
    ///
    /// Named to match [`Collection::durable_flush`](crate::armour::Collection::durable_flush),
    /// the other public durability barrier in the crate.
    fn durable_flush(&self) -> DbResult<()>;
}

/// Transaction participant view of a [`ProjectableIndex`] (crate-internal, T5).
#[allow(private_interfaces)]
pub(crate) trait ProjectableIndexTx: ProjectableIndex {
    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant>;
}

/// Associated key/value types for typed `row`/`key` helpers.
pub trait IndexKeyed: ProjectableIndex {
    type K: Key;
    type V: Clone + PartialEq;
    fn encode_key(k: &Self::K) -> Vec<u8> {
        k.as_bytes().to_vec()
    }
    fn encode_value(&self, v: &Self::V) -> DbResult<Vec<u8>>;
}

/// One projected row buffered inside a sink (applied in Task 2).
#[derive(Debug, Clone)]
pub(crate) struct ProjectedRow {
    pub(crate) target: &'static str,
    pub(crate) key: Vec<u8>,
    pub(crate) value: Vec<u8>,
}

/// Buffer + budgets + declared target table for one `project` evaluation.
pub struct ProjectionSink<'a> {
    targets: &'a IndexTargets,
    max_rows: usize,
    max_bytes: usize,
    row_count: usize,
    byte_count: usize,
    rows: Vec<ProjectedRow>,
}

impl ProjectionSink<'_> {
    pub(crate) fn new(
        targets: &IndexTargets,
        max_rows: usize,
        max_bytes: usize,
    ) -> ProjectionSink<'_> {
        ProjectionSink {
            targets,
            max_rows,
            max_bytes,
            row_count: 0,
            byte_count: 0,
            rows: Vec::new(),
        }
    }

    pub(crate) fn rows(&self) -> &[ProjectedRow] {
        &self.rows
    }

    pub fn row<I>(&mut self, index: &Arc<I>, key: I::K, value: I::V) -> DbResult<()>
    where
        I: IndexKeyed,
    {
        self.check_declared(index)?;
        let key_bytes = I::encode_key(&key);
        let value_bytes = index.encode_value(&value)?;
        self.record(index.collection_name(), &key_bytes, &value_bytes)?;
        Ok(())
    }

    pub fn key<I>(&mut self, index: &Arc<I>, key: I::K) -> DbResult<()>
    where
        I: IndexKeyed,
        I::V: Default,
    {
        self.row(index, key, I::V::default())
    }

    fn check_declared<I: ProjectableIndex>(&self, index: &Arc<I>) -> DbResult<()> {
        let name = index.collection_name();
        if self.targets.contains(name) {
            if self.targets.resolve_tx_participant(name).is_none() {
                return Err(DbError::Internal(
                    "declared index target missing transaction participant",
                ));
            }
            Ok(())
        } else {
            Err(DbError::UndeclaredIndexTarget { index: name })
        }
    }

    fn record(&mut self, target: &'static str, key: &[u8], value: &[u8]) -> DbResult<()> {
        let added_bytes = key.len() + value.len();
        let next_rows = self.row_count + 1;
        let next_bytes = self.byte_count + added_bytes;
        if next_rows > self.max_rows || next_bytes > self.max_bytes {
            return Err(DbError::ProjectionBudgetExceeded {
                rows: next_rows,
                bytes: next_bytes,
            });
        }
        self.row_count = next_rows;
        self.byte_count = next_bytes;
        self.rows.push(ProjectedRow {
            target,
            key: key.to_vec(),
            value: value.to_vec(),
        });
        Ok(())
    }
}

type ProjectFn<K, T> = dyn Fn(&K, &T, &mut ProjectionSink<'_>) -> DbResult<()> + Send + Sync;
type CounterProjectFn<K, T> = dyn Fn(&K, &T, &mut CounterSink<'_>) -> DbResult<()> + Send + Sync;
type ClaimProjectFn<K, T> = dyn Fn(&K, &T, &mut ClaimSink<'_>) -> DbResult<()> + Send + Sync;
#[cfg(all(feature = "typed-tree", feature = "armour"))]
type StreamProjectFn<K, T> =
    dyn Fn(&K, Option<&T>, Option<&T>, &mut EmitSink<'_>) -> DbResult<()> + Send + Sync;

/// Type state for a carrier without a stream declaration.
#[doc(hidden)]
pub struct NoEvents;

/// Type state for a carrier with one stream declaration.
#[cfg(all(feature = "typed-tree", feature = "armour"))]
#[doc(hidden)]
pub struct Events<K, T> {
    targets: QueueTargets,
    project: Arc<StreamProjectFn<K, T>>,
}

#[cfg(all(feature = "typed-tree", feature = "armour"))]
#[doc(hidden)]
pub trait EventState<K, T>: Send + Sync {
    const EMITS: bool;

    fn stream_project(
        &self,
        key: &K,
        old: Option<&T>,
        new: Option<&T>,
        sink: &mut EmitSink<'_>,
    ) -> DbResult<()>;

    fn stream_targets(&self) -> Option<&QueueTargets>;
}

#[cfg(all(feature = "typed-tree", feature = "armour"))]
impl<K, T> EventState<K, T> for NoEvents {
    const EMITS: bool = false;

    fn stream_project(
        &self,
        _: &K,
        _: Option<&T>,
        _: Option<&T>,
        _: &mut EmitSink<'_>,
    ) -> DbResult<()> {
        Ok(())
    }

    fn stream_targets(&self) -> Option<&QueueTargets> {
        None
    }
}

#[cfg(all(feature = "typed-tree", feature = "armour"))]
impl<K, T> EventState<K, T> for Events<K, T> {
    const EMITS: bool = true;

    fn stream_project(
        &self,
        key: &K,
        old: Option<&T>,
        new: Option<&T>,
        sink: &mut EmitSink<'_>,
    ) -> DbResult<()> {
        (self.project)(key, old, new, sink)
    }

    fn stream_targets(&self) -> Option<&QueueTargets> {
        Some(&self.targets)
    }
}

/// Write-hook carrier for a projecting source: maintains secondary indexes
/// (Task 8), engine-maintained counters (T9), and, since T10 Task 3,
/// unique-namespace claims — a source may legitimately need any combination
/// of the three, so this holds one target set + one projection closure per
/// kind rather than requiring nested hooks.
///
/// A collection has exactly one hook. Index projection is set up by
/// [`Projected::new`]; counters are added on top with [`Projected::and_counters`];
/// claims are added on top with [`Projected::and_unique`] — in any order. The
/// non-transactional write path (facade and `atomic()`/TTL replay) diffs the
/// counter deltas and applies them after the source's shard lock is released;
/// `db.tx()` commit diffs the same way but resolves each delta to an absolute
/// value inside the commit, so it lands in the same log record as its source.
pub struct Projected<K, T, H = NoHook, E = NoEvents> {
    targets: IndexTargets,
    project: Arc<ProjectFn<K, T>>,
    counter_targets: CounterTargets,
    counter_project: Option<Arc<CounterProjectFn<K, T>>>,
    unique_targets: UniqueTargets,
    unique_project: Option<Arc<ClaimProjectFn<K, T>>>,
    events: E,
    inner: H,
    max_rows: usize,
    max_bytes: usize,
}

/// Renamed from `Indexed` (T9): the carrier now holds a counter projection
/// alongside the index one. Kept one release for downstream — internal code
/// must use [`Projected`] directly, since a hit here fails the crate's
/// `-D warnings` clippy gate.
#[deprecated(
    since = "0.10.0",
    note = "renamed to `Projected` — it carries counters too"
)]
pub type Indexed<K, T, H = NoHook> = Projected<K, T, H>;

impl<K, T> Projected<K, T, NoHook> {
    pub fn new(
        targets: IndexTargets,
        project: impl Fn(&K, &T, &mut ProjectionSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    ) -> Self {
        Self {
            targets,
            project: Arc::new(project),
            counter_targets: CounterTargets::default(),
            counter_project: None,
            unique_targets: UniqueTargets::default(),
            unique_project: None,
            events: NoEvents,
            inner: NoHook,
            max_rows: DEFAULT_MAX_PROJECTED_ROWS,
            max_bytes: DEFAULT_MAX_PROJECTED_BYTES,
        }
    }
}

impl<K, T, H, E> Projected<K, T, H, E> {
    /// Nest a user hook; projection stays on the outer carrier.
    pub fn with<H2>(self, hook: H2) -> Projected<K, T, H2, E> {
        Projected {
            targets: self.targets,
            project: self.project,
            counter_targets: self.counter_targets,
            counter_project: self.counter_project,
            unique_targets: self.unique_targets,
            unique_project: self.unique_project,
            events: self.events,
            inner: hook,
            max_rows: self.max_rows,
            max_bytes: self.max_bytes,
        }
    }

    /// Override this carrier's per-`project` row/byte budgets (defaults: 4096
    /// rows / 1 MiB).
    ///
    /// This is only one half of the budget: the collection clamps it with its
    /// own `Config::max_projected_*`, so the effective limit is the **minimum**
    /// of the two and raising this alone cannot widen it. The schema validator
    /// reads that resolved minimum back off the open collection, so `validate`
    /// and `repair` always run the same budget as writes.
    #[must_use]
    pub fn with_limits(self, max_rows: usize, max_bytes: usize) -> Self {
        Self {
            max_rows,
            max_bytes,
            ..self
        }
    }

    pub fn targets(&self) -> &IndexTargets {
        &self.targets
    }

    /// Add a counter projection on top of this carrier's index projection.
    /// `targets` declares which counter target(s) `project` may contribute to
    /// (via [`CounterSink::add`] / [`CounterSink::add_one_of`]). The
    /// non-transactional write path diffs and applies the resulting deltas
    /// after the source's shard lock is released; inside `db.tx()` the same
    /// deltas are resolved to absolute values during commit and logged with
    /// their source.
    #[must_use]
    pub fn and_counters(
        self,
        targets: CounterTargets,
        project: impl Fn(&K, &T, &mut CounterSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    ) -> Self {
        Self {
            counter_targets: targets,
            counter_project: Some(Arc::new(project)),
            ..self
        }
    }

    /// Declared counter targets, or `None` when this carrier has no counter
    /// half — lets `COUNTS`-guarded call sites tell "no counters declared"
    /// apart from "declared but empty" without inspecting internals.
    pub fn counter_targets(&self) -> Option<&CounterTargets> {
        (!self.counter_targets.is_empty()).then_some(&self.counter_targets)
    }

    /// Add a claim projection on top of this carrier's index/counter halves
    /// (T10 Task 3). `targets` declares which unique namespace(s) `project`
    /// may claim into (via [`ClaimSink::claim`]). Composes in any order with
    /// [`Self::and_counters`] — the carrier just holds one more target set +
    /// closure pair.
    ///
    /// Panics if a claim half is already attached (review round M1): the spec
    /// says declaring the same half twice on one source panics at
    /// declaration, and unlike the index/counter halves this one is not
    /// merely lossy — [`SchemaRegistry::unique_index`]/`unique_index_on`
    /// (`feature = "schema"`) each push their own `Decl::Unique` into the
    /// registry regardless of how many times this method itself is called, so
    /// a silent overwrite here would leave **two** `Decl::Unique` entries
    /// live while the carrier only ever runs the second call's closure. The
    /// first declaration's namespace would then never see a claim from this
    /// source at write time, yet `schema::validate`/`repair` still expect
    /// one — `repair` would happily *write* rows into a namespace nothing on
    /// the write path will ever release. `and_counters` has the same
    /// structural gap; it stays out of scope here (T9's overwrite is a
    /// data-loss bug, not a permanently-stuck row in an engine-owned,
    /// no-public-write-API collection).
    #[must_use]
    pub fn and_unique(
        self,
        targets: UniqueTargets,
        project: impl Fn(&K, &T, &mut ClaimSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    ) -> Self {
        assert!(
            self.unique_project.is_none(),
            "and_unique: a claim half is already attached to this carrier — declaring the same \
             half twice on one source is rejected rather than silently keeping only the second \
             closure (see this method's own doc for why that would be unrepairable, not merely \
             lossy)"
        );
        // T10 review F2: an empty `UniqueTargets` must not silently disable
        // uniqueness. `claim_targets()` reads "declared but empty" as "no
        // claim half at all" (`(!self.unique_targets.is_empty()).then_some(..)`,
        // mirroring `counter_targets()`), so every guard and every
        // transaction-planning path would treat this source as non-claiming
        // and the projection closure would simply never run — no error, no
        // finding, uniqueness silently off. `SchemaRegistry`'s three
        // `reject_*_decl` checks (`secondary_index`/`counted`/`unique_index`)
        // already assert non-empty at declaration time; this mirrors that
        // assertion on the raw carrier itself, so the registry path cannot be
        // bypassed by calling `and_unique` directly.
        assert!(
            !targets.is_empty(),
            "and_unique: UniqueTargets must declare at least one target"
        );
        Self {
            unique_targets: targets,
            unique_project: Some(Arc::new(project)),
            ..self
        }
    }

    /// Declared unique-namespace targets, or `None` when this carrier has no
    /// unique half — lets `CLAIMS`-guarded call sites tell "no claims
    /// declared" apart from "declared but empty" without inspecting
    /// internals (mirrors [`Self::counter_targets`]).
    pub fn claim_targets(&self) -> Option<&UniqueTargets> {
        (!self.unique_targets.is_empty()).then_some(&self.unique_targets)
    }
}

#[cfg(all(feature = "typed-tree", feature = "armour"))]
impl<K, T, H> Projected<K, T, H, NoEvents> {
    /// Add the carrier's one and only change-stream projection half. The
    /// resulting `Events` type state intentionally has no `and_events` method,
    /// so duplicate declarations cannot be expressed.
    #[must_use]
    pub fn and_events(
        self,
        targets: QueueTargets,
        project: impl Fn(&K, Option<&T>, Option<&T>, &mut EmitSink<'_>) -> DbResult<()>
        + Send
        + Sync
        + 'static,
    ) -> Projected<K, T, H, Events<K, T>> {
        assert!(
            !targets.is_empty(),
            "and_events: QueueTargets must declare at least one target"
        );
        Projected {
            targets: self.targets,
            project: self.project,
            counter_targets: self.counter_targets,
            counter_project: self.counter_project,
            unique_targets: self.unique_targets,
            unique_project: self.unique_project,
            events: Events {
                targets,
                project: Arc::new(project),
            },
            inner: self.inner,
            max_rows: self.max_rows,
            max_bytes: self.max_bytes,
        }
    }
}

#[cfg(all(feature = "typed-tree", feature = "armour"))]
impl<K, T, H, E: EventState<K, T>> Projected<K, T, H, E> {
    /// Declared stream destinations, or `None` when no stream half exists.
    pub fn stream_targets(&self) -> Option<&QueueTargets> {
        self.events.stream_targets()
    }
}

#[cfg(all(feature = "typed-tree", feature = "armour"))]
impl<K: Key, T, H: TypedWriteHook<K, T>, E: EventState<K, T>> TypedWriteHook<K, T>
    for Projected<K, T, H, E>
{
    const WATCHES: bool = H::WATCHES;
    const PROJECTS: bool = true;
    const NEEDS_OLD_VALUE: bool = true;
    const NEEDS_WRITE: bool = true;
    const NEEDS_INIT: bool = H::NEEDS_INIT;
    const EXPIRES: bool = H::EXPIRES;
    const TTL_REVISION: u32 = H::TTL_REVISION;
    const COUNTS: bool = true;
    const CLAIMS: bool = true;
    const EMITS: bool = E::EMITS;

    fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        self.inner.watch_channel()
    }
    fn watch_old(&self) -> bool {
        self.inner.watch_old()
    }

    fn deadline(&self, key: &K, value: &T) -> Option<u64> {
        self.inner.deadline(key, value)
    }

    fn on_write(&self, key: &K, old: Option<&T>, new: Option<&T>) {
        self.inner.on_write(key, old, new);
    }

    fn on_init(&self, key: &K, value: &T) {
        self.inner.on_init(key, value);
    }

    fn project(&self, key: &K, value: &T, sink: &mut ProjectionSink<'_>) -> DbResult<()> {
        (self.project)(key, value, sink)
    }

    fn projection_targets(&self) -> Option<&IndexTargets> {
        Some(&self.targets)
    }

    fn projection_limits(&self) -> (usize, usize) {
        (self.max_rows, self.max_bytes)
    }

    fn counter_project(&self, key: &K, value: &T, sink: &mut CounterSink<'_>) -> DbResult<()> {
        match &self.counter_project {
            Some(f) => f(key, value, sink),
            None => Ok(()),
        }
    }

    fn counter_targets(&self) -> Option<&CounterTargets> {
        self.counter_targets()
    }

    fn claim_project(&self, key: &K, value: &T, sink: &mut ClaimSink<'_>) -> DbResult<()> {
        match &self.unique_project {
            Some(f) => f(key, value, sink),
            None => Ok(()),
        }
    }

    fn claim_targets(&self) -> Option<&UniqueTargets> {
        self.claim_targets()
    }

    fn stream_project(
        &self,
        key: &K,
        old: Option<&T>,
        new: Option<&T>,
        sink: &mut EmitSink<'_>,
    ) -> DbResult<()> {
        self.events.stream_project(key, old, new, sink)
    }

    fn stream_targets(&self) -> Option<&QueueTargets> {
        self.stream_targets()
    }
}

#[cfg(all(feature = "typed-tree", not(feature = "armour")))]
impl<K: Key, T, H: TypedWriteHook<K, T>> TypedWriteHook<K, T> for Projected<K, T, H, NoEvents> {
    const WATCHES: bool = H::WATCHES;
    const PROJECTS: bool = true;
    const NEEDS_OLD_VALUE: bool = true;
    const NEEDS_WRITE: bool = true;
    const NEEDS_INIT: bool = H::NEEDS_INIT;
    const EXPIRES: bool = H::EXPIRES;
    const TTL_REVISION: u32 = H::TTL_REVISION;
    const COUNTS: bool = true;
    const CLAIMS: bool = true;

    fn watch_channel(&self) -> Option<&std::sync::Arc<crate::watch::WatchChannel<K>>> {
        self.inner.watch_channel()
    }
    fn watch_old(&self) -> bool {
        self.inner.watch_old()
    }

    fn deadline(&self, key: &K, value: &T) -> Option<u64> {
        self.inner.deadline(key, value)
    }

    fn on_write(&self, key: &K, old: Option<&T>, new: Option<&T>) {
        self.inner.on_write(key, old, new);
    }

    fn on_init(&self, key: &K, value: &T) {
        self.inner.on_init(key, value);
    }

    fn project(&self, key: &K, value: &T, sink: &mut ProjectionSink<'_>) -> DbResult<()> {
        (self.project)(key, value, sink)
    }

    fn projection_targets(&self) -> Option<&IndexTargets> {
        Some(&self.targets)
    }

    fn projection_limits(&self) -> (usize, usize) {
        (self.max_rows, self.max_bytes)
    }

    fn counter_project(&self, key: &K, value: &T, sink: &mut CounterSink<'_>) -> DbResult<()> {
        match &self.counter_project {
            Some(f) => f(key, value, sink),
            None => Ok(()),
        }
    }

    fn counter_targets(&self) -> Option<&CounterTargets> {
        self.counter_targets()
    }

    fn claim_project(&self, key: &K, value: &T, sink: &mut ClaimSink<'_>) -> DbResult<()> {
        match &self.unique_project {
            Some(f) => f(key, value, sink),
            None => Ok(()),
        }
    }

    fn claim_targets(&self) -> Option<&UniqueTargets> {
        self.claim_targets()
    }
}

/// Test helper: fail the first `n` `put_bytes` calls, then delegate to `inner`.
///
/// `n` matters: a helper that disarms after one put cannot make a multi-write
/// `atomic()` / TTL batch produce more than one `ProjectionIncomplete` plan, so
/// it cannot tell "kept the first plan" from "aggregated all of them".
#[doc(hidden)]
pub struct FailNPutsIndex<I> {
    inner: Arc<I>,
    remaining: std::sync::atomic::AtomicUsize,
}

impl<I> FailNPutsIndex<I> {
    pub fn new(inner: Arc<I>, n: usize) -> Self {
        Self {
            inner,
            remaining: std::sync::atomic::AtomicUsize::new(n),
        }
    }
}

impl<I: ProjectableIndex> sealed::Sealed for FailNPutsIndex<I> {}

impl<I: ProjectableIndex> ProjectableIndex for FailNPutsIndex<I> {
    fn collection_name(&self) -> &'static str {
        self.inner.collection_name()
    }

    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
        let armed = self
            .remaining
            .fetch_update(
                std::sync::atomic::Ordering::Relaxed,
                std::sync::atomic::Ordering::Relaxed,
                |n| n.checked_sub(1),
            )
            .is_ok();
        if armed {
            return Err(DbError::Internal("injected put failure"));
        }
        self.inner.put_bytes(key, value)
    }

    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
        self.inner.delete_bytes(key)
    }

    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
        self.inner.values_equal(a, b)
    }

    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
        self.inner.for_each_bytes(f)
    }

    fn durable_flush(&self) -> DbResult<()> {
        self.inner.durable_flush()
    }
}

impl<I: ProjectableIndexTx> ProjectableIndexTx for FailNPutsIndex<I> {
    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
        Arc::clone(&self.inner).as_tx_participant()
    }
}

impl<I: IndexKeyed> IndexKeyed for FailNPutsIndex<I> {
    type K = I::K;
    type V = I::V;

    fn encode_value(&self, v: &Self::V) -> DbResult<Vec<u8>> {
        self.inner.encode_value(v)
    }
}

/// Test helper: fail the first `delete_bytes`, then delegate to `inner`.
#[doc(hidden)]
pub struct FailFirstDeleteIndex<I> {
    inner: Arc<I>,
    fail: std::sync::Mutex<bool>,
}

impl<I> FailFirstDeleteIndex<I> {
    pub fn new(inner: Arc<I>) -> Self {
        Self {
            inner,
            fail: std::sync::Mutex::new(true),
        }
    }
}

impl<I: ProjectableIndex> sealed::Sealed for FailFirstDeleteIndex<I> {}

impl<I: ProjectableIndex> ProjectableIndex for FailFirstDeleteIndex<I> {
    fn collection_name(&self) -> &'static str {
        self.inner.collection_name()
    }

    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
        self.inner.put_bytes(key, value)
    }

    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
        let mut guard = self
            .fail
            .lock()
            .map_err(|_| DbError::Internal("fail lock poisoned"))?;
        if *guard {
            *guard = false;
            return Err(DbError::Internal("injected delete failure"));
        }
        self.inner.delete_bytes(key)
    }

    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
        self.inner.values_equal(a, b)
    }

    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
        self.inner.for_each_bytes(f)
    }

    fn durable_flush(&self) -> DbResult<()> {
        self.inner.durable_flush()
    }
}

impl<I: ProjectableIndexTx> ProjectableIndexTx for FailFirstDeleteIndex<I> {
    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
        Arc::clone(&self.inner).as_tx_participant()
    }
}

impl<I: IndexKeyed> IndexKeyed for FailFirstDeleteIndex<I> {
    type K = I::K;
    type V = I::V;

    fn encode_value(&self, v: &Self::V) -> DbResult<Vec<u8>> {
        self.inner.encode_value(v)
    }
}

/// Test helper: count `put_bytes` / `delete_bytes` / `durable_flush` on `inner`.
#[doc(hidden)]
pub struct CountingIndex<I> {
    pub inner: Arc<I>,
    pub puts: std::sync::atomic::AtomicUsize,
    pub deletes: std::sync::atomic::AtomicUsize,
    pub flushes: std::sync::atomic::AtomicUsize,
    /// Runs at the *start* of `durable_flush`, before delegating. Lets a test
    /// observe state that must still hold at that instant — e.g. that
    /// `projections_dirty` has not been cleared yet.
    #[allow(clippy::type_complexity)]
    pub on_flush: Option<Box<dyn Fn() + Send + Sync>>,
}

impl<I> CountingIndex<I> {
    pub fn new(inner: Arc<I>) -> Self {
        Self {
            inner,
            puts: std::sync::atomic::AtomicUsize::new(0),
            deletes: std::sync::atomic::AtomicUsize::new(0),
            flushes: std::sync::atomic::AtomicUsize::new(0),
            on_flush: None,
        }
    }

    pub fn with_on_flush(inner: Arc<I>, f: impl Fn() + Send + Sync + 'static) -> Self {
        Self {
            on_flush: Some(Box::new(f)),
            ..Self::new(inner)
        }
    }
}

impl<I: ProjectableIndex> sealed::Sealed for CountingIndex<I> {}

impl<I: ProjectableIndex> ProjectableIndex for CountingIndex<I> {
    fn collection_name(&self) -> &'static str {
        self.inner.collection_name()
    }

    fn put_bytes(&self, key: &[u8], value: &[u8]) -> DbResult<()> {
        self.puts.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.inner.put_bytes(key, value)
    }

    fn delete_bytes(&self, key: &[u8]) -> DbResult<()> {
        self.deletes
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.inner.delete_bytes(key)
    }

    fn values_equal(&self, a: &[u8], b: &[u8]) -> DbResult<bool> {
        self.inner.values_equal(a, b)
    }

    fn for_each_bytes(&self, f: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
        self.inner.for_each_bytes(f)
    }

    fn durable_flush(&self) -> DbResult<()> {
        if let Some(f) = &self.on_flush {
            f();
        }
        self.flushes
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        self.inner.durable_flush()
    }
}

impl<I: ProjectableIndexTx> ProjectableIndexTx for CountingIndex<I> {
    fn as_tx_participant(self: Arc<Self>) -> Arc<dyn crate::txn::participant::TxParticipant> {
        Arc::clone(&self.inner).as_tx_participant()
    }
}

impl<I: IndexKeyed> IndexKeyed for CountingIndex<I> {
    type K = I::K;
    type V = I::V;

    fn encode_value(&self, v: &Self::V) -> DbResult<Vec<u8>> {
        self.inner.encode_value(v)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::txn::participant::{TxBackend, TxParticipant};

    struct StubIndex;
    impl sealed::Sealed for StubIndex {}
    impl ProjectableIndex for StubIndex {
        fn collection_name(&self) -> &'static str {
            "stub_index"
        }
        fn put_bytes(&self, _: &[u8], _: &[u8]) -> DbResult<()> {
            Ok(())
        }
        fn delete_bytes(&self, _: &[u8]) -> DbResult<()> {
            Ok(())
        }
        fn values_equal(&self, _: &[u8], _: &[u8]) -> DbResult<bool> {
            Ok(true)
        }
        fn for_each_bytes(&self, _: &mut dyn FnMut(&[u8], &[u8]) -> DbResult<()>) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
    }

    struct StubParticipant;
    impl TxParticipant for StubParticipant {
        fn name(&self) -> &'static str {
            "stub_index"
        }
        fn backend(&self) -> TxBackend {
            TxBackend::Bitcask
        }
        fn schema_fingerprint(&self) -> u64 {
            0
        }
        fn encrypted(&self) -> bool {
            false
        }
        fn max_entry_bytes(&self) -> Option<usize> {
            None
        }
        fn shard_of(&self, _: &[u8]) -> DbResult<usize> {
            Ok(0)
        }
        fn lock_shards<'a>(
            &'a self,
            _: &[usize],
        ) -> DbResult<Box<dyn crate::txn::participant::TxLocked + 'a>> {
            Err(DbError::Internal("stub"))
        }
    }

    impl ProjectableIndexTx for StubIndex {
        fn as_tx_participant(self: Arc<Self>) -> Arc<dyn TxParticipant> {
            Arc::new(StubParticipant)
        }
    }

    impl IndexKeyed for StubIndex {
        type K = [u8; 4];
        type V = [u8; 1];

        fn encode_value(&self, v: &Self::V) -> DbResult<Vec<u8>> {
            Ok(v.to_vec())
        }
    }

    #[test]
    fn index_targets_resolve_tx_participant() {
        let index = Arc::new(StubIndex);
        let targets = IndexTargets::new().with_index(&index);
        let participant = targets.resolve_tx_participant("stub_index").unwrap();
        assert_eq!(participant.name(), "stub_index");
        assert!(targets.resolve_tx_participant("missing").is_none());
    }

    #[test]
    fn counter_prep_incomplete_plan_carries_only_the_remainder() {
        // Three puts, the second fails: the returned plan must hold rows 2 and 3.
        //
        // `FailNPutsIndex::new(_, 1)` fails the first `put_bytes` call made to
        // *that instance* (see its doc comment), so row 1's target must be a
        // plain, always-succeeding index — routing every row through the same
        // `FailNPutsIndex` would fail on row 1 instead and the remainder would
        // trivially equal the whole plan under both the old and new code,
        // proving nothing about the fix.
        let always_ok: Arc<dyn ProjectableIndex> = Arc::new(StubIndex);
        let failing = Arc::new(FailNPutsIndex::new(Arc::new(StubIndex), 1));
        let mut plan = PendingProjection::default();
        plan.rows.push(PendingRow {
            target: Arc::clone(&always_ok),
            key: vec![0],
            op: ProjectionOp::Put(vec![0]),
        });
        for i in 1..3u8 {
            plan.rows.push(PendingRow {
                target: Arc::clone(&failing) as Arc<dyn ProjectableIndex>,
                key: vec![i],
                op: ProjectionOp::Put(vec![i]),
            });
        }
        let err = plan.apply_or_incomplete().expect_err("second put fails");
        let DbError::ProjectionIncomplete(rest) = err else {
            panic!("wrong error");
        };
        assert_eq!(
            rest.row_count(),
            2,
            "remainder excludes the applied first row"
        );
    }

    /// Delta-half twin of the row test above (Task 6): three deltas, the
    /// second fails — the returned plan must hold deltas 2 and 3, and no rows
    /// (this plan has none). Proves the incomplete-plan remainder excludes
    /// what already landed for deltas too, not just rows — the entire point
    /// of `PendingDelta` carrying non-idempotent counter changes.
    #[test]
    fn counter_delta_incomplete_plan_carries_only_the_remainder() {
        use crate::counter::tests::StubCounterColl;
        use crate::counter::{CounterCollection, OnMissingTarget};
        use crate::upsert::Upsert;

        let ok = Arc::new(StubCounterColl::named("ok_counter", Some(0)));
        let failing = Arc::new(StubCounterColl::named("fail_counter", Some(1)));
        let targets = CounterTargets::new()
            .with_counter(
                &ok,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .with_counter(
                &failing,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                // Always refuses a non-zero result — deterministic failure
                // (mirrors `counter::tests::counter_resolve_rejects_delete_at_a_non_zero_count`).
                |_cur: Option<&u64>, _n: u64| Upsert::Delete,
            );

        let mut plan = PendingProjection::default();
        plan.set_deltas(
            vec![
                PendingDelta {
                    group: 0,
                    key: vec![1],
                    delta: 5,
                },
                PendingDelta {
                    group: 1,
                    key: vec![2],
                    delta: 5,
                },
                PendingDelta {
                    group: 1,
                    key: vec![3],
                    delta: 5,
                },
            ],
            targets,
        );

        let err = plan.apply_or_incomplete().expect_err("second delta fails");
        let DbError::ProjectionIncomplete(rest) = err else {
            panic!("wrong error");
        };
        assert_eq!(rest.row_count(), 0, "this plan never had any rows");
        assert_eq!(
            rest.delta_count(),
            2,
            "remainder excludes the applied first delta"
        );
        // Pin both halves of the invariant: not just that the remainder is
        // the right *size*, but that delta 1 actually landed (0 + 5 = 5) —
        // a remainder that instead *skipped* delta 1 would also count 2.
        let landed = ok
            .read_bytes(&[])
            .expect("read ok_counter")
            .map(|b| u64::from_le_bytes(b.try_into().expect("8-byte stub state")));
        assert_eq!(landed, Some(5), "the applied first delta actually landed");
    }

    /// Proves the fix for the review's Finding 1: `PendingProjection::apply`
    /// — the documented recovery entry point (`docs/schema-metadata.md`,
    /// `docs/breaking-changes/26-08-07-armdb-multi-secondary-index.md` §6) —
    /// must not double-count on retry. Unlike the sibling
    /// `counter_delta_incomplete_plan_carries_only_the_remainder` test (whose
    /// failing target fails *forever*, so it only proves the remainder is
    /// shaped right), this target fails exactly once — simulating a
    /// transient failure a caller retries past — so the test can also prove
    /// that re-applying the returned remainder does not re-apply the delta
    /// that already landed.
    #[test]
    fn apply_public_entry_point_does_not_double_count_on_retry() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        use crate::counter::tests::StubCounterColl;
        use crate::counter::{CounterCollection, OnMissingTarget};
        use crate::upsert::Upsert;

        fn read_state(c: &StubCounterColl) -> Option<u64> {
            c.read_bytes(&[])
                .expect("read")
                .map(|b| u64::from_le_bytes(b.try_into().expect("8-byte stub state")))
        }

        let dir = tempfile::tempdir().expect("tempdir");
        let db = crate::armour::Db::open_test(dir.path()).expect("open db");

        let ok = Arc::new(StubCounterColl::named("ok_counter", Some(0)));
        let failing = Arc::new(StubCounterColl::named("fail_counter", Some(1)));
        // Fails exactly the first `apply_delta` call routed to it, then
        // succeeds on every call after — a transient failure, not a
        // permanent one.
        let fail_budget = Arc::new(AtomicUsize::new(1));
        let targets = CounterTargets::new()
            .with_counter(
                &ok,
                "count",
                OnMissingTarget::Create,
                |v: &u64| *v,
                |_cur: Option<&u64>, n: u64| Upsert::Put(n),
            )
            .with_counter(&failing, "count", OnMissingTarget::Create, |v: &u64| *v, {
                let fail_budget = Arc::clone(&fail_budget);
                move |_cur: Option<&u64>, n: u64| {
                    let armed = fail_budget
                        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |x| x.checked_sub(1))
                        .is_ok();
                    if armed {
                        Upsert::Delete // `n` is non-zero here -> resolve_typed errors
                    } else {
                        Upsert::Put(n)
                    }
                }
            });

        let mut plan = PendingProjection::default();
        plan.set_deltas(
            vec![
                PendingDelta {
                    group: 0,
                    key: vec![1],
                    delta: 5,
                },
                PendingDelta {
                    group: 1,
                    key: vec![2],
                    delta: 5,
                },
                PendingDelta {
                    group: 1,
                    key: vec![3],
                    delta: 5,
                },
            ],
            targets,
        );

        let err = plan
            .apply(&db)
            .expect_err("delta 1 fails its first attempt");
        let DbError::ProjectionIncomplete(rest) = err else {
            panic!("wrong error");
        };
        assert_eq!(
            rest.delta_count(),
            2,
            "remainder excludes the already-applied first delta"
        );
        assert_eq!(
            read_state(&ok),
            Some(5),
            "delta 0 landed via the first apply() call"
        );
        assert_eq!(
            read_state(&failing),
            Some(1),
            "delta 1 failed, so fail_counter is untouched"
        );

        // Retry via the same public `apply` entry point a real caller would
        // use. The transient failure has cleared (fail_budget is spent), so
        // this must succeed — and, critically, must not re-apply delta 0.
        rest.apply(&db)
            .expect("retry succeeds past the transient failure");
        assert_eq!(
            read_state(&ok),
            Some(5),
            "delta 0 was NOT re-applied by the retry — no double count"
        );
        assert_eq!(
            read_state(&failing),
            Some(11), // 1 (initial) + 5 (delta 1, retried) + 5 (delta 2, first try)
            "both remaining deltas landed exactly once"
        );
    }

    /// T9 carrier test (Task 5): a `Projected` built with `.and_counters(...)`
    /// on top of `.new(...)` reports both `PROJECTS` and `COUNTS`, and exposes
    /// the declared counter targets. `Row` is a bare marker value type — the
    /// carrier itself places no bounds on `T`.
    struct Row;

    /// Mirrors `counter::tests::single_stub_targets` for the index half: one
    /// declared index target (`"stub_index"`) plus the handle to pass to
    /// `sink.key`.
    fn stub_index_targets() -> (IndexTargets, Arc<StubIndex>) {
        let idx = Arc::new(StubIndex);
        let targets = IndexTargets::new().with_index(&idx);
        (targets, idx)
    }

    #[test]
    fn counter_carrier_reports_both_projections() {
        let (idx_targets, idx) = stub_index_targets();
        let (cnt_targets, cnt) = crate::counter::tests::single_stub_targets();
        let hook = Projected::<[u8; 8], Row>::new(idx_targets, move |_k, _v, sink| {
            sink.key(&idx, [1u8; 4])
        })
        .and_counters(cnt_targets, move |_k, _v, sink| sink.add(&cnt, [1u8; 8], 1));

        assert!(<Projected<_, _> as TypedWriteHook<[u8; 8], Row>>::PROJECTS);
        assert!(<Projected<_, _> as TypedWriteHook<[u8; 8], Row>>::COUNTS);
        assert_eq!(hook.counter_targets().map(|t| t.groups().len()), Some(1));
    }

    /// T10 Task 3 review fix F1: `Projected::claim_targets()`'s "declared but
    /// empty" distinction (mirrors `counter_targets()`) had zero coverage —
    /// this pins both branches. `CLAIMS` is `true` on any `Projected`
    /// regardless of whether `.and_unique` was called (same as `COUNTS`), so
    /// the constant alone cannot distinguish "no claim half" from "claim half
    /// declared" — `claim_targets()` is what does, and this proves it reads
    /// `None` before `.and_unique()` and `Some` after, against a real
    /// `UniqueIndex` (not a stand-in struct).
    #[test]
    fn claim_carrier_reports_the_declared_but_empty_distinction() {
        let (idx_targets, idx) = stub_index_targets();
        let hook = Projected::<[u8; 8], Row>::new(idx_targets, move |_k, _v, sink| {
            sink.key(&idx, [1u8; 4])
        });
        assert!(<Projected<_, _> as TypedWriteHook<[u8; 8], Row>>::CLAIMS);
        assert!(
            hook.claim_targets().is_none(),
            "no .and_unique() call yet — declared but empty must read None"
        );

        let (_dir, _db, ns) = crate::unique_tests::open_claim_ns();
        let targets = crate::unique::UniqueTargets::new().with(&ns);
        let hook = hook.and_unique(targets, move |_k, _v: &Row, sink| {
            sink.claim(&ns, crate::unique_tests::ClaimFwdKey::new(1))
        });
        assert!(
            hook.claim_targets().is_some(),
            "declared and non-empty must read Some"
        );
    }

    /// T10 review F2: `and_unique` with an empty `UniqueTargets` must panic at
    /// declaration rather than silently build a carrier whose `claim_targets()`
    /// reads `None` forever (see `and_unique`'s own doc for why "declared but
    /// empty" and "not declared" would otherwise be indistinguishable).
    #[test]
    #[should_panic(expected = "and_unique: UniqueTargets must declare at least one target")]
    fn and_unique_rejects_an_empty_target_set() {
        let (idx_targets, idx) = stub_index_targets();
        let hook = Projected::<[u8; 8], Row>::new(idx_targets, move |_k, _v, sink| {
            sink.key(&idx, [1u8; 4])
        });
        let _ = hook.and_unique(
            crate::unique::UniqueTargets::new(),
            |_k, _v: &Row, _sink| Ok(()),
        );
    }

    /// T10 Task 3 review fix F6: `with`/`with_limits` are hand-written struct
    /// literals (not `..self` for `with`), so a future field added to
    /// `Projected` — or a copy/paste of the field list that drops one — can
    /// silently stop being carried across either call, with nothing failing
    /// to compile (both return `Self`/`Projected<K, T, H2>`, a type-correct
    /// value either way). Pins that both calls still carry the unique half.
    #[test]
    fn with_and_with_limits_carry_the_unique_half() {
        let (idx_targets, idx) = stub_index_targets();
        let (_dir, _db, ns) = crate::unique_tests::open_claim_ns();
        let targets = crate::unique::UniqueTargets::new().with(&ns);
        let ns_for_project = Arc::clone(&ns);
        let hook = Projected::<[u8; 8], Row>::new(idx_targets, move |_k, _v, sink| {
            sink.key(&idx, [1u8; 4])
        })
        .and_unique(targets, move |_k, _v: &Row, sink| {
            sink.claim(&ns_for_project, crate::unique_tests::ClaimFwdKey::new(2))
        });

        let hook = hook.with(NoHook);
        assert!(
            hook.claim_targets().is_some(),
            "`.with(...)` must carry the unique half"
        );

        let hook = hook.with_limits(10, 10_000);
        assert!(
            hook.claim_targets().is_some(),
            "`.with_limits(...)` must carry the unique half"
        );
    }
}