use std::collections::HashMap;
use std::time::Duration;

#[cfg(feature = "indexer")]
use jacquard_common::IntoStatic;
use jacquard_common::types::did::Did;
#[cfg(feature = "indexer")]
use jacquard_common::types::string::Tid;
use miette::{IntoDiagnostic, Result};
use smol_str::SmolStr;

use crate::db::counts::{PdsAccountMove, counted_pds, pds_host};
use crate::db::outbox::{Assigned, Outbox};
#[cfg(feature = "indexer")]
use crate::db::types::{DbAction, DbRkey, DbTid};
use crate::db::{CountDeltas, Db, keys};
#[cfg(feature = "indexer")]
use crate::db::{LifecycleCountBatch, RecordCounts};
#[cfg(feature = "indexer")]
use crate::ops::record_events::{EmitOp, RecordEmitter, RecordEventOrigin};
#[cfg(feature = "indexer")]
use crate::state::AppState;
#[cfg(feature = "indexer")]
use crate::types::GaugeState;
use crate::types::RepoState;

#[cfg(feature = "firehose-diagnostics")]
type TxnInstant = std::time::Instant;

#[cfg(not(feature = "firehose-diagnostics"))]
#[derive(Clone, Copy)]
struct TxnInstant;

#[cfg(not(feature = "firehose-diagnostics"))]
impl TxnInstant {
    #[inline(always)]
    fn now() -> Self {
        Self
    }

    #[inline(always)]
    fn elapsed(&self) -> Duration {
        Duration::ZERO
    }
}

pub(crate) struct TxnCommitTimings {
    pub(crate) stage_counts: Duration,
    pub(crate) stage_and_commit: Duration,
    pub(crate) apply_counts: Duration,
}

/// one atomic database write, including its in-memory count projections.
///
/// count deltas are persisted in the same fjall batch and applied in memory only
/// after that batch commits, and so are the collection counts its record writes move.
/// dropping a transaction discards all of them.
pub(crate) struct Txn<'db> {
    pub(crate) batch: fjall::OwnedWriteBatch,
    pub(crate) db: &'db Db,
    pub(crate) counts: CountDeltas,
    /// stream events, given positions and broadcast when the batch commits
    pub(crate) outbox: Outbox,
    /// the pds host each repo row this transaction put counts an account for
    pds_accounts: HashMap<Vec<u8>, Option<SmolStr>>,
    /// heads written or removed in this transaction, so later record ops in the same batch
    /// transition counts and history from what earlier ops staged
    #[cfg(feature = "indexer")]
    heads: HashMap<Vec<u8>, Option<fjall::Slice>>,
    /// latest history death rev staged per record key in this transaction, so two deaths in
    /// the same microsecond never clobber each other and a second delete sees the first
    #[cfg(feature = "indexer")]
    last_death: HashMap<Vec<u8>, DbTid>,
    #[cfg(feature = "indexer")]
    lifecycle: Option<LifecycleCountBatch<'db>>,
    #[cfg(feature = "indexer")]
    record_counts: RecordCounts,
}

impl<'db> Txn<'db> {
    #[allow(dead_code)]
    pub(crate) fn new(db: &'db Db) -> Self {
        Self::from_parts(db, db.inner.batch(), CountDeltas::default())
    }

    pub(crate) fn from_parts(
        db: &'db Db,
        batch: fjall::OwnedWriteBatch,
        counts: CountDeltas,
    ) -> Self {
        Self {
            batch,
            db,
            counts,
            outbox: Outbox::default(),
            pds_accounts: HashMap::new(),
            #[cfg(feature = "indexer")]
            heads: HashMap::new(),
            #[cfg(feature = "indexer")]
            last_death: HashMap::new(),
            #[cfg(feature = "indexer")]
            lifecycle: None,
            #[cfg(feature = "indexer")]
            record_counts: RecordCounts::default(),
        }
    }

    #[cfg(feature = "indexer")]
    pub(crate) fn is_excluded(&self, did: &Did) -> Result<bool> {
        self.db
            .filter
            .contains_key(crate::db::filter::exclude_key(did.as_str())?)
            .into_diagnostic()
    }

    /// whether `pending_key` is queued and is the key the repo's metadata names, so not an
    /// orphan left queued next to it
    #[cfg(feature = "indexer")]
    pub(crate) fn names_pending(&self, did: &Did, pending_key: &[u8]) -> Result<bool> {
        super::lifecycle_counts::names_pending_key(self.db, did, pending_key)
    }

