use std::ops::Bound;

use fjall::{Keyspace, OwnedWriteBatch, Readable};
use miette::{Context, IntoDiagnostic, Result};

use crate::db::Db;
use crate::db::keys::{SEP, VERSIONING_KEY};

#[cfg(feature = "indexer")]
const LEGACY_BLOCKS_KEYSPACE: &str = "blocks";

/// open the pre-v10 CAS only when it actually exists. keeping this out of the
/// live keyspace registry means a fresh database never recreates the retired
/// layout merely so a migration can observe its absence.
#[cfg(feature = "indexer")]
fn legacy_blocks(db: &Db) -> Result<Option<Keyspace>> {
    if !db.inner.keyspace_exists(LEGACY_BLOCKS_KEYSPACE) {
        return Ok(None);
    }
    db.inner
        .keyspace(LEGACY_BLOCKS_KEYSPACE, Default::default)
        .into_diagnostic()
        .map(Some)
}

#[cfg(feature = "indexer")]
fn legacy_block(db: &Db, key: &[u8]) -> Result<Option<fjall::Slice>> {
    legacy_blocks(db)?
        .map(|blocks| blocks.get(key).into_diagnostic())
        .transpose()
        .map(Option::flatten)
}

#[cfg(all(test, feature = "indexer"))]
fn legacy_blocks_for_test(db: &Db) -> Result<Keyspace> {
    db.inner
        .keyspace(LEGACY_BLOCKS_KEYSPACE, Default::default)
        .into_diagnostic()
}

/// the pre-v12 record store. v12's schema drops `records` from the registry,
/// but its on-disk tables persist until the `recompress_record_heads`
/// migration copies them into `heads` and deletes the keyspace.
#[cfg(feature = "indexer")]
const LEGACY_RECORDS_KEYSPACE: &str = "records";

#[cfg(all(test, feature = "indexer"))]
fn legacy_records_for_test(db: &Db) -> Result<Keyspace> {
    db.inner
        .keyspace(LEGACY_RECORDS_KEYSPACE, Default::default)
        .into_diagnostic()
}

#[cfg(feature = "indexer")]
fn legacy_records(db: &Db) -> Result<Option<Keyspace>> {
    if !db.inner.keyspace_exists(LEGACY_RECORDS_KEYSPACE) {
        return Ok(None);
    }
    db.inner
        .keyspace(LEGACY_RECORDS_KEYSPACE, Default::default)
        .into_diagnostic()
        .map(Some)
}

mod v1;
mod v10;
mod v11;
mod v2;
mod v3;
mod v4;
mod v5;
mod v6;
mod v7;
mod v8;
mod v9;

/// a migration whose entire write set is buffered into one batch.
/// returns `Ok(true)` if applied, or `Ok(false)` if skipped due to missing keyspaces in this build.
type AtomicFn = fn(&Db, &mut OwnedWriteBatch) -> Result<bool>;

/// transform one scanned entry. staged writes may target any keyspace.
/// returns the number of output bytes staged into the write batch for this entry.
/// visit one scanned entry, staging its rewrite into the chunk batch, and
/// return the staged byte count for the chunk budget.
///
/// the third argument is the keyspace being scanned: in-place passes stage
/// into it, so a pass keeps writing the same store it reads even when that
/// store is a legacy keyspace the registry no longer knows.
type VisitFn = fn(&Db, &mut OwnedWriteBatch, &Keyspace, &[u8], &[u8]) -> Result<usize>;

/// destructive or otherwise non-batched work that runs only after every
/// chunked pass is durable. returning false means this build does not contain
/// the keyspaces needed by the migration.
type FinalizeFn = fn(&Db) -> Result<bool>;

/// how a migration is applied.
enum Migration {
    /// the whole write set is buffered into one batch, committed atomically with
    /// the version bump.
    ///
    /// a batch is held entirely in memory, so this is only for migrations whose
    /// write set is small and bounded. it is *not* bounded by how much the
    /// migration reads: streaming a large keyspace to accumulate a handful of
    /// counters is fine, staging one write per entry of a large keyspace is not.
    Atomic(AtomicFn),
    /// ordered resumable passes, each committed in chunks with bounded memory.
    ///
    /// the only way to rewrite a keyspace larger than memory. passes run in
    /// order, and each runs to completion before the next starts.
    Chunked {
        passes: &'static [Pass],
        finalize: Option<FinalizeFn>,
    },
}

/// per-chunk limits. both bound peak batch memory; whichever is hit first ends
/// the chunk. they also bound how much work a crash re-does.
#[derive(Clone, Copy)]
struct ChunkBudget {
    /// entries scanned per chunk.
    entries: usize,
    /// staged output bytes per chunk.
    ///
    /// bounds peak write batch memory when converting small input keys/references
    /// into large staged bodies (e.g. v10 record body expansion).
    bytes: usize,
}

