#![cfg(all(
    feature = "var-collections",
    feature = "typed-tree",
    feature = "armour"
))]

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

use armdb::MigrateAction;
use armdb::armour::{CollectionInfo, Db, TypedMigration};
use armdb::{CollectionMeta, Config, NoHook, RapiraCodec, TypedWriteHook};
use armour_core::GetType;
use rapira::Rapira;
use tempfile::tempdir;

// ---------------------------------------------------------------------------
// Domain type
// ---------------------------------------------------------------------------

#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct Item {
    value: i64,
}

impl CollectionMeta for Item {
    type SelfId = [u8; 8];
    const NAME: &'static str = "replay_init_items";
    const VERSION: u16 = 1;
}

fn item_key(id: u64) -> [u8; 8] {
    id.to_be_bytes()
}

// ---------------------------------------------------------------------------
// Hook that counts on_init and on_write separately
// ---------------------------------------------------------------------------

struct InitCounter {
    inits: AtomicUsize,
    writes: AtomicUsize,
}

impl InitCounter {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            inits: AtomicUsize::new(0),
            writes: AtomicUsize::new(0),
        })
    }
}

// Arc<InitCounter> can't impl TypedWriteHook directly (not a local type for
// the trait impl), so we use a thin wrapper struct.
#[derive(Clone)]
struct InitCounterHook(Arc<InitCounter>);

impl TypedWriteHook<[u8; 8], Item> for InitCounterHook {
    const NEEDS_OLD_VALUE: bool = false;
    const NEEDS_INIT: bool = true;

    fn on_write(&self, _key: &[u8; 8], _old: Option<&Item>, _new: Option<&Item>) {
        self.0.writes.fetch_add(1, AtomicOrdering::Relaxed);
    }