    #[cfg(feature = "indexer")]
    pub(crate) fn records<'txn, 'did>(
        &'txn mut self,
        state: &AppState,
        commit_rev: &DbTid,
        did: &'did Did,
    ) -> Result<RecordTxn<'txn, 'db, 'did>> {
        let excluded = self.is_excluded(did)?;
        Ok(self.record_txn(state, commit_rev, did, RecordEventOrigin::Live, excluded))
    }

    #[cfg(all(test, feature = "indexer"))]
    pub(crate) fn backfill_records<'txn, 'did>(
        &'txn mut self,
        state: &AppState,
        commit_rev: &DbTid,
        did: &'did Did,
    ) -> Result<RecordTxn<'txn, 'db, 'did>> {
        let excluded = self.is_excluded(did)?;
        Ok(self.record_txn(
            state,
            commit_rev,
            did,
            RecordEventOrigin::Backfill,
            excluded,
        ))
    }

    /// a backfill's records for one of its steps. its shard read the exclusion once, at the
    /// first step, and [`BackfillStepTxn::apply`] only accepts a [`DiffStep`] minted by
    /// `diff(new, StoredHeads)`
    #[cfg(feature = "indexer")]
    pub(crate) fn backfill_step_records<'txn, 'did>(
        &'txn mut self,
        state: &AppState,
        commit_rev: &DbTid,
        did: &'did Did,
    ) -> BackfillStepTxn<'txn, 'db, 'did> {
        BackfillStepTxn(self.record_txn(
            state,
            commit_rev,
            did,
            RecordEventOrigin::Backfill,
            false,
        ))
    }

    #[cfg(feature = "indexer")]
    fn record_txn<'txn, 'did>(
        &'txn mut self,
        state: &AppState,
        commit_rev: &DbTid,
        did: &'did Did,
        origin: RecordEventOrigin,
        excluded: bool,
    ) -> RecordTxn<'txn, 'db, 'did> {
        RecordTxn {
            txn: self,
            emitter: RecordEmitter::new(state, commit_rev, origin),
            did,
            rev: *commit_rev,
            ephemeral: state.ephemeral,
            only_index_links: state.only_index_links,
            count_ephemeral_records: matches!(origin, RecordEventOrigin::Backfill),
            excluded,
            has_redactions: None,
        }
    }

    /// reads `did`'s repo row and keeps the pds it counts for, so a [`Self::put_repo`] of it
    /// later in this transaction needn't read it again
    #[cfg(feature = "indexer")]
    pub(crate) fn read_repo(&mut self, did: &Did) -> Result<Option<RepoState<'static>>> {
        let key = keys::repo_key(did);
        let row = self
            .db
            .repos
            .get(&key)
            .into_diagnostic()?
            .map(|bytes| crate::db::deser_repo_state(&bytes).map(IntoStatic::into_static))
            .transpose()?;
        let counted = row.as_ref().and_then(counted_pds).map(SmolStr::new);
        self.pds_accounts.insert(key, counted);
        Ok(row)
    }

    /// puts `row` as `did`'s repo row. every repo row write goes through here or
    /// [`Self::remove_repo`], because the per-pds account counts move by the row a write
    /// replaces. only one thread writes a repo's row, its indexer shard (or its relay shard in a
    /// build without the indexer), so that row can't change before the commit
    pub(crate) fn put_repo(&mut self, did: &Did, row: &RepoState<'_>) -> Result<PdsAccountMove> {
        let key = keys::repo_key(did);
        let value = crate::db::ser_repo_state(row)?;
        let moved = self.move_pds_account(key.clone(), counted_pds(row))?;
        self.batch.insert(&self.db.repos, key, value);
        Ok(moved)
    }

    #[cfg(feature = "indexer")]
    pub(crate) fn remove_repo(&mut self, did: &Did) -> Result<PdsAccountMove> {
        let key = keys::repo_key(did);
        let moved = self.move_pds_account(key.clone(), None)?;
        self.batch.remove(&self.db.repos, key);
        Ok(moved)
    }

    fn move_pds_account(&mut self, key: Vec<u8>, to: Option<&str>) -> Result<PdsAccountMove> {
        let from = match self.pds_accounts.remove(&key) {
            Some(staged) => staged,
            None => self
                .db
                .repos
                .get(&key)
                .into_diagnostic()?
                .map(|bytes| {
                    crate::db::deser_repo_state(&bytes)
                        .map(|row| counted_pds(&row).map(SmolStr::new))
                })
                .transpose()?
                .flatten(),
        };
        // the same pds most writes keep, which needs no url parse
        if from.as_deref() == to {
            self.pds_accounts.insert(key, from);
            return Ok(PdsAccountMove::default());
        }
        self.pds_accounts.insert(key, to.map(SmolStr::new));
        let from = from.as_deref().and_then(pds_host);
        let to = to.and_then(pds_host);
        if from == to {
            return Ok(PdsAccountMove::default());
        }
        if let Some(host) = &from {
            self.counts.add_pds_account(host, -1);
        }
        if let Some(host) = &to {
            self.counts.add_pds_account(host, 1);
        }
        Ok(PdsAccountMove { from, to })
    }

    /// moves `did`'s count of `collection` by `delta` when this transaction commits
    #[cfg(feature = "indexer")]
    pub(crate) fn add_record_count(&mut self, did: &Did, collection: &str, delta: i64) {
        self.record_counts.add(did, collection, delta);
    }

    /// replaces `did`'s counts of the collections `matches_collection` takes with `counts`,
    /// including whatever this transaction's record writes moved them by so far
    #[cfg(feature = "indexer")]
    pub(crate) fn replace_record_counts<'a>(
        &mut self,
        did: &Did,
        matches_collection: impl Fn(&str) -> bool,
        counts: impl IntoIterator<Item = (&'a str, u64)>,
    ) -> Result<()> {
        self.record_counts
            .replace(self.db, did, matches_collection, counts)
    }

    #[cfg(feature = "indexer")]
    pub(crate) fn transition_lifecycle(&mut self, did: &Did, gauge: GaugeState) -> Result<bool> {
        let lifecycle = self
            .lifecycle
            .get_or_insert_with(|| self.db.lifecycle_counts());
        lifecycle.transition(&mut self.batch, did, gauge)
    }

    #[cfg(feature = "indexer")]
    pub(crate) fn forget_lifecycle(&mut self, did: &Did) -> Result<()> {
        let lifecycle = self
            .lifecycle
            .get_or_insert_with(|| self.db.lifecycle_counts());
        lifecycle.forget(&mut self.batch, did)
    }

    #[cfg(feature = "indexer")]
    pub(crate) fn transition_pending_key(
        &mut self,
        did: &Did,
        pending_key: &[u8],
        gauge: GaugeState,
    ) -> Result<bool> {
        let lifecycle = self
            .lifecycle
            .get_or_insert_with(|| self.db.lifecycle_counts());
        lifecycle.transition_pending_key(&mut self.batch, did, pending_key, gauge)
    }

    #[allow(dead_code)]
    pub(crate) fn commit(self) -> Result<()> {
        self.commit_with(|_| Ok(())).map(|_| ())
    }

    /// commit, returning the positions the outbox's events were given.
    #[allow(dead_code)]
    pub(crate) fn commit_assigned(self) -> Result<Assigned> {
        self.commit_inner(|_| Ok(()))
            .map(|(_, _, assigned)| assigned)
    }

    /// stages count reservations, lets the caller add commit-coupled writes,
    /// commits once, then applies the in-memory projections.
    ///
    /// the outbox's events are broadcast after the projections and before
    /// this returns. a repo's records are only written on its indexer shard,
    /// which also runs any redaction of them, so a live body can't be published
    /// after a redaction of it commits.
    #[allow(dead_code)]
    pub(crate) fn commit_with<T, F>(self, stage: F) -> Result<(T, TxnCommitTimings)>
    where
        F: FnOnce(&mut fjall::OwnedWriteBatch) -> Result<T>,
    {
        self.commit_inner(stage)
            .map(|(staged, timings, _)| (staged, timings))
    }

    fn commit_inner<T, F>(mut self, stage: F) -> Result<(T, TxnCommitTimings, Assigned)>
    where
        F: FnOnce(&mut fjall::OwnedWriteBatch) -> Result<T>,
    {
        let stage_counts_started = TxnInstant::now();
        #[cfg(feature = "indexer")]
        std::mem::take(&mut self.record_counts).stage(&mut self.batch, self.db)?;
        #[cfg(feature = "indexer")]
        let lifecycle_reservation = self
            .lifecycle
            .map(|lifecycle| lifecycle.stage(&mut self.batch));
        let count_reservation = self.db.stage_count_deltas(&mut self.batch, &self.counts);
        let stage_counts = stage_counts_started.elapsed();

        let stage_and_commit_started = TxnInstant::now();
        let staged = stage(&mut self.batch)?;
        let (db, counts) = (self.db, &self.counts);
        let mut apply_counts = Duration::ZERO;
        // before the outbox broadcasts, so a subscriber reading counts when
        // an event arrives sees it counted
        let assigned = self.outbox.commit(self.db, self.batch, || {
            let apply_counts_started = TxnInstant::now();
            db.apply_count_deltas(counts);
            drop(count_reservation);
            #[cfg(feature = "indexer")]
            if let Some(reservation) = lifecycle_reservation {
                db.apply_lifecycle_counts(reservation);
            }
            apply_counts = apply_counts_started.elapsed();
        })?;
        let stage_and_commit = stage_and_commit_started
            .elapsed()
            .saturating_sub(apply_counts);

        Ok((
            staged,
            TxnCommitTimings {
                stage_counts,
                stage_and_commit,
                apply_counts,
            },
            assigned,
        ))
    }
}

/// one commit's record mutations within a larger atomic transaction.
#[cfg(feature = "indexer")]
pub(crate) struct RecordTxn<'txn, 'db, 'did> {
    txn: &'txn mut Txn<'db>,
    emitter: RecordEmitter,
    did: &'did Did,
    /// the rev of the commit being applied; history entries are keyed by it
    rev: DbTid,
    ephemeral: bool,
    only_index_links: bool,
    count_ephemeral_records: bool,
    excluded: bool,
    /// cached once per transaction step: almost no repo has operator redactions
    has_redactions: Option<bool>,
}