impl ChunkBudget {
    /// what a pass should use unless it has a measured reason not to.
    #[allow(dead_code)]
    const DEFAULT: Self = Self {
        entries: 100_000,
        bytes: 64 * 1024 * 1024,
    };
}

/// one resumable key-ordered scan over a single keyspace.
///
/// the framework owns iteration, chunk commits, and the resume cursor; a pass
/// supplies only [`Pass::visit`].
///
/// # visit must be idempotent
///
/// the resume cursor advances only on a committed chunk, so a crash re-visits
/// every entry of the in-flight chunk.
///
/// more subtly, a pass that writes into the keyspace it scans can observe its
/// own output. each chunk reads a fresh snapshot, so output that sorts *after*
/// the resume cursor is scanned again in a later chunk. `visit` must therefore
/// be safe to apply twice, and re-visiting its own output must reach a fixpoint
/// — normally by making the migrated form recognizable and doing nothing when it
/// is seen. a transform whose output sorts before its input (as key-shortening
/// rewrites usually do) is never rescanned, but that is a property of the
/// encoding, not a guarantee of the framework.
struct Pass {
    /// stable cursor identity. THIS SHOULD ALWAYS BE STABLE. DO NOT CHANGE.
    name: &'static str,
    /// on-disk name of the keyspace to scan.
    ///
    /// the registry resolves names from this build first; a name it no longer
    /// knows falls back to opening an existing on-disk keyspace, so a pass can
    /// still finish migrating data out of a store a later schema dropped.
    ///
    /// a pass whose keyspace exists nowhere is skipped, so a migration can
    /// name mode-specific keyspaces unconditionally — unless
    /// `optional_on_absent` is set, in which case absence means the database
    /// never had the source layout and the pass completes as a no-op.
    ///
    /// a pass scanning `counts` also sees its own resume cursor, which sorts
    /// between the `k|` and `r|` prefixes; `visit` must ignore keys under
    /// [`MIGRATION_CURSOR_PREFIX`].
    scan: &'static str,
    visit: VisitFn,
    budget: ChunkBudget,
    optional_on_absent: bool,
}

/// ordered list of schema migrations.
///
/// invariants:
/// - migration `vN` upgrades on-disk data from schema version `N-1` to `N`
/// - once a migration ships, its input/output wire types are frozen
/// - never "fix up" an older migration to match a newer live type; add a new
///   schema version and a new migration instead
/// - if a stored type changes shape, define a new versioned type in
///   `src/types.rs`, export the newest one for live code, and have the new
///   migration explicitly deserialize the previous version and write the new one
/// - a migration that stages one write per entry of a high-cardinality keyspace
///   must be [`Migration::Chunked`]; [`Migration::Atomic`] would buffer the
///   whole rewrite in memory
///
/// example: if `RepoState` changes again after `types::v7::RepoState`, add a new
/// `types::v8::RepoState`, keep `v7` frozen, and add a migration that rewrites
/// `v7 -> v8`.
/// ordered list of migrations. migration at index `i` upgrades the schema from version `i` to `i+1`.
const MIGRATIONS: &[(&str, Migration)] = &[
    (
        "stable_firehose_cursors",
        Migration::Atomic(v1::stable_firehose_cursors),
    ),
    (
        "repo_state_root_commit",
        Migration::Atomic(v2::repo_state_root_commit),
    ),
    (
        "firehose_source_is_pds",
        Migration::Atomic(v3::firehose_source_is_pds),
    ),
    (
        "repo_state_active",
        Migration::Atomic(v4::repo_state_active),
    ),
    ("pds_meta_layout", Migration::Atomic(v5::pds_meta_layout)),
    (
        "rebuild_pds_account_counts",
        Migration::Atomic(v6::rebuild_pds_account_counts),
    ),
    (
        "repo_state_event_clocks",
        Migration::Atomic(v7::repo_state_event_clocks),
    ),
    (
        "rebuild_lifecycle_counts",
        Migration::Atomic(v8::rebuild_lifecycle_counts),
    ),
    (
        "migrate_excludes_and_pds_keys",
        Migration::Atomic(v9::migrate_v9),
    ),
    (
        "inline_record_bodies",
        Migration::Chunked {
            passes: v10::PASSES,
            finalize: Some(v10::retire_legacy_stores),
        },
    ),
    (
        "canonicalize_pds_status_hosts",
        Migration::Chunked {
            passes: v11::PASSES,
            finalize: None,
        },
    ),
];

pub(crate) const LATEST_VERSION: u64 = MIGRATIONS.len() as u64;

/// resume cursors for chunked passes, in the `counts` keyspace:
/// `mig|{version}|{pass}` -> last committed key.
///
/// the `counts` compaction filter only drops `r|` and old `d|` entries, so these
/// survive compaction.
const MIGRATION_CURSOR_PREFIX: &[u8] = b"mig|";

/// applied markers for migrations, in the `counts` keyspace:
/// `mig_applied|{name}` -> `b"1"`.
const MIGRATION_APPLIED_PREFIX: &[u8] = b"mig_applied|";