    fn on_init(&self, _key: &[u8; 8], _value: &Item) {
        self.0.inits.fetch_add(1, AtomicOrdering::Relaxed);
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

/// When no migration runs, Db::open_var_typed_tree calls replay_init() so that
/// every live entry is surfaced to hooks with NEEDS_INIT = true.
#[test]
fn db_open_var_typed_tree_calls_replay_init_when_no_migration() {
    let dir = tempdir().unwrap();

    // Phase 1: populate 3 entries without any hook.
    {
        let db = Db::open_test(dir.path()).unwrap();
        let tree = db
            .open_var_typed_tree::<Item, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        for i in 0u64..3 {
            tree.put(&item_key(i), &Item { value: i as i64 }).unwrap();
        }
        // Db and tree are dropped here — closes files.
    }

    // Phase 2: reopen with the counting hook, no migrations.
    let counter = InitCounter::new();
    let hook = InitCounterHook(counter.clone());

    let db = Db::open_test(dir.path()).unwrap();
    let _tree = db
        .open_var_typed_tree::<Item, RapiraCodec, _>(Config::test(), hook, &[])
        .unwrap();

    assert_eq!(
        counter.inits.load(AtomicOrdering::Relaxed),
        3,
        "on_init should fire once per live entry"
    );
    assert_eq!(
        counter.writes.load(AtomicOrdering::Relaxed),
        0,
        "on_write must not fire during replay_init"
    );
}

/// Same guarantee for open_var_typed_map.
#[test]
fn db_open_var_typed_map_calls_replay_init_when_no_migration() {
    let dir = tempdir().unwrap();

    // Phase 1: populate 3 entries without any hook.
    {
        let db = Db::open_test(dir.path()).unwrap();
        let map = db
            .open_var_typed_map::<Item, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        for i in 0u64..3 {
            map.put(&item_key(i), &Item { value: i as i64 }).unwrap();
        }
        // Db and map are dropped here — closes files.
    }

    // Phase 2: reopen with the counting hook, no migrations.
    let counter = InitCounter::new();
    let hook = InitCounterHook(counter.clone());

    let db = Db::open_test(dir.path()).unwrap();
    let _map = db
        .open_var_typed_map::<Item, RapiraCodec, _>(Config::test(), hook, &[])
        .unwrap();

    assert_eq!(
        counter.inits.load(AtomicOrdering::Relaxed),
        3,
        "on_init should fire once per live entry"
    );
    assert_eq!(
        counter.writes.load(AtomicOrdering::Relaxed),
        0,
        "on_write must not fire during replay_init"
    );
}

// ---------------------------------------------------------------------------
// Additional types for the "skips replay_init when migration ran" tests.
// These use a distinct NAME to avoid sharing state with the tests above.
// ---------------------------------------------------------------------------

/// V2 type used for both phases. Data is written and read using the same
/// VERSION=2 path. To simulate a migration trigger, db.info is patched to
/// claim stored.version=1 after Phase 1, so run_migration fires on Phase 2.
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct MigrateItemV2 {
    value: i64,
}

impl CollectionMeta for MigrateItemV2 {
    type SelfId = [u8; 8];
    const NAME: &'static str = "migrate_replay_items";
    const VERSION: u16 = 2;
}

// Hook impl for MigrateItemV2 — the same InitCounterHook wrapper reused.
impl TypedWriteHook<[u8; 8], MigrateItemV2> for InitCounterHook {
    const NEEDS_OLD_VALUE: bool = false;
    const NEEDS_INIT: bool = true;

    fn on_write(&self, _key: &[u8; 8], _old: Option<&MigrateItemV2>, _new: Option<&MigrateItemV2>) {
        self.0.writes.fetch_add(1, AtomicOrdering::Relaxed);
    }

    fn on_init(&self, _key: &[u8; 8], _value: &MigrateItemV2) {
        self.0.inits.fetch_add(1, AtomicOrdering::Relaxed);
    }
}

/// Identity migration: keep every entry unchanged.
fn identity_migration(_k: &[u8; 8], _v: &MigrateItemV2) -> MigrateAction<MigrateItemV2> {
    MigrateAction::Keep
}

// ---------------------------------------------------------------------------
// Tests: replay_init is skipped when a migration ran
// ---------------------------------------------------------------------------

/// When a migration runs, on_init fires exactly once per live entry (from
/// inside migrate), and replay_init must NOT fire a second time.
///
/// Setup: populate 3 entries with V2 (no hook), then patch db.info to claim
/// stored.version=1. On reopen with migration (1→2) + counting hook, Db
/// detects the version mismatch, calls tree.migrate() which fires on_init
/// exactly once per live entry, and must NOT call replay_init() afterward.
#[test]
fn db_open_var_typed_tree_skips_replay_init_when_migration_ran() {
    let dir = tempdir().unwrap();

    // Phase 1: populate 3 entries at the VERSION=2 path, no hook.
    // After closing, patch db.info to say stored.version=1, simulating the
    // state where data exists on disk but was last saved under an older schema
    // version.
    {
        let db = Db::open_test(dir.path()).unwrap();
        let tree = db
            .open_var_typed_tree::<MigrateItemV2, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        for i in 0u64..3 {
            tree.put(&item_key(i), &MigrateItemV2 { value: i as i64 })
                .unwrap();
        }
        // Patch: make db.info report version=1 for this collection so that
        // run_migration sees stored.version(1) != meta.version(2) on the next open.
        db.db_info
            .update(|info| {
                info.collections.insert(
                    MigrateItemV2::NAME.to_owned(),
                    CollectionInfo {
                        version: 1,
                        typ_hash: 0,
                        seq: 3,
                    },
                );
            })
            .expect("test persist");
        // Drop closes all file handles and flushes db.info.
    }

    // Phase 2: reopen with migration (1→2) and counting hook.
    // run_migration finds stored.version=1 != meta.version=2 and a matching
    // migration entry, calls tree.migrate() which fires on_init once per entry.
    // Db must NOT additionally call replay_init().
    let counter = InitCounter::new();
    let hook = InitCounterHook(counter.clone());

    let migrations: &[TypedMigration<[u8; 8], MigrateItemV2>] = &[(1u16, identity_migration)];

    let db = Db::open_test(dir.path()).unwrap();
    let _tree = db
        .open_var_typed_tree::<MigrateItemV2, RapiraCodec, _>(Config::test(), hook, migrations)
        .unwrap();

    assert_eq!(
        counter.inits.load(AtomicOrdering::Relaxed),
        3,
        "on_init must fire exactly once per live entry (from migrate, not replay_init)"
    );
    assert_eq!(
        counter.writes.load(AtomicOrdering::Relaxed),
        0,
        "on_write must not fire during migration"
    );
}

/// Same guarantee for open_var_typed_map.
#[test]
fn db_open_var_typed_map_skips_replay_init_when_migration_ran() {
    let dir = tempdir().unwrap();

    // Phase 1: populate 3 entries at the VERSION=2 path, no hook.
    // Patch db.info to simulate stored.version=1 (pre-migration state).
    {
        let db = Db::open_test(dir.path()).unwrap();
        let map = db
            .open_var_typed_map::<MigrateItemV2, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        for i in 0u64..3 {
            map.put(&item_key(i), &MigrateItemV2 { value: i as i64 })
                .unwrap();
        }
        // Patch: make db.info report version=1 for this collection.
        db.db_info
            .update(|info| {
                info.collections.insert(
                    MigrateItemV2::NAME.to_owned(),
                    CollectionInfo {
                        version: 1,
                        typ_hash: 0,
                        seq: 3,
                    },
                );
            })
            .expect("test persist");
        // Drop closes all file handles and flushes db.info.
    }

    // Phase 2: reopen with migration (1→2) and counting hook.
    let counter = InitCounter::new();
    let hook = InitCounterHook(counter.clone());

    let migrations: &[TypedMigration<[u8; 8], MigrateItemV2>] = &[(1u16, identity_migration)];

    let db = Db::open_test(dir.path()).unwrap();
    let _map = db
        .open_var_typed_map::<MigrateItemV2, RapiraCodec, _>(Config::test(), hook, migrations)
        .unwrap();

    assert_eq!(
        counter.inits.load(AtomicOrdering::Relaxed),
        3,
        "on_init must fire exactly once per live entry (from migrate, not replay_init)"
    );
    assert_eq!(
        counter.writes.load(AtomicOrdering::Relaxed),
        0,
        "on_write must not fire during migration"
    );
}
