use std::collections::{BTreeMap, HashSet};
use std::sync::Arc;

use fjall::Keyspace;
use smallvec::SmallVec;
use uuid::Uuid;

use super::MetastoreError;
use super::encoding::{KeyReader, exclusive_upper_bound};
use super::keys::UserHash;
use super::records::{
    RecordValue, record_by_cid_built_key, record_by_cid_index_prefix, record_by_cid_key,
    record_by_cid_key_from_suffix, record_by_cid_prefix, record_by_cid_suffix,
    record_by_cid_user_prefix, record_collection_prefix, record_key,
    record_key_user_hash_and_suffix, record_user_prefix, records_prefix,
};
use super::repo_ops::{bytes_to_cid_link, cid_link_to_bytes};
use super::scan::{
    count_prefix, delete_all_by_prefix, delete_all_by_prefix_chunked, point_lookup, prefix_entries,
    stage_in_chunks,
};
use super::user_hash::UserHashMap;

use tranquil_types::{CidLink, Nsid, Rkey};

pub struct RecordWrite<'a> {
    pub collection: &'a Nsid,
    pub rkey: &'a Rkey,
    pub cid: &'a CidLink,
}

pub struct RecordDelete<'a> {
    pub collection: &'a Nsid,
    pub rkey: &'a Rkey,
}

pub struct ListRecordsQuery<'a> {
    pub user_id: Uuid,
    pub collection: &'a Nsid,
    pub cursor: Option<&'a Rkey>,
    pub limit: usize,
    pub reverse: bool,
    pub rkey_start: Option<&'a Rkey>,
    pub rkey_end: Option<&'a Rkey>,
}

#[derive(Debug, Clone)]
pub struct RecordInfo {
    pub rkey: Rkey,
    pub record_cid: CidLink,
}

#[derive(Debug, Clone)]
pub struct FullRecordInfo {
    pub collection: Nsid,
    pub rkey: Rkey,
    pub record_cid: CidLink,
}

#[derive(Debug, Clone)]
pub struct RecordWithTakedown {
    pub id: Uuid,
    pub collection: Nsid,
    pub rkey: Rkey,
    pub takedown_ref: Option<String>,
}

pub struct RecordOps {
    repo_data: Keyspace,
    user_hashes: Arc<UserHashMap>,
}

impl RecordOps {
    pub fn new(repo_data: Keyspace, user_hashes: Arc<UserHashMap>) -> Self {
        Self {
            repo_data,
            user_hashes,
        }
    }