fn pass_cursor_key(version: usize, pass: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(MIGRATION_CURSOR_PREFIX.len() + 8 + 1 + pass.len());
    key.extend_from_slice(MIGRATION_CURSOR_PREFIX);
    key.extend_from_slice(&(version as u64).to_be_bytes());
    key.push(SEP);
    key.extend_from_slice(pass.as_bytes());
    key
}

fn migration_applied_key(name: &str) -> Vec<u8> {
    let mut key = Vec::with_capacity(MIGRATION_APPLIED_PREFIX.len() + name.len());
    key.extend_from_slice(MIGRATION_APPLIED_PREFIX);
    key.extend_from_slice(name.as_bytes());
    key
}

/// test helper: rewind the db to `version` as if later migrations never ran.
/// opening a fresh test db already runs (and marks) every migration, so
/// simulating an old database means clearing the applied markers too.
#[cfg(test)]
pub(crate) fn rewind_version_for_test(db: &Db, version: u64) -> Result<()> {
    let mut batch = db.inner.batch();
    for (name, _) in MIGRATIONS {
        batch.remove(&db.counts, migration_applied_key(name));
    }
    batch.insert(&db.counts, VERSIONING_KEY, version.to_be_bytes());
    batch.commit().into_diagnostic()
}

fn is_migration_applied(db: &Db, index: usize, name: &str, stored_version: u64) -> Result<bool> {
    let key = migration_applied_key(name);
    if db.counts.contains_key(&key).into_diagnostic()? {
        return Ok(true);
    }

    // Legacy versions predate per-migration applied keys. Their global version
    // proves ordinary migrations ran, while feature-gated migrations need a
    // cheap migration-specific proof: reduced-feature builds advanced the old
    // version even when they skipped the work.
    if (index as u64) < stored_version {
        return match name {
            "rebuild_lifecycle_counts" => v8::legacy_migration_applied(db),
            // The old v9 marker is trusted; v10 safely and boundedly finishes
            // any work a reduced-feature open skipped.
            "migrate_excludes_and_pds_keys" => Ok(true),
            // no intermediate v10 layout shipped. accepting a v10 global
            // version without this migration's marker would silently bless a
            // branch-local draft as the final schema.
            "inline_record_bodies" => {
                miette::bail!("unsupported pre-release v10 database; rebuild it from v9")
            }
            _ => Ok(true),
        };
    }

    Ok(false)
}

fn read_version(db: &Db) -> Result<u64> {
    db.counts
        .get(VERSIONING_KEY)
        .into_diagnostic()?
        .map(|r| {
            r.as_ref()
                .try_into()
                .into_diagnostic()
                .wrap_err("db version key expected to be 8 bytes")
                .map(u64::from_be_bytes)
        })
        .transpose()
        .map(|v| v.unwrap_or(0))
}

struct ChunkOutcome {
    /// entries visited and committed.
    seen: usize,
    /// the scan reached the end of the keyspace.
    exhausted: bool,
}

/// visit one chunk of `pass`, committing the transformed entries and the
/// advanced resume cursor as one atomic batch.
fn run_chunk(db: &Db, cursor_key: &[u8], ks: &Keyspace, pass: &Pass) -> Result<ChunkOutcome> {
    let resume = db.counts.get(cursor_key).into_diagnostic()?;
    let start = resume
        .as_ref()
        .map_or(Bound::Unbounded, |k| Bound::Excluded(k.to_vec()));

    let mut batch = db.inner.batch();
    let mut seen = 0usize;
    let mut bytes = 0usize;
    let mut last = None;
    let mut exhausted = true;

    {
        // a fresh snapshot per chunk keeps the scan consistent while the batch is
        // staged, without pinning superseded versions for the whole pass.
        let snapshot = db.inner.snapshot();
        for guard in snapshot.range(ks, (start, Bound::<Vec<u8>>::Unbounded)) {
            // checked before visiting, so breaking here proves an entry remains.
            // a chunk always visits at least one entry, so neither a degenerate
            // budget nor an entry larger than the byte budget can stall the pass
            // or make an unfinished scan look exhausted.
            if seen > 0 && (seen >= pass.budget.entries || bytes >= pass.budget.bytes) {
                exhausted = false;
                break;
            }
            let (key, value) = guard.into_inner().into_diagnostic()?;
            let staged_bytes = (pass.visit)(db, &mut batch, ks, &key, &value)
                .wrap_err_with(|| format!("migration pass {} failed on an entry", pass.name))?;
            bytes += staged_bytes;
            last = Some(key);
            seen += 1;
        }
    }

    let Some(last) = last else {
        // nothing left to do; leave the cursor for `run` to clear with the
        // version bump
        return Ok(ChunkOutcome {
            seen: 0,
            exhausted: true,
        });
    };

    batch.insert(&db.counts, cursor_key, last);
    batch.commit().into_diagnostic()?;

    Ok(ChunkOutcome { seen, exhausted })
}