/// a backfill step's record transaction on its owner shard, which only accepts a [`DiffStep`]
/// minted by `diff(new, StoredHeads)` where keys are strictly increasing and `Create` is
/// proven absent from `heads`
#[cfg(feature = "indexer")]
pub(crate) struct BackfillStepTxn<'txn, 'db, 'did>(RecordTxn<'txn, 'db, 'did>);

#[cfg(feature = "indexer")]
impl BackfillStepTxn<'_, '_, '_> {
    pub(crate) fn apply(
        &mut self,
        step: crate::backfill::worker::apply::DiffStep,
    ) -> Result<()> {
        use crate::ingest::indexer::message::RecordWrite;

        for write in step.into_writes() {
            match write {
                RecordWrite::Create {
                    collection,
                    rkey,
                    block,
                } => self.0.create_from_diff(&collection, &rkey, &block)?,
                RecordWrite::Update {
                    collection,
                    rkey,
                    block,
                } => self.0.update_from_diff(&collection, &rkey, &block)?,
                RecordWrite::Delete { collection, rkey } => {
                    self.0.delete_from_diff(&collection, &rkey)?
                }
            }
        }
        Ok(())
    }

    pub(crate) fn read_repo(&mut self) -> Result<Option<RepoState<'static>>> {
        self.0.read_repo()
    }

    pub(crate) fn update_repo_state(&mut self, state: &RepoState<'_>) -> Result<()> {
        self.0.update_repo_state(state)
    }
}