    pub fn upsert_records(
        &self,
        batch: &mut fjall::OwnedWriteBatch,
        user_hash: UserHash,
        records: &[RecordWrite<'_>],
    ) -> Result<(), MetastoreError> {
        let last_write_per_key: BTreeMap<(&Nsid, &Rkey), &CidLink> = records
            .iter()
            .map(|rec| ((rec.collection, rec.rkey), rec.cid))
            .collect();

        last_write_per_key
            .into_iter()
            .try_for_each(|((collection, rkey), cid)| {
                let key = record_key(user_hash, collection, rkey);
                let cid_bytes = cid_link_to_bytes(cid)?;
                let existing = self
                    .repo_data
                    .get(key.as_slice())
                    .map_err(MetastoreError::Fjall)?
                    .and_then(|raw| RecordValue::deserialize(&raw));
                if let Some(prev) = &existing
                    && prev.record_cid != cid_bytes
                {
                    let stale = record_by_cid_key(user_hash, &prev.record_cid, collection, rkey);
                    batch.remove(&self.repo_data, stale.as_slice());
                }
                let value = RecordValue {
                    record_cid: cid_bytes,
                    takedown_ref: existing.and_then(|v| v.takedown_ref),
                };
                let reverse = record_by_cid_key(user_hash, &value.record_cid, collection, rkey);
                batch.insert(&self.repo_data, key.as_slice(), value.serialize());
                batch.insert(&self.repo_data, reverse.as_slice(), []);
                Ok::<(), MetastoreError>(())
            })
    }

    pub fn delete_records(
        &self,
        batch: &mut fjall::OwnedWriteBatch,
        user_hash: UserHash,
        records: &[RecordDelete<'_>],
    ) -> Result<(), MetastoreError> {
        records.iter().try_for_each(|rec| {
            let key = record_key(user_hash, rec.collection, rec.rkey);
            let existing = self
                .repo_data
                .get(key.as_slice())
                .map_err(MetastoreError::Fjall)?
                .and_then(|raw| RecordValue::deserialize(&raw));
            if let Some(prev) = existing {
                let reverse =
                    record_by_cid_key(user_hash, &prev.record_cid, rec.collection, rec.rkey);
                batch.remove(&self.repo_data, reverse.as_slice());
            }
            batch.remove(&self.repo_data, key.as_slice());
            Ok(())
        })
    }

    pub fn delete_all_records(
        &self,
        batch: &mut fjall::OwnedWriteBatch,
        user_hash: UserHash,
    ) -> Result<(), MetastoreError> {
        delete_all_by_prefix(
            &self.repo_data,
            batch,
            record_by_cid_user_prefix(user_hash).as_slice(),
        )?;
        delete_all_by_prefix(
            &self.repo_data,
            batch,
            record_user_prefix(user_hash).as_slice(),
        )
    }

    pub fn referenced_record_cids(
        &self,
        repo_id: Uuid,
        cids: &[CidLink],
        excluded_keys: &[(&Nsid, &Rkey)],
    ) -> Result<Vec<CidLink>, MetastoreError> {
        let user_hash = self
            .user_hashes
            .get(&repo_id)
            .ok_or(MetastoreError::InvalidInput(
                "repo id has no user hash mapping",
            ))?;

        let excluded: HashSet<SmallVec<[u8; 128]>> = excluded_keys
            .iter()
            .map(|(collection, rkey)| record_by_cid_suffix(collection, rkey))
            .collect();

        cids.iter()
            .map(|cid| {
                let cid_bytes = cid_link_to_bytes(cid)?;
                let prefix = record_by_cid_prefix(user_hash, &cid_bytes);
                let referenced = self
                    .repo_data
                    .prefix(prefix.as_slice())
                    .find_map(|guard| match guard.into_inner() {
                        Err(e) => Some(Err(MetastoreError::Fjall(e))),
                        Ok((key_bytes, _)) => key_bytes
                            .get(prefix.len()..)
                            .is_none_or(|suffix| !excluded.contains(suffix))
                            .then_some(Ok::<(), MetastoreError>(())),
                    })
                    .transpose()?
                    .is_some();
                Ok(referenced.then(|| cid.clone()))
            })
            .filter_map(Result::transpose)
            .collect()
    }

    pub fn get_record_cid(
        &self,
        user_id: Uuid,
        collection: &Nsid,
        rkey: &Rkey,
    ) -> Result<Option<CidLink>, MetastoreError> {
        let user_hash = match self.user_hashes.get(&user_id) {
            Some(h) => h,
            None => return Ok(None),
        };
        let key = record_key(user_hash, collection, rkey);

        point_lookup(
            &self.repo_data,
            key.as_slice(),
            RecordValue::deserialize,
            "invalid record value",
        )?
        .map(|v| bytes_to_cid_link(&v.record_cid))
        .transpose()
    }

    pub fn list_records(
        &self,
        query: &ListRecordsQuery<'_>,
    ) -> Result<Vec<RecordInfo>, MetastoreError> {
        let user_hash = match self.user_hashes.get(&query.user_id) {
            Some(h) => h,
            None => return Ok(Vec::new()),
        };

        let coll_prefix = record_collection_prefix(user_hash, query.collection);
        let coll_upper = exclusive_upper_bound(coll_prefix.as_slice())
            .expect("collection prefix always contains non-0xFF bytes");

        let start_key = query
            .rkey_start
            .map(|rs| record_key(user_hash, query.collection, rs));
        let end_key_upper = query
            .rkey_end
            .map(|re| record_key(user_hash, query.collection, re))
            .map(|ek| {
                exclusive_upper_bound(ek.as_slice())
                    .expect("record key always contains non-0xFF bytes")
            });
        let cursor_key = query
            .cursor
            .map(|c| record_key(user_hash, query.collection, c));

        let mut range_lo: &[u8] = coll_prefix.as_slice();
        let mut range_hi: &[u8] = coll_upper.as_slice();

        if let Some(sk) = start_key.as_ref().filter(|sk| sk.as_slice() > range_lo) {
            range_lo = sk.as_slice();
        }

        if let Some(eu) = end_key_upper.as_ref().filter(|eu| eu.as_slice() < range_hi) {
            range_hi = eu.as_slice();
        }

        let effective_cursor = match query.reverse {
            false => {
                if let Some(ck) = cursor_key.as_ref().filter(|ck| ck.as_slice() < range_hi) {
                    range_hi = ck.as_slice();
                }
                None
            }
            true => {
                let narrowed = cursor_key.as_ref().filter(|ck| ck.as_slice() > range_lo);
                match narrowed {
                    Some(ck) => {
                        range_lo = ck.as_slice();
                        Some(ck.as_slice())
                    }
                    None => None,
                }
            }
        };

        match range_lo >= range_hi {
            true => Ok(Vec::new()),
            false => match query.reverse {
                false => self.list_records_reverse(range_lo, range_hi, query.limit),
                true => {
                    self.list_records_forward(range_lo, range_hi, effective_cursor, query.limit)
                }
            },
        }
    }

    fn list_records_forward(
        &self,
        range_start: &[u8],
        range_end: &[u8],
        cursor_key: Option<&[u8]>,
        limit: usize,
    ) -> Result<Vec<RecordInfo>, MetastoreError> {
        self.repo_data
            .range(range_start..range_end)
            .filter_map(|guard| {
                let (key_bytes, val_bytes) = match guard.into_inner() {
                    Ok(pair) => pair,
                    Err(e) => return Some(Err(MetastoreError::Fjall(e))),
                };

                match cursor_key {
                    Some(ck) if key_bytes.as_ref() <= ck => None,
                    _ => Some(decode_record_info(&key_bytes, &val_bytes)),
                }
            })
            .take(limit)
            .collect()
    }

    fn list_records_reverse(
        &self,
        range_start: &[u8],
        range_end: &[u8],
        limit: usize,
    ) -> Result<Vec<RecordInfo>, MetastoreError> {
        self.repo_data
            .range(range_start..range_end)
            .rev()
            .map(|guard| {
                let (key_bytes, val_bytes) = guard.into_inner().map_err(MetastoreError::Fjall)?;
                decode_record_info(&key_bytes, &val_bytes)
            })
            .take(limit)
            .collect()
    }

    pub fn get_all_records(&self, user_id: Uuid) -> Result<Vec<FullRecordInfo>, MetastoreError> {
        let user_hash = match self.user_hashes.get(&user_id) {
            Some(h) => h,
            None => return Ok(Vec::new()),
        };
        let prefix = record_user_prefix(user_hash);

        self.repo_data
            .prefix(prefix.as_slice())
            .map(|guard| {
                let (key_bytes, val_bytes) = guard.into_inner().map_err(MetastoreError::Fjall)?;
                decode_full_record_info(&key_bytes, &val_bytes)
            })
            .collect()
    }

    pub fn list_collections(&self, user_id: Uuid) -> Result<Vec<Nsid>, MetastoreError> {
        let user_hash = match self.user_hashes.get(&user_id) {
            Some(h) => h,
            None => return Ok(Vec::new()),
        };
        let user_pfx = record_user_prefix(user_hash);
        let user_upper = exclusive_upper_bound(user_pfx.as_slice())
            .expect("user prefix always contains non-0xFF bytes");

        let mut collections: Vec<Nsid> = Vec::new();
        let mut seek_from: SmallVec<[u8; 128]> = user_pfx.clone();

        loop {
            let entry = self
                .repo_data
                .range(seek_from.as_slice()..user_upper.as_slice())
                .next();
            let guard = match entry {
                Some(g) => g,
                None => break,
            };
            let (key_bytes, _) = guard.into_inner().map_err(MetastoreError::Fjall)?;
            let collection = parse_record_key_collection(&key_bytes)
                .ok_or(MetastoreError::CorruptData("invalid record key"))?;
            let coll_prefix = record_collection_prefix(user_hash, &collection);
            seek_from = exclusive_upper_bound(coll_prefix.as_slice())
                .expect("collection prefix always contains non-0xFF bytes");
            collections.push(collection);
        }

        Ok(collections)
    }

    pub fn count_records(&self, user_id: Uuid) -> Result<i64, MetastoreError> {
        let user_hash = match self.user_hashes.get(&user_id) {
            Some(h) => h,
            None => return Ok(0),
        };
        let prefix = record_user_prefix(user_hash);
        count_prefix(&self.repo_data, prefix.as_slice())
    }

    pub fn count_all_records(&self) -> Result<i64, MetastoreError> {
        let prefix = records_prefix();
        count_prefix(&self.repo_data, prefix.as_slice())
    }

    pub fn get_record_by_cid(
        &self,
        cid: &CidLink,
        scope_user: Option<Uuid>,
    ) -> Result<Option<RecordWithTakedown>, MetastoreError> {
        let target_bytes = cid_link_to_bytes(cid)?;
        let prefix = match scope_user {
            Some(uid) => match self.user_hashes.get(&uid) {
                Some(hash) => record_user_prefix(hash),
                None => return Ok(None),
            },
            None => records_prefix(),
        };

        self.repo_data
            .prefix(prefix.as_slice())
            .find_map(|guard| {
                let (key_bytes, val_bytes) = match guard.into_inner() {
                    Ok(pair) => pair,
                    Err(e) => return Some(Err(MetastoreError::Fjall(e))),
                };
                let value = match RecordValue::deserialize(&val_bytes) {
                    Some(v) => v,
                    None => return Some(Err(MetastoreError::CorruptData("invalid record value"))),
                };

                match value.record_cid == target_bytes {
                    true => {
                        let (collection, rkey) = match parse_record_key_fields(&key_bytes) {
                            Some(pair) => pair,
                            None => {
                                return Some(Err(MetastoreError::CorruptData(
                                    "invalid record key",
                                )));
                            }
                        };
                        let user_hash = match parse_record_key_user_hash(&key_bytes) {
                            Some(h) => h,
                            None => {
                                return Some(Err(MetastoreError::CorruptData(
                                    "invalid record key",
                                )));
                            }
                        };
                        let user_id = match self.user_hashes.get_uuid(&user_hash) {
                            Some(id) => id,
                            None => {
                                return Some(Err(MetastoreError::CorruptData(
                                    "record user_hash has no reverse mapping",
                                )));
                            }
                        };
                        Some(Ok(RecordWithTakedown {
                            id: user_id,
                            collection,
                            rkey,
                            takedown_ref: value.takedown_ref,
                        }))
                    }
                    false => None,
                }
            })
            .transpose()
    }

    pub fn set_record_takedown(
        &self,
        db: &fjall::Database,
        cid: &CidLink,
        takedown_ref: Option<&str>,
        scope_user: Option<Uuid>,
    ) -> Result<(), MetastoreError> {
        let target_bytes = cid_link_to_bytes(cid)?;
        let prefix = match scope_user {
            Some(uid) => match self.user_hashes.get(&uid) {
                Some(hash) => record_user_prefix(hash),
                None => return Ok(()),
            },
            None => records_prefix(),
        };

        let found = self
            .repo_data
            .prefix(prefix.as_slice())
            .find_map(|guard| {
                let (key_bytes, val_bytes) = match guard.into_inner() {
                    Ok(pair) => pair,
                    Err(e) => return Some(Err(MetastoreError::Fjall(e))),
                };
                let value = match RecordValue::deserialize(&val_bytes) {
                    Some(v) => v,
                    None => return Some(Err(MetastoreError::CorruptData("invalid record value"))),
                };
                match value.record_cid == target_bytes {
                    true => Some(Ok((key_bytes.to_vec(), value))),
                    false => None,
                }
            })
            .transpose()?;

        match found {
            Some((key, mut value)) => {
                value.takedown_ref = takedown_ref.map(str::to_string);
                let mut batch = db.batch();
                batch.insert(&self.repo_data, &key, value.serialize());
                batch.commit().map_err(MetastoreError::Fjall)
            }
            None => Ok(()),
        }
    }
}

fn decode_record_info(key_bytes: &[u8], val_bytes: &[u8]) -> Result<RecordInfo, MetastoreError> {
    let value = RecordValue::deserialize(val_bytes)
        .ok_or(MetastoreError::CorruptData("invalid record value"))?;
    let (_collection, rkey) = parse_record_key_fields(key_bytes)
        .ok_or(MetastoreError::CorruptData("invalid record key"))?;
    let cid = bytes_to_cid_link(&value.record_cid)?;
    Ok(RecordInfo {
        rkey,
        record_cid: cid,
    })
}

fn decode_full_record_info(
    key_bytes: &[u8],
    val_bytes: &[u8],
) -> Result<FullRecordInfo, MetastoreError> {
    let value = RecordValue::deserialize(val_bytes)
        .ok_or(MetastoreError::CorruptData("invalid record value"))?;
    let (collection, rkey) = parse_record_key_fields(key_bytes)
        .ok_or(MetastoreError::CorruptData("invalid record key"))?;
    let cid = bytes_to_cid_link(&value.record_cid)?;
    Ok(FullRecordInfo {
        collection,
        rkey,
        record_cid: cid,
    })
}

fn parse_record_key_fields(key_bytes: &[u8]) -> Option<(Nsid, Rkey)> {
    let mut reader = KeyReader::new(key_bytes);
    let _tag = reader.tag()?;
    let _user_hash = reader.u64()?;
    let collection = Nsid::new(reader.string()?).ok()?;
    let rkey = Rkey::new(reader.string()?).ok()?;
    Some((collection, rkey))
}

fn parse_record_key_collection(key_bytes: &[u8]) -> Option<Nsid> {
    let mut reader = KeyReader::new(key_bytes);
    let _tag = reader.tag()?;
    let _user_hash = reader.u64()?;
    Nsid::new(reader.string()?).ok()
}

// Committing in chunks like this means
// the rebuild leaves entries under the index prefix
// that only cover some of the records
// if anything interrupts it part-way through
// where the marker goes in last
// and is the only evidence a rebuild ever finished
// so its absence means wiping whatever is there and starting over
// as opposed to trusting a partial index.
pub fn rebuild_record_by_cid_index(
    db: &fjall::Database,
    repo_data: &Keyspace,
) -> Result<usize, MetastoreError> {
    let built_key = record_by_cid_built_key();
    if repo_data
        .get(built_key.as_slice())
        .map_err(MetastoreError::Fjall)?
        .is_some()
    {
        return Ok(0);
    }

    const COMMIT_EVERY: usize = 10_000;
    let discarded = delete_all_by_prefix_chunked(
        db,
        repo_data,
        record_by_cid_index_prefix().as_slice(),
        COMMIT_EVERY,
    )?;
    if discarded > 0 {
        tracing::warn!(
            entries = discarded,
            "discarded a record_by_cid index that an interrupted rebuild left wihtout its marker"
        );
    }

    let mut skipped = 0usize;
    let written = stage_in_chunks(
        db,
        prefix_entries(repo_data, records_prefix().as_slice()),
        COMMIT_EVERY,
        |batch, (key_bytes, val_bytes)| {
            match (
                record_key_user_hash_and_suffix(&key_bytes),
                RecordValue::deserialize(&val_bytes),
            ) {
                (Some((user_hash, suffix)), Some(value)) => {
                    let reverse =
                        record_by_cid_key_from_suffix(user_hash, &value.record_cid, suffix);
                    batch.insert(repo_data, reverse.as_slice(), []);
                }
                _ => skipped += 1,
            }
            Ok(())
        },
    )?;
    if skipped > 0 {
        tracing::warn!(
            skipped,
            "record_by_cid rebuild skipped rows whose key or value doesn't decode. \
             Those records are unreadable through every other path too."
        );
    }

    let mut batch = db.batch();
    batch.insert(repo_data, built_key.as_slice(), []);
    batch.commit()?;
    Ok(written.saturating_sub(skipped))
}

fn parse_record_key_user_hash(key_bytes: &[u8]) -> Option<UserHash> {
    let mut reader = KeyReader::new(key_bytes);
    let _tag = reader.tag()?;
    let hash = reader.u64()?;
    Some(UserHash::from_raw(hash))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::metastore::{Metastore, MetastoreConfig};

    fn test_config() -> MetastoreConfig {
        MetastoreConfig {
            cache_size_bytes: 64 * 1024 * 1024,
        }
    }

    fn test_cid_link(seed: u8) -> CidLink {
        let digest: [u8; 32] = std::array::from_fn(|i| seed.wrapping_add(i as u8));
        let mh = multihash::Multihash::<64>::wrap(0x12, &digest).unwrap();
        let c = cid::Cid::new_v1(0x71, mh);
        CidLink::from_cid(&c)
    }

    fn test_rev(seq: u64) -> tranquil_types::Tid {
        const ALPHABET: &[u8] = b"234567abcdefghijklmnopqrstuvwxyz";
        let s: String = (0..13)
            .rev()
            .map(|i| ALPHABET[((seq >> (i * 5)) & 0x1F) as usize] as char)
            .collect();
        tranquil_types::Tid::new(s).expect("generated TID is valid")
    }

    fn open_fresh() -> (tempfile::TempDir, Metastore) {
        let dir = tempfile::TempDir::new().unwrap();
        let ms = Metastore::open(dir.path(), test_config()).unwrap();
        (dir, ms)
    }

    fn test_did(name: &str) -> tranquil_types::Did {
        tranquil_types::Did::new(format!("did:plc:{name}")).expect("test DID is well-formed")
    }

    fn test_handle(name: &str) -> tranquil_types::Handle {
        tranquil_types::Handle::new(format!("{name}.oyster.cafe")).expect("test handle is valid")
    }

    fn test_nsid(name: impl Into<String>) -> Nsid {
        Nsid::new(name).expect("test collection is a valid NSID")
    }

    fn test_rkey(key: impl Into<String>) -> Rkey {
        Rkey::new(key).expect("test record key is a valid rkey")
    }

    fn setup_user(ms: &Metastore) -> (Uuid, super::super::keys::UserHash) {
        let user_id = Uuid::new_v4();
        let did = test_did("limpet");
        let handle = test_handle("limpet");
        let cid = test_cid_link(0);
        ms.repo_ops()
            .create_repo(ms.database(), user_id, &did, &handle, &cid, &test_rev(0))
            .unwrap();
        let user_hash = ms.user_hashes().get(&user_id).unwrap();
        (user_id, user_hash)
    }

    fn rw<'a>(collection: &'a Nsid, rkey: &'a Rkey, cid: &'a CidLink) -> RecordWrite<'a> {
        RecordWrite {
            collection,
            rkey,
            cid,
        }
    }