/// run one pass to completion, resuming from its stored cursor if present.
fn run_pass(db: &Db, version: usize, pass: &Pass) -> Result<bool> {
    let ks = match db.keyspace_by_name(pass.scan) {
        Some(ks) => Some(ks),
        // a keyspace dropped from the registry by a later schema may still
        // exist on disk; open it so the pass can finish migrating its data
        None if db.inner.keyspace_exists(pass.scan) => Some(
            db.inner
                .keyspace(pass.scan, Default::default)
                .into_diagnostic()?,
        ),
        None => None,
    };
    let Some(ks) = ks else {
        if pass.optional_on_absent {
            // the source layout never existed on this database, so there is
            // genuinely nothing to migrate
            tracing::debug!(
                "db: migration pass {} found no {} keyspace, nothing to migrate",
                pass.name,
                pass.scan
            );
            return Ok(true);
        }
        // keyspace is not compiled into this build, so there is nothing to migrate
        tracing::debug!(
            "db: migration pass {} skipped, keyspace {} not in this build",
            pass.name,
            pass.scan
        );
        return Ok(false);
    };
    let cursor_key = pass_cursor_key(version, pass.name);

    if db.counts.contains_key(&cursor_key).into_diagnostic()? {
        tracing::info!("db: resuming migration pass {} from its cursor", pass.name);
    }

    let mut total: u64 = 0;
    loop {
        let outcome = run_chunk(db, &cursor_key, &ks, pass)?;
        total += outcome.seen as u64;
        if outcome.seen > 0 {
            tracing::info!(
                "db: migration pass {} committed {total} entries so far",
                pass.name
            );
        }
        if outcome.exhausted {
            break;
        }
    }

    tracing::info!(
        "db: migration pass {} complete ({total} entries)",
        pass.name
    );
    Ok(true)
}