#[cfg(feature = "indexer")]
impl RecordTxn<'_, '_, '_> {
    pub(crate) fn is_excluded(&self) -> bool {
        self.excluded
    }

    /// stores a record block read from a car, hashing it first unless the parse already did.
    pub(crate) fn put_record(
        &mut self,
        collection: &str,
        rkey: &DbRkey,
        block: &crate::car::CarBlock,
        action: DbAction,
    ) -> Result<()> {
        if self.is_excluded() {
            return Ok(());
        }

        let block = block
            .verify()
            .map_err(|e| miette::miette!("record {}/{collection}/{rkey}: {e}", self.did))?;
        let cid = *block.cid();
        let block = block.bytes();

        if self.ephemeral {
            if action == DbAction::Create && self.count_ephemeral_records {
                self.txn.counts.add_records(1);
            }
            return self.emitter.emit(
                &mut self.txn.outbox,
                EmitOp {
                    did: self.did,
                    collection,
                    rkey,
                    action,
                    cid: Some(cid),
                    block: Some(block),
                },
            );
        }

        let key = keys::record_key(self.did, collection, rkey);
        let head = self.read_head(&key)?;

        // counts derive from the actual head transition, never from the
        // claimed action: replayed creates, updates of records we never
        // saw, and rewrites all count as the storage transition they
        // perform, which makes mutation replay idempotent
        let head_cid_is_current = head
            .as_ref()
            .map(|old| crate::db::ipld_cid_from_record_value(old.as_ref()))
            .transpose()?
            .is_some_and(|old_cid| old_cid == cid);
        let version_is_redacted = self.is_redacted(&key, &cid)?;
        let store_cid_marker = self.only_index_links || version_is_redacted;
        let head_is_cid_marker = head
            .as_ref()
            .is_some_and(|old| crate::db::is_cid_record_value(old));
        let storage_is_current = head_cid_is_current && head_is_cid_marker == store_cid_marker;
        let cid_bytes = cid.to_bytes();
        let value: &[u8] = if store_cid_marker {
            // to_bytes is [u8; 36]-sized for cids we produce here
            &cid_bytes
        } else {
            block.as_ref()
        };
        match head {
            _ if storage_is_current => {
                // identical replay (buffered live events vs an imported
                // car): no storage change
            }
            Some(_) if head_cid_is_current => {
                // the logical version is unchanged, but its representation is
                // not: either enforce a durable operator redaction or repair a
                // non-redaction CID placeholder once the body is available.
                self.put_head(key, value);
            }
            Some(old) => {
                if !self.only_index_links && !crate::db::is_cid_record_value(&old) {
                    // a differing head moves to history before being
                    // replaced, so replays of earlier events can still
                    // resolve it
                    self.write_history(&key, old.as_ref());
                }
                self.put_head(key, value);
            }
            None => {
                self.count_record(collection, 1);
                self.put_head(key, value);
            }
        }
        self.index_and_emit(collection, rkey, action, cid, block, version_is_redacted)
    }

    fn count_record(&mut self, collection: &str, delta: i64) {
        self.txn.counts.add_records(delta);
        self.txn.add_record_count(self.did, collection, delta);
    }

    fn is_redacted(&mut self, key: &[u8], cid: &jacquard_common::types::cid::IpldCid) -> Result<bool> {
        if self.only_index_links {
            return Ok(false);
        }
        let has_any = match self.has_redactions {
            Some(has) => has,
            None => {
                let prefix = keys::record_prefix_did(self.did);
                let has = self
                    .txn
                    .db
                    .indexer
                    .redactions
                    .prefix(&prefix)
                    .next()
                    .map(|guard| guard.key())
                    .transpose()
                    .into_diagnostic()?
                    .is_some();
                self.has_redactions = Some(has);
                has
            }
        };
        if !has_any {
            return Ok(false);
        }
        self.txn
            .db
            .indexer
            .redactions
            .contains_key(keys::redaction_key(key, cid))
            .into_diagnostic()
    }

    fn create_from_diff(
        &mut self,
        collection: &str,
        rkey: &DbRkey,
        block: &crate::car::CarBlock,
    ) -> Result<()> {
        let block = block
            .verify()
            .map_err(|e| miette::miette!("record {}/{collection}/{rkey}: {e}", self.did))?;
        let cid = *block.cid();
        let block = block.bytes();

        if self.ephemeral {
            if self.count_ephemeral_records {
                self.txn.counts.add_records(1);
            }
            return self.emitter.emit(
                &mut self.txn.outbox,
                EmitOp {
                    did: self.did,
                    collection,
                    rkey,
                    action: DbAction::Create,
                    cid: Some(cid),
                    block: Some(block),
                },
            );
        }

        let key = keys::record_key(self.did, collection, rkey);
        let version_is_redacted = self.is_redacted(&key, &cid)?;
        let cid_bytes = cid.to_bytes();
        let value: &[u8] = if self.only_index_links || version_is_redacted {
            &cid_bytes
        } else {
            block.as_ref()
        };
        self.count_record(collection, 1);
        self.txn.batch.insert(&self.txn.db.indexer.heads, key, value);
        self.index_and_emit(collection, rkey, DbAction::Create, cid, block, version_is_redacted)
    }

    fn update_from_diff(
        &mut self,
        collection: &str,
        rkey: &DbRkey,
        block: &crate::car::CarBlock,
    ) -> Result<()> {
        let block = block
            .verify()
            .map_err(|e| miette::miette!("record {}/{collection}/{rkey}: {e}", self.did))?;
        let cid = *block.cid();
        let block = block.bytes();

        if self.ephemeral {
            return self.emitter.emit(
                &mut self.txn.outbox,
                EmitOp {
                    did: self.did,
                    collection,
                    rkey,
                    action: DbAction::Update,
                    cid: Some(cid),
                    block: Some(block),
                },
            );
        }

        let key = keys::record_key(self.did, collection, rkey);
        let head = self.txn.db.indexer.head(&key).into_diagnostic()?;
        let version_is_redacted = self.is_redacted(&key, &cid)?;
        let cid_bytes = cid.to_bytes();
        let value: &[u8] = if self.only_index_links || version_is_redacted {
            &cid_bytes
        } else {
            block.as_ref()
        };
        match head {
            Some(old) => {
                if !self.only_index_links && !crate::db::is_cid_record_value(&old) {
                    self.txn.batch.insert(
                        &self.txn.db.indexer.history,
                        keys::history_key(&key, &self.death_rev(&key)),
                        old.as_ref(),
                    );
                }
            }
            None => self.count_record(collection, 1),
        }
        self.txn.batch.insert(&self.txn.db.indexer.heads, key, value);
        self.index_and_emit(collection, rkey, DbAction::Update, cid, block, version_is_redacted)
    }

    fn delete_from_diff(&mut self, collection: &str, rkey: &DbRkey) -> Result<()> {
        if self.ephemeral {
            if self.count_ephemeral_records {
                self.txn.counts.add_records(-1);
            }
            return self.emitter.emit(
                &mut self.txn.outbox,
                EmitOp {
                    did: self.did,
                    collection,
                    rkey,
                    action: DbAction::Delete,
                    cid: None,
                    block: None,
                },
            );
        }

        let key = keys::record_key(self.did, collection, rkey);
        if let Some(old) = self.txn.db.indexer.head(&key).into_diagnostic()? {
            self.count_record(collection, -1);
            if !self.only_index_links {
                let history_value: &[u8] = if crate::db::is_cid_record_value(&old) {
                    &[]
                } else {
                    old.as_ref()
                };
                self.txn.batch.insert(
                    &self.txn.db.indexer.history,
                    keys::history_key(&key, &self.death_rev(&key)),
                    history_value,
                );
            }
            self.txn.batch.remove(&self.txn.db.indexer.heads, key);
        }
        crate::ops::backlink_ops::delete_record(
            &mut self.txn.batch,
            self.txn.db,
            self.did,
            collection,
            &rkey.to_smolstr(),
        )?;
        self.emitter.emit(
            &mut self.txn.outbox,
            EmitOp {
                did: self.did,
                collection,
                rkey,
                action: DbAction::Delete,
                cid: None,
                block: None,
            },
        )
    }

    fn index_and_emit(
        &mut self,
        collection: &str,
        rkey: &DbRkey,
        action: DbAction,
        cid: jacquard_common::types::cid::IpldCid,
        block: &bytes::Bytes,
        version_is_redacted: bool,
    ) -> Result<()> {
        if version_is_redacted {
            crate::ops::backlink_ops::delete_record(
                &mut self.txn.batch,
                self.txn.db,
                self.did,
                collection,
                &rkey.to_smolstr(),
            )?;
        } else {
            crate::ops::backlink_ops::index_record(
                &mut self.txn.batch,
                self.txn.db,
                self.did,
                collection,
                &rkey.to_smolstr(),
                block,
            )?;
        }
        self.emitter.emit(
            &mut self.txn.outbox,
            EmitOp {
                did: self.did,
                collection,
                rkey,
                action,
                cid: Some(cid),
                block: (!version_is_redacted).then_some(block),
            },
        )
    }

    fn read_head(&self, key: &[u8]) -> Result<Option<fjall::Slice>> {
        if let Some(staged) = self.txn.heads.get(key) {
            return Ok(staged.clone());
        }
        self.txn.db.indexer.head(key).into_diagnostic()
    }

    fn put_head(&mut self, key: Vec<u8>, value: &[u8]) {
        let value = fjall::Slice::from(value);
        self.txn
            .batch
            .insert(&self.txn.db.indexer.heads, key.clone(), value.clone());
        self.txn.heads.insert(key, Some(value));
    }

    fn remove_head(&mut self, key: Vec<u8>) {
        self.txn.batch.remove(&self.txn.db.indexer.heads, key.clone());
        self.txn.heads.insert(key, None);
    }

    fn write_history(&mut self, key: &[u8], value: &[u8]) {
        let death = self.death_rev(key);
        self.txn.batch.insert(
            &self.txn.db.indexer.history,
            keys::history_key(key, &death),
            value,
        );
        self.txn.last_death.insert(key.to_vec(), death);
    }

    /// history death key for this commit: the later of the commit rev and
    /// the local clock, and strictly after any death already staged for `key`
    /// in this transaction. retention ages history on the same local clock as
    /// event retention, so delayed commits keep their bodies resolvable for
    /// the full configured window, and deaths always sort after the events
    /// that precede them
    fn death_rev(&self, key: &[u8]) -> DbTid {
        let rev = self.rev.max(DbTid::from(&Tid::now_0()));
        match self.txn.last_death.get(key) {
            Some(last) => rev.max(last.add_micros(1)),
            None => rev,
        }
    }

    /// true when a death entry already exists for `key` at or after this
    /// commit's rev: this delete (or a later one) has already been applied
    fn already_deleted(&self, key: &[u8]) -> Result<bool> {
        if self
            .txn
            .last_death
            .get(key)
            .is_some_and(|death| *death >= self.rev)
        {
            return Ok(true);
        }
        let lo = keys::history_key(key, &self.rev);
        let mut hi = key.to_vec();
        hi.push(1);
        Ok(self
            .txn
            .db
            .indexer
            .history
            .range(lo..hi)
            .next()
            .map(|guard| guard.into_inner())
            .transpose()
            .into_diagnostic()?
            .is_some())
    }

    pub(crate) fn delete_record(&mut self, collection: &str, rkey: &DbRkey) -> Result<()> {
        if self.is_excluded() {
            return Ok(());
        }

        if self.ephemeral {
            if self.count_ephemeral_records {
                self.txn.counts.add_records(-1);
            }
            return self.emitter.emit(
                &mut self.txn.outbox,
                EmitOp {
                    did: self.did,
                    collection,
                    rkey,
                    action: DbAction::Delete,
                    cid: None,
                    block: None,
                },
            );
        }

        let key = keys::record_key(self.did, collection, rkey);

        // like put_record, counts derive from the actual head transition:
        // deleting a missing record is not a decrement, and replaying a
        // delete is a full no-op so the originally preserved body survives
        match self.read_head(&key)? {
            Some(old) => {
                self.count_record(collection, -1);
                if !self.only_index_links {
                    // preserve the dying body so replays of earlier events
                    // can still resolve it
                    let history_value: &[u8] = if crate::db::is_cid_record_value(&old) {
                        &[]
                    } else {
                        old.as_ref()
                    };
                    self.write_history(&key, history_value);
                }
                self.remove_head(key);
            }
            None => {
                if !self.only_index_links && !self.already_deleted(&key)? {
                    // a delete of a record we never saw leaves an empty
                    // tombstone, distinguishing "deleted at rev" from
                    // "never existed"
                    self.write_history(&key, &[]);
                }
            }
        }
        crate::ops::backlink_ops::delete_record(
            &mut self.txn.batch,
            self.txn.db,
            self.did,
            collection,
            &rkey.to_smolstr(),
        )?;

        self.emitter.emit(
            &mut self.txn.outbox,
            EmitOp {
                did: self.did,
                collection,
                rkey,
                action: DbAction::Delete,
                cid: None,
                block: None,
            },
        )
    }

    /// the repo's row, read through the transaction so [`Self::update_repo_state`] doesn't
    /// read it again
    pub(crate) fn read_repo(&mut self) -> Result<Option<RepoState<'static>>> {
        self.txn.read_repo(self.did)
    }

    pub(crate) fn update_repo_state(&mut self, state: &RepoState<'_>) -> Result<()> {
        if self.is_excluded() {
            return Ok(());
        }
        self.txn.put_repo(self.did, state).map(drop)
    }
}