    fn rd<'a>(collection: &'a Nsid, rkey: &'a Rkey) -> RecordDelete<'a> {
        RecordDelete { collection, rkey }
    }

    fn lrq<'a>(
        user_id: Uuid,
        collection: &'a Nsid,
        cursor: Option<&'a Rkey>,
        limit: usize,
        reverse: bool,
        rkey_start: Option<&'a Rkey>,
        rkey_end: Option<&'a Rkey>,
    ) -> ListRecordsQuery<'a> {
        ListRecordsQuery {
            user_id,
            collection,
            cursor,
            limit,
            reverse,
            rkey_start,
            rkey_end,
        }
    }

    #[test]
    fn upsert_and_get_record() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("3k2abcd");
        let cid = test_cid_link(1);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        let found = rec_ops.get_record_cid(user_id, &collection, &rkey).unwrap();
        assert_eq!(found, Some(cid));
    }

    #[test]
    fn get_record_returns_none_for_missing() {
        let (_dir, ms) = open_fresh();
        let (user_id, _) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("nonexistent");
        assert!(
            rec_ops
                .get_record_cid(user_id, &collection, &rkey)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn upsert_overwrites_existing() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("3k2abcd");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid1)])
            .unwrap();
        batch.commit().unwrap();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid2)])
            .unwrap();
        batch.commit().unwrap();

        let found = rec_ops.get_record_cid(user_id, &collection, &rkey).unwrap();
        assert_eq!(found, Some(cid2));
    }

    #[test]
    fn delete_records_removes_entries() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("3k2abcd");
        let cid = test_cid_link(1);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        let mut batch = ms.database().batch();
        rec_ops
            .delete_records(&mut batch, user_hash, &[rd(&collection, &rkey)])
            .unwrap();
        batch.commit().unwrap();

        assert!(
            rec_ops
                .get_record_cid(user_id, &collection, &rkey)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn delete_all_records_clears_user() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let coll1 = test_nsid("app.bsky.feed.post");
        let coll2 = test_nsid("app.bsky.feed.like");
        let rkey_a = test_rkey("a");
        let rkey_b = test_rkey("b");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[rw(&coll1, &rkey_a, &cid1), rw(&coll2, &rkey_b, &cid2)],
            )
            .unwrap();
        batch.commit().unwrap();

        assert_eq!(rec_ops.count_records(user_id).unwrap(), 2);

        let mut batch = ms.database().batch();
        rec_ops.delete_all_records(&mut batch, user_hash).unwrap();
        batch.commit().unwrap();

        assert_eq!(rec_ops.count_records(user_id).unwrap(), 0);
    }

    #[test]
    fn list_records_default_desc_with_limit() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..5).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..5).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let results = rec_ops
            .list_records(&lrq(user_id, &collection, None, 3, false, None, None))
            .unwrap();
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].rkey.as_str(), "rkey004");
        assert_eq!(results[1].rkey.as_str(), "rkey003");
        assert_eq!(results[2].rkey.as_str(), "rkey002");
    }

    #[test]
    fn list_records_default_desc_with_cursor() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..5).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..5).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let cursor = test_rkey("rkey003");
        let results = rec_ops
            .list_records(&lrq(
                user_id,
                &collection,
                Some(&cursor),
                10,
                false,
                None,
                None,
            ))
            .unwrap();
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].rkey.as_str(), "rkey002");
        assert_eq!(results[1].rkey.as_str(), "rkey001");
        assert_eq!(results[2].rkey.as_str(), "rkey000");
    }

    #[test]
    fn list_records_reverse_asc() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..5).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..5).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let results = rec_ops
            .list_records(&lrq(user_id, &collection, None, 3, true, None, None))
            .unwrap();
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].rkey.as_str(), "rkey000");
        assert_eq!(results[1].rkey.as_str(), "rkey001");
        assert_eq!(results[2].rkey.as_str(), "rkey002");
    }

    #[test]
    fn list_records_reverse_asc_with_cursor() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..5).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..5).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let cursor = test_rkey("rkey001");
        let results = rec_ops
            .list_records(&lrq(
                user_id,
                &collection,
                Some(&cursor),
                10,
                true,
                None,
                None,
            ))
            .unwrap();
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].rkey.as_str(), "rkey002");
        assert_eq!(results[1].rkey.as_str(), "rkey003");
        assert_eq!(results[2].rkey.as_str(), "rkey004");
    }

    #[test]
    fn list_records_default_desc_rkey_range_bounds() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..10).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..10).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let rkey_start = test_rkey("rkey003");
        let rkey_end = test_rkey("rkey006");
        let results = rec_ops
            .list_records(&lrq(
                user_id,
                &collection,
                None,
                100,
                false,
                Some(&rkey_start),
                Some(&rkey_end),
            ))
            .unwrap();
        assert_eq!(results.len(), 4);
        assert_eq!(results[0].rkey.as_str(), "rkey006");
        assert_eq!(results[3].rkey.as_str(), "rkey003");
    }

    #[test]
    fn get_all_records_across_collections() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let coll1 = test_nsid("app.bsky.feed.like");
        let coll2 = test_nsid("app.bsky.feed.post");
        let rkey_a = test_rkey("a");
        let rkey_b = test_rkey("b");
        let rkey_c = test_rkey("c");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);
        let cid3 = test_cid_link(3);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&coll1, &rkey_a, &cid1),
                    rw(&coll2, &rkey_b, &cid2),
                    rw(&coll1, &rkey_c, &cid3),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        let all = rec_ops.get_all_records(user_id).unwrap();
        assert_eq!(all.len(), 3);
        assert_eq!(all[0].collection.as_str(), "app.bsky.feed.like");
        assert_eq!(all[1].collection.as_str(), "app.bsky.feed.like");
        assert_eq!(all[2].collection.as_str(), "app.bsky.feed.post");
    }

    #[test]
    fn list_collections_returns_distinct_sorted() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let coll1 = test_nsid("app.bsky.feed.like");
        let coll2 = test_nsid("app.bsky.feed.post");
        let coll3 = test_nsid("app.bsky.graph.follow");
        let rkeys: Vec<Rkey> = ["a", "b", "c", "d", "e"]
            .iter()
            .map(|s| test_rkey(s.to_string()))
            .collect();
        let cids: Vec<CidLink> = (1..=5).map(test_cid_link).collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&coll1, &rkeys[0], &cids[0]),
                    rw(&coll2, &rkeys[1], &cids[1]),
                    rw(&coll1, &rkeys[2], &cids[2]),
                    rw(&coll3, &rkeys[3], &cids[3]),
                    rw(&coll2, &rkeys[4], &cids[4]),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        let collections = rec_ops.list_collections(user_id).unwrap();
        assert_eq!(collections.len(), 3);
        assert_eq!(collections[0].as_str(), "app.bsky.feed.like");
        assert_eq!(collections[1].as_str(), "app.bsky.feed.post");
        assert_eq!(collections[2].as_str(), "app.bsky.graph.follow");
    }

    #[test]
    fn count_records_and_count_all() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        assert_eq!(rec_ops.count_records(user_id).unwrap(), 0);
        assert_eq!(rec_ops.count_all_records().unwrap(), 0);

        let collection = test_nsid("app.bsky.feed.post");
        let rkey_a = test_rkey("a");
        let rkey_b = test_rkey("b");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&collection, &rkey_a, &cid1),
                    rw(&collection, &rkey_b, &cid2),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        assert_eq!(rec_ops.count_records(user_id).unwrap(), 2);
        assert_eq!(rec_ops.count_all_records().unwrap(), 2);
    }

    #[test]
    fn records_isolated_between_users() {
        let (_dir, ms) = open_fresh();
        let rec_ops = ms.record_ops();

        let user1 = Uuid::new_v4();
        let did1 = test_did("user1");
        let handle1 = test_handle("user1");
        ms.repo_ops()
            .create_repo(
                ms.database(),
                user1,
                &did1,
                &handle1,
                &test_cid_link(0),
                &test_rev(0),
            )
            .unwrap();
        let hash1 = ms.user_hashes().get(&user1).unwrap();

        let user2 = Uuid::new_v4();
        let did2 = test_did("user2");
        let handle2 = test_handle("user2");
        ms.repo_ops()
            .create_repo(
                ms.database(),
                user2,
                &did2,
                &handle2,
                &test_cid_link(0),
                &test_rev(0),
            )
            .unwrap();
        let hash2 = ms.user_hashes().get(&user2).unwrap();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey_a = test_rkey("a");
        let rkey_b = test_rkey("b");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, hash1, &[rw(&collection, &rkey_a, &cid1)])
            .unwrap();
        rec_ops
            .upsert_records(&mut batch, hash2, &[rw(&collection, &rkey_b, &cid2)])
            .unwrap();
        batch.commit().unwrap();

        assert_eq!(rec_ops.count_records(user1).unwrap(), 1);
        assert_eq!(rec_ops.count_records(user2).unwrap(), 1);
        assert_eq!(rec_ops.count_all_records().unwrap(), 2);
    }

    #[test]
    fn get_record_by_cid_finds_match() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("r1");
        let cid = test_cid_link(42);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        let found = rec_ops.get_record_by_cid(&cid, None).unwrap().unwrap();
        assert_eq!(found.id, user_id);
        assert_eq!(found.collection.as_str(), "app.bsky.feed.post");
        assert_eq!(found.rkey.as_str(), "r1");
        assert!(found.takedown_ref.is_none());
    }

    #[test]
    fn get_record_by_cid_returns_none_for_missing() {
        let (_dir, ms) = open_fresh();
        let rec_ops = ms.record_ops();
        assert!(
            rec_ops
                .get_record_by_cid(&test_cid_link(99), None)
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn get_record_by_cid_scoped_to_user() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("r1");
        let cid = test_cid_link(42);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        let found = rec_ops
            .get_record_by_cid(&cid, Some(user_id))
            .unwrap()
            .unwrap();
        assert_eq!(found.id, user_id);

        let other_user = Uuid::new_v4();
        assert!(
            rec_ops
                .get_record_by_cid(&cid, Some(other_user))
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn set_and_get_takedown() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("r1");
        let cid = test_cid_link(42);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        rec_ops
            .set_record_takedown(ms.database(), &cid, Some("DMCA-789"), None)
            .unwrap();

        let found = rec_ops.get_record_by_cid(&cid, None).unwrap().unwrap();
        assert_eq!(found.id, user_id);
        assert_eq!(found.takedown_ref.as_deref(), Some("DMCA-789"));

        rec_ops
            .set_record_takedown(ms.database(), &cid, None, None)
            .unwrap();

        let found = rec_ops.get_record_by_cid(&cid, None).unwrap().unwrap();
        assert!(found.takedown_ref.is_none());
    }

    #[test]
    fn upsert_preserves_existing_takedown() {
        let (_dir, ms) = open_fresh();
        let (_user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("r1");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid1)])
            .unwrap();
        batch.commit().unwrap();

        rec_ops
            .set_record_takedown(ms.database(), &cid1, Some("DMCA-999"), None)
            .unwrap();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid2)])
            .unwrap();
        batch.commit().unwrap();

        let found = rec_ops.get_record_by_cid(&cid2, None).unwrap().unwrap();
        assert_eq!(found.takedown_ref.as_deref(), Some("DMCA-999"));
    }

    #[test]
    fn records_survive_reopen() {
        let dir = tempfile::TempDir::new().unwrap();
        let user_id;
        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("durable");
        let cid = test_cid_link(77);

        {
            let ms = Metastore::open(dir.path(), test_config()).unwrap();
            user_id = Uuid::new_v4();
            let did = test_did("persist");
            let handle = test_handle("persist");
            ms.repo_ops()
                .create_repo(
                    ms.database(),
                    user_id,
                    &did,
                    &handle,
                    &test_cid_link(0),
                    &test_rev(0),
                )
                .unwrap();
            let user_hash = ms.user_hashes().get(&user_id).unwrap();

            let rec_ops = ms.record_ops();
            let mut batch = ms.database().batch();
            rec_ops
                .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
                .unwrap();
            batch.commit().unwrap();
            ms.persist().unwrap();
        }

        {
            let ms = Metastore::open(dir.path(), test_config()).unwrap();
            let rec_ops = ms.record_ops();
            let found = rec_ops.get_record_cid(user_id, &collection, &rkey).unwrap();
            assert_eq!(found, Some(cid));
        }
    }

    #[test]
    fn rkey_ordering_matches_lexicographic() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys_unordered = ["zebra", "apple", "mango", "banana"];

        let rkeys: Vec<Rkey> = rkeys_unordered
            .iter()
            .map(|rk| test_rkey(rk.to_string()))
            .collect();
        let cids: Vec<CidLink> = (0..rkeys.len())
            .map(|i| test_cid_link(i as u8 + 1))
            .collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let results = rec_ops
            .list_records(&lrq(user_id, &collection, None, 100, false, None, None))
            .unwrap();
        let result_rkeys: Vec<&str> = results.iter().map(|r| r.rkey.as_str()).collect();
        assert_eq!(result_rkeys, ["zebra", "mango", "banana", "apple"]);
    }

    #[test]
    fn record_with_shortest_rkey() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("z");
        let cid = test_cid_link(1);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        let found = rec_ops.get_record_cid(user_id, &collection, &rkey).unwrap();
        assert_eq!(found, Some(cid));

        let results = rec_ops
            .list_records(&lrq(user_id, &collection, None, 100, false, None, None))
            .unwrap();
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn record_rkey_extending_another_stays_distinct() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey_extended = test_rkey("abc.def");
        let rkey_plain = test_rkey("abc");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&collection, &rkey_extended, &cid1),
                    rw(&collection, &rkey_plain, &cid2),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        let found1 = rec_ops
            .get_record_cid(user_id, &collection, &rkey_extended)
            .unwrap();
        assert_eq!(found1, Some(cid1));

        let found2 = rec_ops
            .get_record_cid(user_id, &collection, &rkey_plain)
            .unwrap();
        assert_eq!(found2, Some(cid2));

        let results = rec_ops
            .list_records(&lrq(user_id, &collection, None, 100, false, None, None))
            .unwrap();
        assert_eq!(results.len(), 2);
        assert_eq!(results[0].rkey.as_str(), "abc.def");
        assert_eq!(results[1].rkey.as_str(), "abc");
    }

    #[test]
    fn record_collection_extending_another_stays_distinct() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let coll_normal = test_nsid("app.bsky.feed.post");
        let coll_extended = test_nsid("app.bsky.feed.post.extra");
        let rkey = test_rkey("r1");
        let cid1 = test_cid_link(1);
        let cid2 = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&coll_normal, &rkey, &cid1),
                    rw(&coll_extended, &rkey, &cid2),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        let found1 = rec_ops
            .get_record_cid(user_id, &coll_normal, &rkey)
            .unwrap();
        assert_eq!(found1, Some(cid1));

        let found2 = rec_ops
            .get_record_cid(user_id, &coll_extended, &rkey)
            .unwrap();
        assert_eq!(found2, Some(cid2));

        let results_normal = rec_ops
            .list_records(&lrq(user_id, &coll_normal, None, 100, false, None, None))
            .unwrap();
        assert_eq!(results_normal.len(), 1);

        let results_extended = rec_ops
            .list_records(&lrq(user_id, &coll_extended, None, 100, false, None, None))
            .unwrap();
        assert_eq!(results_extended.len(), 1);

        let collections = rec_ops.list_collections(user_id).unwrap();
        assert_eq!(collections.len(), 2);
    }

    #[test]
    fn list_records_cursor_before_all_returns_empty() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..5).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..5).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let cursor = test_rkey("a");
        let results = rec_ops
            .list_records(&lrq(
                user_id,
                &collection,
                Some(&cursor),
                100,
                false,
                None,
                None,
            ))
            .unwrap();
        assert!(results.is_empty());
    }

    #[test]
    fn list_records_reverse_asc_with_rkey_bounds() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..10).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..10).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let rkey_start = test_rkey("rkey002");
        let rkey_end = test_rkey("rkey007");
        let results = rec_ops
            .list_records(&lrq(
                user_id,
                &collection,
                None,
                100,
                true,
                Some(&rkey_start),
                Some(&rkey_end),
            ))
            .unwrap();
        assert_eq!(results.len(), 6);
        assert_eq!(results[0].rkey.as_str(), "rkey002");
        assert_eq!(results[5].rkey.as_str(), "rkey007");
    }

    #[test]
    fn list_records_default_desc_cursor_narrows_range() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkeys: Vec<Rkey> = (0..10).map(|i| test_rkey(format!("rkey{i:03}"))).collect();
        let cids: Vec<CidLink> = (0..10).map(|i| test_cid_link(i + 1)).collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rk, c)| rw(&collection, rk, c))
            .collect();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        let cursor = test_rkey("rkey005");
        let results = rec_ops
            .list_records(&lrq(
                user_id,
                &collection,
                Some(&cursor),
                3,
                false,
                None,
                None,
            ))
            .unwrap();
        assert_eq!(results.len(), 3);
        assert_eq!(results[0].rkey.as_str(), "rkey004");
        assert_eq!(results[1].rkey.as_str(), "rkey003");
        assert_eq!(results[2].rkey.as_str(), "rkey002");
    }

    #[test]
    fn exclusive_upper_bound_basic() {
        let prefix = &[0x01, 0x02, 0x03];
        let upper = exclusive_upper_bound(prefix).unwrap();
        assert_eq!(upper.as_slice(), &[0x01, 0x02, 0x04]);
    }

    #[test]
    fn exclusive_upper_bound_with_trailing_ff() {
        let prefix = &[0x01, 0xFF, 0xFF];
        let upper = exclusive_upper_bound(prefix).unwrap();
        assert_eq!(upper.as_slice(), &[0x02]);
    }

    #[test]
    fn exclusive_upper_bound_all_ff_returns_none() {
        assert!(exclusive_upper_bound(&[0xFF, 0xFF, 0xFF]).is_none());
    }

    #[test]
    fn exclusive_upper_bound_empty_returns_none() {
        assert!(exclusive_upper_bound(&[]).is_none());
    }

    #[test]
    fn exclusive_upper_bound_preserves_prefix_ordering() {
        let prefix = &[
            0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x2A, 0x61, 0x00, 0x00,
        ];
        let upper = exclusive_upper_bound(prefix).unwrap();

        let key_inside = {
            let mut k = prefix.to_vec();
            k.extend_from_slice(&[0x62, 0x00, 0x00]);
            k
        };
        assert!(key_inside.as_slice() < upper.as_slice());

        let key_outside = {
            let mut k = Vec::from(&prefix[..prefix.len() - 2]);
            k.extend_from_slice(&[0x00, 0x01, 0x00, 0x00]);
            k
        };
        assert!(key_outside.as_slice() >= upper.as_slice());
    }

    #[test]
    fn referenced_record_cids_reports_a_cid_held_by_another_key() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let kept = test_rkey("kept");
        let dropped = test_rkey("dropped");
        let shared = test_cid_link(9);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&collection, &kept, &shared),
                    rw(&collection, &dropped, &shared),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        assert_eq!(
            rec_ops
                .referenced_record_cids(
                    user_id,
                    std::slice::from_ref(&shared),
                    &[(&collection, &dropped)]
                )
                .unwrap(),
            vec![shared.clone()]
        );
        assert!(
            rec_ops
                .referenced_record_cids(
                    user_id,
                    &[shared],
                    &[(&collection, &dropped), (&collection, &kept)]
                )
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn referenced_record_cids_errors_for_a_repo_with_no_user_hash() {
        let (_dir, ms) = open_fresh();
        let rec_ops = ms.record_ops();

        assert!(matches!(
            rec_ops.referenced_record_cids(Uuid::new_v4(), &[test_cid_link(1)], &[]),
            Err(MetastoreError::InvalidInput(_))
        ));
    }

    #[test]
    fn two_upserts_of_one_key_in_a_batch_leave_only_the_last_cid_indexed() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("twice");
        let first = test_cid_link(20);
        let second = test_cid_link(21);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&collection, &rkey, &first),
                    rw(&collection, &rkey, &second),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        assert!(
            rec_ops
                .referenced_record_cids(user_id, &[first], &[])
                .unwrap()
                .is_empty(),
            "overwritten cid must not stay in the reverse index"
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, std::slice::from_ref(&second), &[])
                .unwrap(),
            vec![second]
        );
    }

    #[test]
    fn referenced_record_cids_forgets_a_cid_an_upsert_replaced() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("churned");
        let first = test_cid_link(1);
        let second = test_cid_link(2);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &first)])
            .unwrap();
        batch.commit().unwrap();

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &second)])
            .unwrap();
        batch.commit().unwrap();

        assert!(
            rec_ops
                .referenced_record_cids(user_id, &[first], &[])
                .unwrap()
                .is_empty()
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, std::slice::from_ref(&second), &[])
                .unwrap(),
            vec![second]
        );
    }

    #[test]
    fn referenced_record_cids_forgets_a_deleted_record() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("gone");
        let cid = test_cid_link(3);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        let mut batch = ms.database().batch();
        rec_ops
            .delete_records(&mut batch, user_hash, &[rd(&collection, &rkey)])
            .unwrap();
        batch.commit().unwrap();

        assert!(
            rec_ops
                .referenced_record_cids(user_id, &[cid], &[])
                .unwrap()
                .is_empty()
        );
    }

    #[test]
    fn delete_all_records_clears_the_reverse_index() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("only");
        let cid = test_cid_link(4);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        let mut batch = ms.database().batch();
        rec_ops.delete_all_records(&mut batch, user_hash).unwrap();
        batch.commit().unwrap();

        assert!(
            ms.partition(crate::metastore::Partition::RepoData)
                .prefix(record_by_cid_user_prefix(user_hash).as_slice())
                .next()
                .is_none()
        );
        assert!(
            rec_ops
                .referenced_record_cids(user_id, &[cid], &[])
                .unwrap()
                .is_empty()
        );
    }

    fn drop_index_and_marker(ms: &Metastore, repo_data: &Keyspace) {
        let mut batch = ms.database().batch();
        crate::metastore::scan::delete_all_by_prefix(
            repo_data,
            &mut batch,
            record_by_cid_index_prefix().as_slice(),
        )
        .unwrap();
        batch.remove(repo_data, record_by_cid_built_key().as_slice());
        batch.commit().unwrap();
    }

    #[test]
    fn rebuild_backfills_a_store_written_before_the_index_existed() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();
        let repo_data = ms.partition(crate::metastore::Partition::RepoData).clone();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("legacy");
        let cid = test_cid_link(5);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        drop_index_and_marker(&ms, &repo_data);
        assert!(
            rec_ops
                .referenced_record_cids(user_id, std::slice::from_ref(&cid), &[])
                .unwrap()
                .is_empty()
        );

        assert_eq!(
            rebuild_record_by_cid_index(ms.database(), &repo_data).unwrap(),
            1
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, &[cid], &[])
                .unwrap()
                .len(),
            1
        );
    }

    #[test]
    fn rebuild_redoes_an_index_left_partial_by_a_crash() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();
        let repo_data = ms.partition(crate::metastore::Partition::RepoData).clone();

        let collection = test_nsid("app.bsky.feed.post");
        let indexed = test_rkey("indexed");
        let missed = test_rkey("missed");
        let indexed_cid = test_cid_link(7);
        let missed_cid = test_cid_link(8);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[
                    rw(&collection, &indexed, &indexed_cid),
                    rw(&collection, &missed, &missed_cid),
                ],
            )
            .unwrap();
        batch.commit().unwrap();

        drop_index_and_marker(&ms, &repo_data);
        let mut batch = ms.database().batch();
        batch.insert(
            &repo_data,
            record_by_cid_key(
                user_hash,
                &cid_link_to_bytes(&indexed_cid).unwrap(),
                &collection,
                &indexed,
            )
            .as_slice(),
            [],
        );
        batch.commit().unwrap();

        assert_eq!(
            rebuild_record_by_cid_index(ms.database(), &repo_data).unwrap(),
            2,
            "a partial index without the marker must be rebuilt from scratch"
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, &[indexed_cid, missed_cid], &[])
                .unwrap()
                .len(),
            2
        );
    }

    #[test]
    fn rebuild_drops_a_stale_entry_an_older_binary_left_behind() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();
        let repo_data = ms.partition(crate::metastore::Partition::RepoData).clone();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("churned");
        let stale_cid = test_cid_link(10);
        let live_cid = test_cid_link(11);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &live_cid)])
            .unwrap();
        batch.insert(
            &repo_data,
            record_by_cid_key(
                user_hash,
                &cid_link_to_bytes(&stale_cid).unwrap(),
                &collection,
                &rkey,
            )
            .as_slice(),
            [],
        );
        batch.remove(&repo_data, record_by_cid_built_key().as_slice());
        batch.commit().unwrap();

        assert_eq!(
            rebuild_record_by_cid_index(ms.database(), &repo_data).unwrap(),
            1
        );
        assert!(
            rec_ops
                .referenced_record_cids(user_id, &[stale_cid], &[])
                .unwrap()
                .is_empty()
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, &[live_cid], &[])
                .unwrap()
                .len(),
            1
        );
    }

    #[test]
    fn rebuild_commits_in_chunks_past_the_batch_threshold() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();
        let repo_data = ms.partition(crate::metastore::Partition::RepoData).clone();

        let collection = test_nsid("app.bsky.feed.post");
        let record_count = 10_001usize;
        let rkeys: Vec<Rkey> = (0..record_count)
            .map(|i| test_rkey(format!("chunk{i:05}")))
            .collect();
        let cids: Vec<CidLink> = (0..record_count)
            .map(|i| test_cid_link((i % 251) as u8))
            .collect();

        let writes: Vec<RecordWrite<'_>> = rkeys
            .iter()
            .zip(cids.iter())
            .map(|(rkey, cid)| rw(&collection, rkey, cid))
            .collect();
        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &writes)
            .unwrap();
        batch.commit().unwrap();

        drop_index_and_marker(&ms, &repo_data);

        assert_eq!(
            rebuild_record_by_cid_index(ms.database(), &repo_data).unwrap(),
            record_count
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, std::slice::from_ref(&cids[0]), &[])
                .unwrap()
                .len(),
            1
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, std::slice::from_ref(cids.last().unwrap()), &[])
                .unwrap()
                .len(),
            1
        );
    }

    #[test]
    fn rebuild_discards_a_large_stale_index_past_the_batch_threshold() {
        let (_dir, ms) = open_fresh();
        let (user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();
        let repo_data = ms.partition(crate::metastore::Partition::RepoData).clone();

        let collection = test_nsid("app.bsky.feed.post");
        let live_rkey = test_rkey("live");
        let live_cid = test_cid_link(1);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(
                &mut batch,
                user_hash,
                &[rw(&collection, &live_rkey, &live_cid)],
            )
            .unwrap();
        batch.commit().unwrap();

        let stale_count = 10_001usize;
        let stale: Vec<(Rkey, Vec<u8>)> = (0..stale_count)
            .map(|i| {
                let rkey = test_rkey(format!("stale{i:05}"));
                let cid = cid_link_to_bytes(&test_cid_link(2)).unwrap();
                let cid = cid
                    .iter()
                    .copied()
                    .chain((i as u32).to_be_bytes())
                    .collect();
                (rkey, cid)
            })
            .collect();
        let mut batch = ms.database().batch();
        stale.iter().for_each(|(rkey, cid)| {
            batch.insert(
                &repo_data,
                record_by_cid_key(user_hash, cid, &collection, rkey).as_slice(),
                [],
            );
        });
        batch.remove(&repo_data, record_by_cid_built_key().as_slice());
        batch.commit().unwrap();

        assert_eq!(
            rebuild_record_by_cid_index(ms.database(), &repo_data).unwrap(),
            1
        );
        assert_eq!(
            repo_data
                .prefix(record_by_cid_index_prefix().as_slice())
                .count(),
            1,
            "rebuild must leave exactly one index entry, so no stale key survived the chunked delete"
        );
        assert_eq!(
            rec_ops
                .referenced_record_cids(user_id, &[live_cid], &[])
                .unwrap()
                .len(),
            1
        );
    }

    #[test]
    fn rebuild_is_a_noop_once_the_marker_exists() {
        let (_dir, ms) = open_fresh();
        let (_user_id, user_hash) = setup_user(&ms);
        let rec_ops = ms.record_ops();
        let repo_data = ms.partition(crate::metastore::Partition::RepoData).clone();

        let collection = test_nsid("app.bsky.feed.post");
        let rkey = test_rkey("present");
        let cid = test_cid_link(6);

        let mut batch = ms.database().batch();
        rec_ops
            .upsert_records(&mut batch, user_hash, &[rw(&collection, &rkey, &cid)])
            .unwrap();
        batch.commit().unwrap();

        assert_eq!(
            rebuild_record_by_cid_index(ms.database(), &repo_data).unwrap(),
            0
        );
    }
}