/// run all pending database migrations in order.
///
/// an atomic migration commits its write set and applied marker together. a
/// chunked migration commits each chunk with its resume cursor, durably runs
/// any finalizer, then commits cursor cleanup with the applied marker. the
/// global version is derived from the contiguous markers afterward. a crash
/// before the marker therefore resumes from the durable pass cursors.
///
/// safe to run on a fresh database (all migrations are no-ops when no relevant
/// data exists). called during [`Db::open`].
pub(super) fn run(db: &Db) -> Result<()> {
    let stored_version = read_version(db)?;
    if stored_version > LATEST_VERSION {
        miette::bail!(
            "database schema version {stored_version} is newer than supported version {LATEST_VERSION}"
        );
    }
    for (i, (name, migration)) in MIGRATIONS.iter().enumerate() {
        if is_migration_applied(db, i, name, stored_version)? {
            let key = migration_applied_key(name);
            if !db.counts.contains_key(&key).into_diagnostic()? {
                db.counts.insert(key, b"1").into_diagnostic()?;
                tracing::info!("db: adopted legacy applied marker for {name}");
            }
            tracing::debug!("db: migration {name} already applied");
            continue;
        }

        tracing::info!("db: running migration {name} (v{i} -> v{})", i + 1);
        match migration {
            Migration::Atomic(f) => {
                let mut batch = db.inner.batch();
                if f(db, &mut batch)? {
                    batch.insert(&db.counts, migration_applied_key(name), b"1");
                    batch.commit().into_diagnostic()?;
                    tracing::info!("db: migration {name} complete");
                } else {
                    tracing::info!(
                        "db: migration {name} skipped (required keyspace not in this build)"
                    );
                }
            }
            Migration::Chunked { passes, finalize } => {
                let mut all_passed = !(passes.is_empty() && finalize.is_none());
                for pass in *passes {
                    if !run_pass(db, i, pass)? {
                        all_passed = false;
                        break;
                    }
                }
                if all_passed
                    && let Some(finalize) = finalize
                    && !finalize(db)?
                {
                    all_passed = false;
                }
                if all_passed {
                    let mut batch = db.inner.batch();
                    for pass in *passes {
                        batch.remove(&db.counts, pass_cursor_key(i, pass.name));
                    }
                    batch.insert(&db.counts, migration_applied_key(name), b"1");
                    batch.commit().into_diagnostic()?;
                    tracing::info!("db: migration {name} complete");
                } else {
                    tracing::info!(
                        "db: migration {name} skipped (required keyspace not in this build)"
                    );
                }
            }
        }
    }

    // Advance VERSIONING_KEY to highest contiguous applied migration version
    let mut max_contiguous_version = 0u64;
    for (i, (name, _)) in MIGRATIONS.iter().enumerate() {
        if is_migration_applied(db, i, name, stored_version)? {
            max_contiguous_version = (i + 1) as u64;
        } else {
            break;
        }
    }
    if max_contiguous_version != stored_version {
        let mut batch = db.inner.batch();
        batch.insert(
            &db.counts,
            VERSIONING_KEY,
            max_contiguous_version.to_be_bytes(),
        );
        batch.commit().into_diagnostic()?;
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    const TEST_PREFIX: &[u8] = b"chunktest|";
    const MIGRATED_PREFIX: &[u8] = b"chunkdone|";

    /// rewrites `chunktest|{n}` to `chunkdone|{n}`, keeping the value.
    ///
    /// idempotent: already-migrated keys carry the new prefix and are ignored.
    /// note the output sorts *before* the input (`chunkd` < `chunkt`), which is
    /// the usual shape for a key rewrite and means output is never rescanned.
    fn rewrite_prefix(
        db: &Db,
        batch: &mut OwnedWriteBatch,
        scanned: &Keyspace,
        key: &[u8],
        value: &[u8],
    ) -> Result<usize> {
        let Some(suffix) = key.strip_prefix(TEST_PREFIX) else {
            return Ok(0);
        };
        let mut new_key = MIGRATED_PREFIX.to_vec();
        new_key.extend_from_slice(suffix);
        let _ = db;
        let staged_bytes = new_key.len() + value.len();
        batch.insert(scanned, new_key, value);
        batch.remove(scanned, key);
        Ok(staged_bytes)
    }

    fn test_pass(entries: usize) -> Pass {
        Pass {
            name: "rewrite_prefix",
            scan: "cursors",
            visit: rewrite_prefix,
            budget: ChunkBudget {
                entries,
                bytes: usize::MAX,
            },
            optional_on_absent: false,
        }
    }

    fn open_db() -> Result<(tempfile::TempDir, Db)> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let cfg = crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let db = Db::open(&cfg)?;
        Ok((tmp, db))
    }

    fn seed(db: &Db, count: usize) -> Result<()> {
        let mut batch = db.inner.batch();
        for i in 0..count {
            let mut key = TEST_PREFIX.to_vec();
            key.extend_from_slice(&(i as u64).to_be_bytes());
            batch.insert(&db.cursors, key, (i as u64).to_be_bytes());
        }
        batch.commit().into_diagnostic()?;
        Ok(())
    }

    fn migrated(db: &Db) -> Result<Vec<(u64, u64)>> {
        db.cursors
            .prefix(MIGRATED_PREFIX)
            .map(|guard| {
                let (k, v) = guard.into_inner().into_diagnostic()?;
                let n =
                    u64::from_be_bytes(k[MIGRATED_PREFIX.len()..].try_into().into_diagnostic()?);
                let v = u64::from_be_bytes(v.as_ref().try_into().into_diagnostic()?);
                Ok((n, v))
            })
            .collect()
    }

    fn expected(count: usize) -> Vec<(u64, u64)> {
        (0..count as u64).map(|i| (i, i)).collect()
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn v0_database_migrates_all_early_representation_boundaries() -> Result<()> {
        use crate::config::Config;
        use crate::db::{
            keys,
            types::{DbTid, DidKey},
        };
        use crate::types::v2::RepoStatus as V2RepoStatus;
        use jacquard_common::CowStr;
        use jacquard_common::types::{
            string::{Did, Handle},
            tid::Tid,
        };

        let tmp = tempfile::tempdir().into_diagnostic()?;
        let cfg = Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let did = Did::new_static("did:web:migration.test").into_diagnostic()?;
        let repo_key = keys::repo_key(&did);
        let old_cursor_url = url::Url::parse("wss://relay.example:7443/x").into_diagnostic()?;
        let old_cursor_key = [
            keys::FIREHOSE_CURSOR_PREFIX,
            old_cursor_url.as_str().as_bytes(),
        ]
        .concat();
        let new_cursor_key = keys::v1::firehose_cursor_key_from_url(&old_cursor_url);
        let relay_source = "wss://bsky.network";
        let pds_source = "wss://pds.example";

        {
            let db = Db::open(&cfg)?;
            let old_state = v2::OldRepoState {
                status: V2RepoStatus::Error("stinkpot".into()),
                rev: Some(DbTid::from(&Tid::new("3jzfcijpj2z2a").unwrap())),
                data: Some(jacquard_repo::mst::util::compute_cid(b"root")?),
                last_message_time: Some(-1),
                last_updated_at: 42,
                tracked: true,
                index_id: 7,
                signing_key: Some(DidKey(std::borrow::Cow::Borrowed(&[0xed, 1, 2, 3]))),
                pds: Some(CowStr::Borrowed("https://pds.example")),
                handle: Some(Handle::new_static("alice.test").into_diagnostic()?),
            };
            let mut batch = db.inner.batch();
            batch.insert(
                &db.repos,
                &repo_key,
                rmp_serde::to_vec(&old_state).into_diagnostic()?,
            );
            batch.insert(&db.cursors, &old_cursor_key, 42i64.to_be_bytes());
            batch.insert(&db.crawler, keys::firehose_source_key(relay_source), []);
            batch.insert(&db.crawler, keys::firehose_source_key(pds_source), []);
            batch.insert(&db.filter, b"pt|pds.example", b"trusted");
            batch.insert(&db.filter, b"pb|banned.example", []);
            batch.insert(
                &db.counts,
                keys::count_keyspace_key("p|stale.example"),
                99u64.to_be_bytes(),
            );
            batch.commit().into_diagnostic()?;
            rewind_version_for_test(&db, 0)?;
            db.persist()?;
        }

        let db = Db::open(&cfg)?;
        assert!(!db.cursors.contains_key(&old_cursor_key).into_diagnostic()?);
        let cursor = db
            .cursors
            .get(&new_cursor_key)
            .into_diagnostic()?
            .expect("v1 cursor");
        assert_eq!(
            i64::from_be_bytes(cursor.as_ref().try_into().into_diagnostic()?),
            42
        );

        for (source, is_pds) in [(relay_source, false), (pds_source, true)] {
            let encoded = db
                .crawler
                .get(keys::firehose_source_key(source))
                .into_diagnostic()?
                .expect("v3 source metadata");
            let metadata: crate::db::FirehoseSourceMeta =
                rmp_serde::from_slice(&encoded).into_diagnostic()?;
            assert_eq!(metadata.is_pds, is_pds);
        }

        let state_bytes = db
            .repos
            .get(&repo_key)
            .into_diagnostic()?
            .expect("migrated repo state");
        let state = crate::db::deser_repo_state(&state_bytes)?;
        assert!(state.active);
        assert!(matches!(
            &state.status,
            crate::types::RepoStatus::Error(error) if error == "stinkpot"
        ));
        let root = state.root.as_ref().expect("migrated commit");
        assert_eq!(root.version, -1);
        assert_eq!(root.rev, DbTid::from(&Tid::new("3jzfcijpj2z2a").unwrap()));
        assert_eq!(root.data, jacquard_repo::mst::util::compute_cid(b"root")?);
        assert!(root.prev.is_none());
        assert!(root.sig.is_empty());
        assert_eq!(state.last_message_time, Some(-1));
        assert_eq!(state.last_identity_time, None);
        assert_eq!(state.last_account_time, None);
        assert_eq!(state.last_updated_at, 42);
        assert_eq!(
            state.signing_key.as_ref().map(|key| key.0.as_ref()),
            Some([0xed, 1, 2, 3].as_slice())
        );
        assert_eq!(state.pds.as_deref(), Some("https://pds.example"));
        assert_eq!(
            state.handle.as_ref().map(Handle::as_str),
            Some("alice.test")
        );
        let metadata = crate::db::deser_repo_meta(
            &db.repo_metadata
                .get(keys::repo_metadata_key(&did))
                .into_diagnostic()?
                .expect("v4 repo metadata"),
        )?;
        assert!(metadata.tracked);
        assert_eq!(metadata.index_id, 7);

        assert_eq!(
            db.filter
                .get(crate::db::pds_meta::pds_tier_key("pds.example"))
                .into_diagnostic()?
                .as_deref(),
            Some(b"trusted".as_slice())
        );
        assert!(
            !db.filter
                .contains_key(crate::db::pds_meta::pds_status_key("banned.example"))
                .into_diagnostic()?
        );
        assert!(
            db.filter
                .contains_key(crate::db::pds_meta::pds_ban_key("banned.example"))
                .into_diagnostic()?
        );
        assert!(
            !db.counts
                .contains_key(keys::count_keyspace_key("p|stale.example"))
                .into_diagnostic()?
        );
        let account_count = db
            .counts
            .get(keys::count_keyspace_key("p|pds.example"))
            .into_diagnostic()?
            .expect("v6 PDS account count");
        assert_eq!(
            u64::from_be_bytes(account_count.as_ref().try_into().into_diagnostic()?),
            1
        );

        let version = db
            .counts
            .get(VERSIONING_KEY)
            .into_diagnostic()?
            .expect("schema version");
        assert_eq!(
            u64::from_be_bytes(version.as_ref().try_into().into_diagnostic()?),
            LATEST_VERSION
        );
        Ok(())
    }

    #[test]
    fn chunked_pass_rewrites_every_entry_across_many_chunks() -> Result<()> {
        let (_tmp, db) = open_db()?;
        seed(&db, 50)?;

        let pass = test_pass(7);
        run_pass(&db, 99, &pass)?;

        assert_eq!(migrated(&db)?, expected(50));
        assert_eq!(db.cursors.prefix(TEST_PREFIX).count(), 0);
        Ok(())
    }

    #[test]
    fn interrupted_pass_resumes_and_matches_an_uninterrupted_run() -> Result<()> {
        let (_tmp, db) = open_db()?;
        seed(&db, 50)?;

        let pass = test_pass(7);
        let cursor_key = pass_cursor_key(99, pass.name);
        let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");

        // one chunk only, standing in for a crash mid-pass
        let first = run_chunk(&db, &cursor_key, &ks, &pass)?;
        assert_eq!(first.seen, 7);
        assert!(!first.exhausted);
        assert_eq!(migrated(&db)?, expected(7));
        assert!(db.counts.contains_key(&cursor_key).into_diagnostic()?);

        // reopening resumes from the cursor rather than restarting
        run_pass(&db, 99, &pass)?;

        assert_eq!(migrated(&db)?, expected(50));
        assert_eq!(db.cursors.prefix(TEST_PREFIX).count(), 0);
        Ok(())
    }

    #[test]
    fn completed_pass_is_a_cheap_no_op_when_rerun() -> Result<()> {
        let (_tmp, db) = open_db()?;
        seed(&db, 20)?;

        let pass = test_pass(7);
        run_pass(&db, 99, &pass)?;
        let after_first = migrated(&db)?;

        // the cursor survives until the version bump, so a rerun scans nothing
        let cursor_key = pass_cursor_key(99, pass.name);
        let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");
        let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?;
        assert_eq!(outcome.seen, 0);
        assert!(outcome.exhausted);

        assert_eq!(migrated(&db)?, after_first);
        Ok(())
    }

    #[test]
    fn visiting_migrated_entries_again_is_a_fixpoint() -> Result<()> {
        let (_tmp, db) = open_db()?;
        seed(&db, 20)?;

        let pass = test_pass(7);
        run_pass(&db, 99, &pass)?;
        let after_first = migrated(&db)?;

        // drop the cursor to force a full rescan over already-migrated entries,
        // the worst case the idempotency contract has to survive
        db.counts
            .remove(pass_cursor_key(99, pass.name))
            .into_diagnostic()?;
        run_pass(&db, 99, &pass)?;

        assert_eq!(migrated(&db)?, after_first);
        Ok(())
    }

    /// fails partway through a chunk, standing in for a crash mid-chunk.
    fn fail_partway(
        db: &Db,
        batch: &mut OwnedWriteBatch,
        scanned: &Keyspace,
        key: &[u8],
        value: &[u8],
    ) -> Result<usize> {
        let Some(suffix) = key.strip_prefix(TEST_PREFIX) else {
            return Ok(0);
        };
        let n = u64::from_be_bytes(suffix.try_into().into_diagnostic()?);
        if n == 4 {
            miette::bail!("simulated failure at entry 4");
        }
        rewrite_prefix(db, batch, scanned, key, value)
    }
    #[test]
    fn a_chunk_that_fails_partway_commits_nothing() -> Result<()> {
        let (_tmp, db) = open_db()?;
        seed(&db, 50)?;

        let pass = Pass {
            visit: fail_partway,
            ..test_pass(7)
        };
        let cursor_key = pass_cursor_key(99, pass.name);
        let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");

        assert!(run_chunk(&db, &cursor_key, &ks, &pass).is_err());

        // entries 0..=3 were staged before the failure, but the batch is dropped
        // uncommitted, so neither they nor the cursor survive
        assert_eq!(migrated(&db)?, vec![]);
        assert!(!db.counts.contains_key(&cursor_key).into_diagnostic()?);
        assert_eq!(db.cursors.prefix(TEST_PREFIX).count(), 50);

        // retrying with a working visit starts over from the beginning
        run_pass(&db, 99, &test_pass(7))?;
        assert_eq!(migrated(&db)?, expected(50));
        Ok(())
    }

    #[test]
    fn byte_budget_ends_a_chunk_before_the_entry_budget() -> Result<()> {
        let (_tmp, db) = open_db()?;
        seed(&db, 50)?;

        // each entry is 26 scanned bytes (10-byte prefix + 8-byte key + 8-byte
        // value), and the budget is checked before visiting, so a 60-byte budget
        // admits 3 entries (26, 52, 78) before it trips
        let pass = Pass {
            budget: ChunkBudget {
                entries: usize::MAX,
                bytes: 60,
            },
            ..test_pass(0)
        };
        let cursor_key = pass_cursor_key(99, pass.name);
        let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");

        let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?;
        assert_eq!(outcome.seen, 3);
        assert!(!outcome.exhausted);

        run_pass(&db, 99, &pass)?;
        assert_eq!(migrated(&db)?, expected(50));
        Ok(())
    }

    #[test]
    fn a_degenerate_budget_still_makes_progress() -> Result<()> {
        let (_tmp, db) = open_db()?;
        seed(&db, 5)?;

        // a zero budget would otherwise trip before visiting anything, leaving
        // the chunk with no cursor to advance and reporting itself exhausted
        let pass = Pass {
            budget: ChunkBudget {
                entries: 0,
                bytes: 0,
            },
            ..test_pass(0)
        };
        let cursor_key = pass_cursor_key(99, pass.name);
        let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");

        let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?;
        assert_eq!(outcome.seen, 1);
        assert!(!outcome.exhausted);

        run_pass(&db, 99, &pass)?;
        assert_eq!(migrated(&db)?, expected(5));
        Ok(())
    }

    #[test]
    fn pass_on_a_keyspace_missing_from_this_build_is_skipped() -> Result<()> {
        let (_tmp, db) = open_db()?;

        let pass = Pass {
            scan: "not_a_keyspace",
            ..test_pass(7)
        };
        run_pass(&db, 99, &pass)?;

        let cursor_key = pass_cursor_key(99, pass.name);
        assert!(!db.counts.contains_key(&cursor_key).into_diagnostic()?);
        Ok(())
    }

    #[test]
    fn cursor_keys_are_distinct_per_version_and_pass() {
        assert_ne!(pass_cursor_key(1, "a"), pass_cursor_key(2, "a"));
        assert_ne!(pass_cursor_key(1, "a"), pass_cursor_key(1, "b"));
        assert!(pass_cursor_key(1, "a").starts_with(MIGRATION_CURSOR_PREFIX));
    }

    #[test]
    fn staged_byte_budget_bounds_chunk_write_size() -> Result<()> {
        let (_tmp, db) = open_db()?;
        let mut batch = db.inner.batch();
        // Insert small input keys/values that expand into large staged outputs
        for i in 0..10 {
            let mut key = TEST_PREFIX.to_vec();
            key.extend_from_slice(&(i as u64).to_be_bytes());
            batch.insert(&db.cursors, key, vec![0u8; 1000]); // 1 KB payload
        }
        batch.commit().into_diagnostic()?;

        // Set byte budget to 2500 bytes. Each output staged is ~1018 bytes.
        // First entry: 1018 bytes. Second entry: 2036 bytes. Third entry: 3054 bytes (>= 2500).
        // 4th check trips because 3054 >= 2500. So outcome.seen == 3.
        let pass = Pass {
            budget: ChunkBudget {
                entries: usize::MAX,
                bytes: 2500,
            },
            ..test_pass(0)
        };
        let cursor_key = pass_cursor_key(99, pass.name);
        let ks = db.keyspace_by_name("cursors").expect("cursors keyspace");

        let outcome = run_chunk(&db, &cursor_key, &ks, &pass)?;
        assert_eq!(outcome.seen, 3);
        assert!(!outcome.exhausted);
        Ok(())
    }

    #[test]
    fn applied_migration_keys_track_per_migration_status() -> Result<()> {
        let (_tmp, db) = open_db()?;
        assert!(
            db.counts
                .contains_key(&migration_applied_key("stable_firehose_cursors"))
                .into_diagnostic()?
        );
        Ok(())
    }

    #[test]
    fn branch_local_v10_without_the_final_marker_is_rejected() -> Result<()> {
        let tmp = tempfile::tempdir().into_diagnostic()?;
        let cfg = crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        {
            let db = Db::open(&cfg)?;
            rewind_version_for_test(&db, 10)?;
            db.persist()?;
        }

        let err = match Db::open(&cfg) {
            Ok(_) => panic!("an unmarked branch-local v10 must not be adopted"),
            Err(err) => err,
        };
        assert!(format!("{err:?}").contains("unsupported pre-release v10"));
        Ok(())
    }

    #[cfg(feature = "indexer")]
    #[test]
    fn legacy_marker_adoption_does_not_rerun_lifecycle_rebuild() -> Result<()> {
        use jacquard_common::types::string::Did;

        let tmp = tempfile::tempdir().into_diagnostic()?;
        let cfg = crate::config::Config {
            database_path: tmp.path().to_path_buf(),
            ..Default::default()
        };
        let repo = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?;
        let stale = Did::new_static("did:plc:yk4q3id7id6p5z3bypvshc64").into_diagnostic()?;
        let mut stale_gauge = crate::db::keys::COUNT_GAUGE_PREFIX.to_vec();
        stale_gauge.extend_from_slice(&crate::db::keys::repo_key(&stale));

        {
            let db = Db::open(&cfg)?;
            let mut batch = db.inner.batch();
            batch.insert(
                &db.repos,
                crate::db::keys::repo_key(&repo),
                crate::db::ser_repo_state(&crate::types::RepoState::synced())?,
            );
            let mut gauge = crate::db::keys::COUNT_GAUGE_PREFIX.to_vec();
            gauge.extend_from_slice(&crate::db::keys::repo_key(&repo));
            let synced =
                crate::db::lifecycle_counts::encode_gauge(crate::types::GaugeState::Synced);
            batch.insert(&db.counts, gauge, synced);
            // v8 would clear this stale-but-valid row. retaining it proves the
            // legacy global version was adopted rather than rebuilt.
            batch.insert(&db.counts, &stale_gauge, synced);
            batch.commit().into_diagnostic()?;
            // v9 is the last schema that existed before this change.
            rewind_version_for_test(&db, 9)?;
            db.persist()?;
        }

        let db = Db::open(&cfg)?;
        assert!(db.counts.contains_key(stale_gauge).into_diagnostic()?);
        assert!(
            db.counts
                .contains_key(migration_applied_key("rebuild_lifecycle_counts"))
                .into_diagnostic()?
        );
        Ok(())
    }
}