#[cfg(all(test, feature = "indexer"))]
mod tests {
    use super::*;
    use crate::db::types::DbAction;
    use bytes::Bytes;
    use jacquard_common::types::string::Tid;
    use std::ops::Bound;
    use tempfile::TempDir;

    const DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";

    fn test_state() -> (TempDir, AppState) {
        let tmp = tempfile::tempdir().unwrap();
        let mut config = crate::config::Config::default();
        config.database_path = tmp.path().to_path_buf();
        let state = AppState::new(&config).unwrap();
        (tmp, state)
    }

    fn rev(s: &str) -> DbTid {
        DbTid::from(&Tid::new(s).unwrap())
    }

    fn body(s: &str) -> Bytes {
        Bytes::copy_from_slice(s.as_bytes())
    }

    fn did() -> Did {
        Did::new_static(DID).unwrap()
    }

    fn write(
        state: &AppState,
        rev: &DbTid,
        collection: &str,
        rkey: &str,
        record: &Bytes,
        action: DbAction,
    ) {
        let mut txn = Txn::new(&state.db);
        let did = did();
        let mut record_txn = txn.records(state, rev, &did).unwrap();
        let block = crate::car::CarBlock::from_body(record.clone());
        record_txn
            .put_record(collection, &DbRkey::new(rkey), &block, action)
            .unwrap();
        txn.commit().unwrap();
    }

    fn delete(state: &AppState, rev: &DbTid, collection: &str, rkey: &str) {
        let mut txn = Txn::new(&state.db);
        let did = did();
        let mut record_txn = txn.records(state, rev, &did).unwrap();
        record_txn
            .delete_record(collection, &DbRkey::new(rkey))
            .unwrap();
        txn.commit().unwrap();
    }

    fn head(state: &AppState, collection: &str, rkey: &str) -> Option<Vec<u8>> {
        let key = keys::record_key(&did(), collection, &DbRkey::new(rkey));
        state.db.indexer.head(&key).unwrap().map(|v| v.to_vec())
    }

    fn history_bodies(state: &AppState, collection: &str, rkey: &str) -> Vec<Vec<u8>> {
        let key = keys::record_key(&did(), collection, &DbRkey::new(rkey));
        let mut end = key.clone();
        end.push(0xFF);
        state
            .db
            .indexer
            .history
            .range((Bound::Included(key.clone()), Bound::Excluded(end)))
            .filter_map(|guard| guard.into_inner().ok())
            // `{record key} 00 {rev}` for exactly this record (history_hit_for)
            .filter(|(k, _)| k.len() == key.len() + 9 && k.starts_with(&key))
            .map(|(_, v)| v.to_vec())
            .collect()
    }

    #[test]
    fn repo_rows_count_for_the_pds_they_are_active_on() {
        let (_tmp, state) = test_state();
        let did = did();
        let accounts = |host| state.db.get_count_sync(&keys::pds_account_count_key(host));
        let row = |active, pds: &'static str| RepoState {
            active,
            pds: Some(jacquard_common::CowStr::Borrowed(pds)),
            ..RepoState::backfilling()
        };
        let put = |row: &RepoState| {
            let mut txn = Txn::new(&state.db);
            txn.put_repo(&did, row).unwrap();
            txn.commit().unwrap();
        };

        put(&row(true, "https://one.example/"));
        put(&row(true, "https://one.example/"));
        assert_eq!(accounts("one.example"), 1);
        put(&row(false, "https://one.example/"));
        assert_eq!(accounts("one.example"), 0);
        put(&row(true, "https://two.example/"));
        assert_eq!((accounts("one.example"), accounts("two.example")), (0, 1));

        // the second put moves the count from what the first one staged, not the stored row
        let mut txn = Txn::new(&state.db);
        txn.put_repo(&did, &row(true, "https://one.example/"))
            .unwrap();
        txn.put_repo(&did, &row(true, "https://three.example/"))
            .unwrap();
        txn.commit().unwrap();
        assert_eq!(
            (
                accounts("one.example"),
                accounts("two.example"),
                accounts("three.example")
            ),
            (0, 0, 1)
        );

