armdb 0.9.0

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

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

use crate::Key;
use crate::error::{DbError, DbResult};
use crate::hook::{NoHook, TypedWriteHook, WriteHook};

/// 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(())
}

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 computed before apply; safe to retry idempotently.
#[derive(Clone, Default)]
pub struct PendingProjection {
    rows: Vec<PendingRow>,
}

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())
            .finish()
    }
}

impl PendingProjection {
    /// Apply every row via `put_bytes` / `delete_bytes` only (no re-project).
    #[cfg(feature = "armour")]
    pub fn apply(&self, db: &crate::armour::Db) -> DbResult<()> {
        let _ = db;
        self.apply_rows()
    }

    pub(crate) fn apply_rows(&self) -> DbResult<()> {
        for row in &self.rows {
            match &row.op {
                ProjectionOp::Put(value) => row.target.put_bytes(&row.key, value)?,
                ProjectionOp::Delete => row.target.delete_bytes(&row.key)?,
            }
        }
        Ok(())
    }

    pub(crate) fn apply_or_incomplete(self) -> DbResult<()> {
        for row in &self.rows {
            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() {
                return Err(DbError::ProjectionIncomplete(self));
            }
        }
        Ok(())
    }

    /// Append another plan's rows (atomic/TTL multi-event aggregation).
    pub(crate) fn extend(&mut self, other: PendingProjection) {
        self.rows.extend(other.rows);
    }

    pub(crate) fn is_empty(&self) -> bool {
        self.rows.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()
    }
}
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 })
}

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 `Indexed` 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 `Indexed`, 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.
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 })
}

/// `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(),
                "projection failed with no recoverable plan; discarding the pending \
                 rows 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 `Indexed` 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;

pub struct Indexed<K, T, H = NoHook> {
    targets: IndexTargets,
    project: Arc<ProjectFn<K, T>>,
    inner: H,
    max_rows: usize,
    max_bytes: usize,
}

impl<K, T> Indexed<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),
            inner: NoHook,
            max_rows: DEFAULT_MAX_PROJECTED_ROWS,
            max_bytes: DEFAULT_MAX_PROJECTED_BYTES,
        }
    }
}

impl<K, T, H> Indexed<K, T, H> {
    /// Nest a user hook; projection stays on the outer carrier.
    pub fn with<H2>(self, hook: H2) -> Indexed<K, T, H2> {
        Indexed {
            targets: self.targets,
            project: self.project,
            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
    }
}

impl<K: Key, T, H: TypedWriteHook<K, T>> TypedWriteHook<K, T> for Indexed<K, T, H> {
    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;

    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)
    }
}

/// 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());
    }
}