        let mut txn = Txn::new(&state.db);
        txn.remove_repo(&did).unwrap();
        txn.commit().unwrap();
        assert_eq!(accounts("three.example"), 0);
    }

    #[test]
    fn a_put_after_a_read_counts_from_the_row_the_read_saw() {
        let (_tmp, state) = test_state();
        let did = did();
        let accounts = |host| state.db.get_count_sync(&keys::pds_account_count_key(host));
        let row = |pds: &'static str| RepoState {
            pds: Some(jacquard_common::CowStr::Borrowed(pds)),
            ..RepoState::synced()
        };
        let mut txn = Txn::new(&state.db);
        txn.put_repo(&did, &row("https://one.example/")).unwrap();
        txn.commit().unwrap();

        let mut txn = Txn::new(&state.db);
        let read = txn.read_repo(&did).unwrap().expect("repo row");
        assert_eq!(read.pds.as_deref(), Some("https://one.example/"));
        // only the repo's owner writes the row, so a put after the read counts from what the
        // read saw. this stored row, written behind the txn's back, shows the put didn't read
        // the row again
        state
            .db
            .repos
            .insert(
                keys::repo_key(&did),
                crate::db::ser_repo_state(&RepoState::backfilling()).unwrap(),
            )
            .unwrap();
        txn.put_repo(&did, &row("https://two.example/")).unwrap();
        txn.commit().unwrap();
        assert_eq!((accounts("one.example"), accounts("two.example")), (0, 1));

        let other = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap();
        let mut txn = Txn::new(&state.db);
        assert!(txn.read_repo(&other).unwrap().is_none());
        txn.put_repo(&other, &row("https://two.example/")).unwrap();
        txn.commit().unwrap();
        assert_eq!(accounts("two.example"), 2);
    }

    #[test]
    fn create_writes_head_without_history() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );

        assert_eq!(
            head(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            Some(b"v1".to_vec())
        );
        assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty());
    }

    #[test]
    fn update_moves_superseded_body_to_history() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        write(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v2"),
            DbAction::Update,
        );

        assert_eq!(
            head(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            Some(b"v2".to_vec())
        );
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![b"v1".to_vec()]
        );
    }

    #[test]
    fn identical_update_leaves_no_history() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        write(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Update,
        );

        assert_eq!(
            head(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            Some(b"v1".to_vec())
        );
        assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty());
    }

    #[test]
    fn delete_moves_body_to_history() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        delete(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
        );

        assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None);
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![b"v1".to_vec()]
        );
    }

    #[test]
    fn recreate_after_delete_writes_new_head() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        delete(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
        );
        write(
            &state,
            &rev("3kzbif5mog44m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v3"),
            DbAction::Create,
        );

        assert_eq!(
            head(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            Some(b"v3".to_vec())
        );
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![b"v1".to_vec()]
        );
    }

    #[test]
    fn replayed_create_over_identical_head_is_noop() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        // buffered live event replayed after the car import carried the same body
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );

        assert_eq!(
            head(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            Some(b"v1".to_vec())
        );
        assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty());
    }

    #[test]
    fn same_cid_replay_does_not_restore_a_redacted_head() {
        let (_tmp, state) = test_state();
        let collection = "app.bsky.feed.post";
        let rkey = "3kzbif5moe22m";
        let body = body("v1");
        write(
            &state,
            &rev("3kzbif5moe22m"),
            collection,
            rkey,
            &body,
            DbAction::Create,
        );

        crate::db::redact_record_bodies(
            &state.db,
            &did(),
            collection,
            &DbRkey::new(rkey),
            crate::types::DeleteBodyTarget::Head,
        )
        .unwrap();
        write(
            &state,
            &rev("3kzbif5mof33m"),
            collection,
            rkey,
            &body,
            DbAction::Update,
        );

        let head = head(&state, collection, rkey).unwrap();
        assert!(crate::db::is_cid_record_value(&head));
        assert!(history_bodies(&state, collection, rkey).is_empty());
    }

    #[test]
    fn genuinely_new_cid_replaces_a_redacted_head_without_archiving_marker() {
        let (_tmp, state) = test_state();
        let collection = "app.bsky.feed.post";
        let rkey = "3kzbif5moe22m";
        write(
            &state,
            &rev("3kzbif5moe22m"),
            collection,
            rkey,
            &body("v1"),
            DbAction::Create,
        );
        crate::db::redact_record_bodies(
            &state.db,
            &did(),
            collection,
            &DbRkey::new(rkey),
            crate::types::DeleteBodyTarget::Head,
        )
        .unwrap();

        write(
            &state,
            &rev("3kzbif5mof33m"),
            collection,
            rkey,
            &body("v2"),
            DbAction::Update,
        );

        assert_eq!(head(&state, collection, rkey), Some(b"v2".to_vec()));
        assert!(history_bodies(&state, collection, rkey).is_empty());
    }

    #[test]
    fn redacted_cid_stays_redacted_after_an_intervening_version() {
        let (_tmp, state) = test_state();
        let collection = "app.bsky.feed.post";
        let rkey = "3kzbif5moe22m";
        let v1 = body("v1");
        write(
            &state,
            &rev("3kzbif5moe22m"),
            collection,
            rkey,
            &v1,
            DbAction::Create,
        );
        crate::db::redact_record_bodies(
            &state.db,
            &did(),
            collection,
            &DbRkey::new(rkey),
            crate::types::DeleteBodyTarget::Head,
        )
        .unwrap();

        write(
            &state,
            &rev("3kzbif5mof33m"),
            collection,
            rkey,
            &body("v2"),
            DbAction::Update,
        );
        write(
            &state,
            &rev("3kzbif5mog44m"),
            collection,
            rkey,
            &v1,
            DbAction::Update,
        );

        let head = head(&state, collection, rkey).unwrap();
        assert!(crate::db::is_cid_record_value(&head));
    }

    #[test]
    fn redacted_cid_stays_redacted_after_reopen() {
        let tmp = tempfile::tempdir().unwrap();
        let config = crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let collection = "app.bsky.feed.post";
        let rkey = "3kzbif5moe22m";
        let v1 = body("v1");

        {
            let state = AppState::new(&config).unwrap();
            write(
                &state,
                &rev("3kzbif5moe22m"),
                collection,
                rkey,
                &v1,
                DbAction::Create,
            );
            crate::db::redact_record_bodies(
                &state.db,
                &did(),
                collection,
                &DbRkey::new(rkey),
                crate::types::DeleteBodyTarget::Head,
            )
            .unwrap();
            state.db.persist().unwrap();
        }

        let state = AppState::new(&config).unwrap();
        write(
            &state,
            &rev("3kzbif5mof33m"),
            collection,
            rkey,
            &body("v2"),
            DbAction::Update,
        );
        write(
            &state,
            &rev("3kzbif5mog44m"),
            collection,
            rkey,
            &v1,
            DbAction::Update,
        );

        assert!(crate::db::is_cid_record_value(
            &head(&state, collection, rkey).unwrap()
        ));
    }

    #[test]
    fn protocol_delete_of_redacted_head_writes_only_a_tombstone() {
        let (_tmp, state) = test_state();
        let collection = "app.bsky.feed.post";
        let rkey = "3kzbif5moe22m";
        write(
            &state,
            &rev("3kzbif5moe22m"),
            collection,
            rkey,
            &body("v1"),
            DbAction::Create,
        );
        crate::db::redact_record_bodies(
            &state.db,
            &did(),
            collection,
            &DbRkey::new(rkey),
            crate::types::DeleteBodyTarget::Head,
        )
        .unwrap();

        delete(&state, &rev("3kzbif5mof33m"), collection, rkey);

        assert_eq!(head(&state, collection, rkey), None);
        assert_eq!(
            history_bodies(&state, collection, rkey),
            vec![Vec::<u8>::new()]
        );

        write(
            &state,
            &rev("3kzbif5mog44m"),
            collection,
            rkey,
            &body("v1"),
            DbAction::Create,
        );
        assert!(crate::db::is_cid_record_value(
            &head(&state, collection, rkey).unwrap()
        ));
    }

    #[test]
    fn history_redaction_suppresses_that_version_when_it_returns_to_head() {
        let (_tmp, state) = test_state();
        let collection = "app.bsky.feed.post";
        let rkey = "3kzbif5moe22m";
        let v1 = body("v1");
        write(
            &state,
            &rev("3kzbif5moe22m"),
            collection,
            rkey,
            &v1,
            DbAction::Create,
        );
        write(
            &state,
            &rev("3kzbif5mof33m"),
            collection,
            rkey,
            &body("v2"),
            DbAction::Update,
        );
        crate::db::redact_record_bodies(
            &state.db,
            &did(),
            collection,
            &DbRkey::new(rkey),
            crate::types::DeleteBodyTarget::History,
        )
        .unwrap();

        write(
            &state,
            &rev("3kzbif5mog44m"),
            collection,
            rkey,
            &v1,
            DbAction::Update,
        );

        assert!(crate::db::is_cid_record_value(
            &head(&state, collection, rkey).unwrap()
        ));
    }

    #[test]
    fn create_over_differing_head_preserves_it() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        // out-of-order replay: the old head still belongs in history
        write(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v2"),
            DbAction::Create,
        );

        assert_eq!(
            head(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            Some(b"v2".to_vec())
        );
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![b"v1".to_vec()]
        );
    }

    #[test]
    fn update_of_never_seen_record_just_writes_head() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Update,
        );

        assert_eq!(
            head(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            Some(b"v1".to_vec())
        );
        assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty());
    }

    #[test]
    fn delete_of_never_seen_record_writes_empty_tombstone() {
        let (_tmp, state) = test_state();
        delete(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
        );

        assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None);
        // an empty entry records "deleted at rev" even when there was no
        // body to preserve, distinguishing it from "never existed"
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![Vec::<u8>::new()]
        );
        // never seen means nothing to decrement
        assert_eq!(state.db.get_count_sync("records"), 0);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(),
            0
        );
    }

    #[test]
    fn replayed_create_keeps_counts_and_history() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );

        assert_eq!(state.db.get_count_sync("records"), 1);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(),
            1
        );
        assert!(history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_empty());
    }

    #[test]
    fn update_of_never_seen_record_counts_as_create() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Update,
        );

        assert_eq!(state.db.get_count_sync("records"), 1);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(),
            1
        );
    }

    #[test]
    fn update_of_existing_record_leaves_counts_alone() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        write(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v2"),
            DbAction::Update,
        );

        assert_eq!(state.db.get_count_sync("records"), 1);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(),
            1
        );
    }

    #[test]
    fn duplicate_delete_preserves_original_body() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        delete(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
        );
        // buffered replay applies the same delete again at the same rev
        delete(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
        );

        // the original body survives; no empty tombstone overwrites it
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![b"v1".to_vec()]
        );
        assert_eq!(state.db.get_count_sync("records"), 0);
        assert_eq!(state.db.indexer.approximate_history_count(), 1);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(),
            0
        );
    }

    #[test]
    fn duplicate_delete_of_never_seen_writes_one_tombstone() {
        let (_tmp, state) = test_state();
        delete(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
        );
        delete(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
        );

        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![Vec::<u8>::new()]
        );
        assert_eq!(state.db.get_count_sync("records"), 0);
        assert_eq!(state.db.indexer.approximate_history_count(), 1);
    }

    #[test]
    fn multiple_writes_in_one_txn_transition_against_staged_heads() {
        let (_tmp, state) = test_state();
        let did = did();
        let collection = "app.bsky.feed.post";

        // 1. create -> update -> delete in one txn (e.g. resync buffer drain)
        let mut txn = Txn::new(&state.db);
        {
            let mut r1 = txn.records(&state, &rev("3kzbif5moe22m"), &did).unwrap();
            r1.put_record(
                collection,
                &DbRkey::new("create-update"),
                &crate::car::CarBlock::from_body(body("v1")),
                DbAction::Create,
            )
            .unwrap();
            r1.put_record(
                collection,
                &DbRkey::new("create-delete"),
                &crate::car::CarBlock::from_body(body("ephemeral")),
                DbAction::Create,
            )
            .unwrap();
            r1.delete_record(collection, &DbRkey::new("never-seen"))
                .unwrap();
        }
        {
            let mut r2 = txn.records(&state, &rev("3kzbif5mof33m"), &did).unwrap();
            r2.put_record(
                collection,
                &DbRkey::new("create-update"),
                &crate::car::CarBlock::from_body(body("v2")),
                DbAction::Update,
            )
            .unwrap();
            r2.delete_record(collection, &DbRkey::new("create-delete"))
                .unwrap();
            r2.delete_record(collection, &DbRkey::new("create-delete"))
                .unwrap();
            r2.delete_record(collection, &DbRkey::new("never-seen"))
                .unwrap();
        }
        {
            let mut r3 = txn.records(&state, &rev("3kzbif5mog44m"), &did).unwrap();
            r3.put_record(
                collection,
                &DbRkey::new("create-update"),
                &crate::car::CarBlock::from_body(body("v3")),
                DbAction::Update,
            )
            .unwrap();
        }
        txn.commit().unwrap();

        assert_eq!(head(&state, collection, "create-update"), Some(b"v3".to_vec()));
        assert_eq!(
            history_bodies(&state, collection, "create-update"),
            vec![b"v1".to_vec(), b"v2".to_vec()]
        );
        assert_eq!(head(&state, collection, "create-delete"), None);
        assert_eq!(
            history_bodies(&state, collection, "create-delete"),
            vec![b"ephemeral".to_vec()]
        );
        assert_eq!(
            history_bodies(&state, collection, "never-seen"),
            vec![Vec::<u8>::new()]
        );
        assert_eq!(state.db.get_count_sync("records"), 1);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did, collection).unwrap(),
            1
        );

        // 2. delete -> recreate in one txn on an already-stored record
        let mut txn = Txn::new(&state.db);
        {
            let mut r1 = txn.records(&state, &rev("3kzbif5moh55m"), &did).unwrap();
            r1.delete_record(collection, &DbRkey::new("create-update"))
                .unwrap();
        }
        {
            let mut r2 = txn.records(&state, &rev("3kzbif5moi66m"), &did).unwrap();
            r2.put_record(
                collection,
                &DbRkey::new("create-update"),
                &crate::car::CarBlock::from_body(body("v4")),
                DbAction::Create,
            )
            .unwrap();
        }
        txn.commit().unwrap();

        assert_eq!(head(&state, collection, "create-update"), Some(b"v4".to_vec()));
        assert_eq!(
            history_bodies(&state, collection, "create-update"),
            vec![b"v1".to_vec(), b"v2".to_vec(), b"v3".to_vec()]
        );
        assert_eq!(state.db.get_count_sync("records"), 1);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did, collection).unwrap(),
            1
        );
    }

    #[test]
    fn unchecked_block_with_a_forged_cid_is_rejected() {
        let (_tmp, state) = test_state();
        let mut txn = Txn::new(&state.db);
        let did = did();
        let mut record_txn = txn.records(&state, &rev("3kzbif5moe22m"), &did).unwrap();
        let wrong_cid = crate::car::compute_cid(b"not the body");
        let blocks = crate::car::Blocks::unverified(std::collections::BTreeMap::from([(
            wrong_cid,
            body("v1"),
        )]));
        let err = record_txn
            .put_record(
                "app.bsky.feed.post",
                &DbRkey::new("3kzbif5moe22m"),
                &blocks.block(&wrong_cid).unwrap(),
                DbAction::Create,
            )
            .unwrap_err();
        assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
        drop(txn);

        assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None);
        assert_eq!(state.db.get_count_sync("records"), 0);
    }

    #[test]
    fn delayed_commit_death_rev_uses_local_clock() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        // a delayed commit arrives with an old rev: the death must still be
        // keyed on the local clock so event retention and history retention
        // age out together
        let before = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_micros() as i64;
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v2"),
            DbAction::Update,
        );
        let after = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_micros() as i64;

        let key = keys::record_key(&did(), "app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m"));
        let mut end = key.clone();
        end.push(0xFF);
        let death_micros: Vec<i64> = state
            .db
            .indexer
            .history
            .range((Bound::Included(key.clone()), Bound::Excluded(end)))
            .filter_map(|guard| guard.into_inner().ok())
            .filter(|(k, _)| k.len() == key.len() + 9 && k.starts_with(&key))
            .map(|(k, _)| {
                let raw = u64::from_be_bytes(k[key.len() + 1..].try_into().unwrap());
                (raw >> 10) as i64
            })
            .collect();
        assert_eq!(death_micros.len(), 1);
        assert!(
            death_micros[0] >= before && death_micros[0] <= after + 1_000_000,
            "death rev {} should be on the local clock, not the old commit rev",
            death_micros[0]
        );
    }

    #[test]
    fn repo_purge_moves_all_bodies_to_history() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        write(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.like",
            "3kzbif5mof33m",
            &body("v2"),
            DbAction::Create,
        );

        let mut txn = Txn::new(&state.db);
        crate::db::delete_repo_records(&mut txn, &did(), Some(&rev("3kzbif5mog44m")), false)
            .unwrap();
        txn.commit().unwrap();

        assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None);
        assert_eq!(head(&state, "app.bsky.feed.like", "3kzbif5mof33m"), None);
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![b"v1".to_vec()]
        );
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.like", "3kzbif5mof33m"),
            vec![b"v2".to_vec()]
        );
    }

    #[test]
    fn a_links_only_repo_purge_writes_no_history() {
        let tmp = tempfile::tempdir().unwrap();
        let mut config = crate::config::Config::default();
        config.database_path = tmp.path().to_path_buf();
        config.only_index_links = true;
        let state = AppState::new(&config).unwrap();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        assert!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m").is_some());

        let mut txn = Txn::new(&state.db);
        crate::db::delete_repo_records(&mut txn, &did(), None, state.only_index_links).unwrap();
        txn.commit().unwrap();

        assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None);
        assert_eq!(state.db.indexer.history.iter().count(), 0);
        assert_eq!(state.db.get_count_sync("records"), 0);
    }

    #[test]
    fn repo_purge_with_future_skew_root_sorts_death_strictly_after_root_and_updates_counts() {
        let (_tmp, state) = test_state();
        let future_root_rev = rev("3kzbif5mog44m");

        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );
        write(
            &state,
            &rev("3kzbif5mof33m"),
            "app.bsky.feed.like",
            "3kzbif5mof33m",
            &body("v2"),
            DbAction::Create,
        );

        assert_eq!(state.db.get_count_sync("records"), 2);
        assert_eq!(state.db.indexer.approximate_history_count(), 0);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(),
            1
        );
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.like").unwrap(),
            1
        );

        let mut txn = Txn::new(&state.db);
        crate::db::delete_repo_records(&mut txn, &did(), Some(&future_root_rev), false).unwrap();
        txn.commit().unwrap();

        // heads gone, bodies in history
        assert_eq!(head(&state, "app.bsky.feed.post", "3kzbif5moe22m"), None);
        assert_eq!(head(&state, "app.bsky.feed.like", "3kzbif5mof33m"), None);
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.post", "3kzbif5moe22m"),
            vec![b"v1".to_vec()]
        );
        assert_eq!(
            history_bodies(&state, "app.bsky.feed.like", "3kzbif5mof33m"),
            vec![b"v2".to_vec()]
        );

        // count assertions
        assert_eq!(state.db.get_count_sync("records"), 0);
        assert_eq!(state.db.indexer.approximate_history_count(), 2);
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.post").unwrap(),
            0
        );
        assert_eq!(
            crate::db::get_record_count(&state.db, &did(), "app.bsky.feed.like").unwrap(),
            0
        );

        // death rev sorting check: death key in history sorts strictly AFTER future_root_rev
        let post_key =
            keys::record_key(&did(), "app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m"));
        // `history_range_after` inline: (Excluded({key} 00 {rev}), Excluded({key} 01))
        let lo = std::ops::Bound::Excluded(keys::history_key(&post_key, &future_root_rev));
        let mut upper = post_key.clone();
        upper.push(1);
        let hi = std::ops::Bound::Excluded(upper);
        let history_hit = state.db.indexer.history.range((lo, hi)).next();
        assert!(
            history_hit.is_some(),
            "history entry death rev must sort strictly after future root rev so event replay resolves it"
        );
    }

    fn put(records: &mut RecordTxn<'_, '_, '_>, collection: &str, rkey: &str, record: &str) {
        let block = crate::car::CarBlock::from_body(body(record));
        records
            .put_record(collection, &DbRkey::new(rkey), &block, DbAction::Create)
            .unwrap();
    }

    fn record_count(state: &AppState, collection: &str) -> u64 {
        crate::db::get_record_count(&state.db, &did(), collection).unwrap()
    }

    #[test]
    fn record_writes_are_counted_by_the_commit_alone() {
        let (_tmp, state) = test_state();
        let did = did();
        let mut txn = Txn::new(&state.db);
        let mut records = txn.records(&state, &rev("3kzbif5moe22m"), &did).unwrap();
        put(&mut records, "app.bsky.feed.post", "3kzbif5moe22m", "v1");
        put(&mut records, "app.bsky.feed.like", "3kzbif5moe22m", "v1");
        txn.commit().unwrap();

        assert_eq!(state.db.get_count_sync("records"), 2);
        assert_eq!(record_count(&state, "app.bsky.feed.post"), 1);
        assert_eq!(record_count(&state, "app.bsky.feed.like"), 1);
    }

    #[test]
    fn record_txns_in_one_txn_add_up_their_counts() {
        let (_tmp, state) = test_state();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.post",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );

        // a drain step applies several buffered commits in one txn
        let did = did();
        let mut txn = Txn::new(&state.db);
        for rkey in ["3kzbif5mof33m", "3kzbif5mog44m"] {
            let mut records = txn.records(&state, &rev(rkey), &did).unwrap();
            put(&mut records, "app.bsky.feed.post", rkey, rkey);
        }
        let mut records = txn.records(&state, &rev("3kzbif5moh55m"), &did).unwrap();
        records
            .delete_record("app.bsky.feed.post", &DbRkey::new("3kzbif5moe22m"))
            .unwrap();
        txn.commit().unwrap();

        assert_eq!(state.db.get_count_sync("records"), 2);
        assert_eq!(record_count(&state, "app.bsky.feed.post"), 2);
    }

    #[test]
    fn replaced_counts_keep_the_writes_after_them() {
        let (_tmp, state) = test_state();
        let did = did();
        write(
            &state,
            &rev("3kzbif5moe22m"),
            "app.bsky.feed.like",
            "3kzbif5moe22m",
            &body("v1"),
            DbAction::Create,
        );

        let mut txn = Txn::new(&state.db);
        let mut records = txn.records(&state, &rev("3kzbif5mof33m"), &did).unwrap();
        put(&mut records, "app.bsky.feed.post", "3kzbif5mof33m", "v1");
        put(&mut records, "app.bsky.graph.follow", "3kzbif5mof33m", "v1");
        txn.replace_record_counts(&did, |_| true, [("app.bsky.feed.post", 5)])
            .unwrap();
        let mut records = txn.records(&state, &rev("3kzbif5mog44m"), &did).unwrap();
        put(&mut records, "app.bsky.feed.post", "3kzbif5mog44m", "v2");
        txn.commit().unwrap();

        assert_eq!(record_count(&state, "app.bsky.feed.post"), 6);
        // replaced away, whether stored before the txn or only moved in it
        assert_eq!(record_count(&state, "app.bsky.feed.like"), 0);
        assert_eq!(record_count(&state, "app.bsky.graph.follow"), 0);
        assert_eq!(
            state
                .db
                .counts
                .prefix(keys::did_collection_prefix(&did))
                .count(),
            1
        );
    }
}