//! The **mem column** — the contract-reference column (ADR-011 §2):
//! the mem engine's column of the reference row set ([`crate::rows`]),
//! the first column bound to the real facade, and INV-7's basis —
//! every other engine's rows mirror this set.
//!
//! The factories (README §The factory contract): `open()` vends the
//! concrete [`alkblobs::Store`] facade over a fresh, isolated backing
//! store per call — the mem engine is `BTreeMap`-shaped and shares
//! nothing across instances, so isolation is by construction. Teardown
//! is a no-op (the mem engine holds no external state), which makes it
//! trivially idempotent and safe against abandoned instances — the
//! panicking-row path included, which the structural tests below
//! exercise. Mem never skips (README §SKIP posture — the SKIP
//! machinery stays unneeded until the SQL waves).
//!
//! The INV rows run against the reference `mem-only` factory (the kv
//! tier, no dispatch threshold in effect); the two constructor-mode
//! rows run against the mode factories they name — `mem-only`, and
//! `kv-only(mem)` at the column-declared threshold, which the kv-only
//! row discovers by probing.

#![allow(dead_code)]

use std::io::Cursor;
use std::sync::Arc;

use crate::factory::StoreFactory;
use crate::rows;
use alkblobs::{Error, Key, Store};

/// The reference factory: the mem-only mode (ADR-011 §4) — mem as the
/// sole engine, no dispatch threshold in effect. The column never
/// skips.
pub struct MemOnlyFactory;

impl StoreFactory for MemOnlyFactory {
    fn name(&self) -> &'static str {
        "mem-only"
    }

    async fn open(&self) -> Result<Store, Error> {
        Ok(Store::mem_only())
    }

    async fn teardown(&self) -> Result<(), Error> {
        Ok(())
    }
}

/// The kv-only(mem) factory (ADR-011 §4; ADR-012 §6.2): no large
/// tier; the threshold is the put-time rejection point the kv-only
/// row discovers by probing.
pub struct KvOnlyMemFactory {
    /// The rejection point this column's configuration declares. The
    /// kv-only row discovers it behaviorally and requires it
    /// materializable for its boundary legs.
    pub threshold: u64,
}

impl StoreFactory for KvOnlyMemFactory {
    fn name(&self) -> &'static str {
        "kv-only(mem)"
    }

    async fn open(&self) -> Result<Store, Error> {
        Ok(Store::kv_only_mem(self.threshold))
    }

    async fn teardown(&self) -> Result<(), Error> {
        Ok(())
    }
}

/// The column's declared kv-only threshold — small, so every leg of
/// the rejection row stays trivially materializable.
const KV_ONLY_THRESHOLD: u64 = 1024;

const MEM: MemOnlyFactory = MemOnlyFactory;
const KV_ONLY: KvOnlyMemFactory = KvOnlyMemFactory {
    threshold: KV_ONLY_THRESHOLD,
};

// ------------------------------------------------------------------
// The reference row set, run against the mem column's factories.
// ------------------------------------------------------------------

#[tokio::test]
async fn inv1_round_trip_puts_and_gets_are_byte_identical() {
    rows::round_trip_puts_and_gets_are_byte_identical(&MEM).await;
}

#[tokio::test]
async fn inv2_dedup_identical_content_is_one_pool_entry() {
    rows::dedup_identical_content_is_one_pool_entry(&MEM).await;
}

#[tokio::test]
async fn inv3_stage_hygiene_zero_residue_on_failure_one_entry_on_success() {
    rows::stage_hygiene_zero_residue_on_failure_exactly_one_entry_on_success(&MEM).await;
}

#[tokio::test]
async fn inv4_sweep_counts_are_exact_across_the_full_protection_set() {
    rows::sweep_counts_are_exact_across_the_full_protection_set(&MEM).await;
}

#[tokio::test]
async fn inv4_sweep_without_liveness_sources_aborts_and_deletes_nothing() {
    rows::sweep_without_liveness_sources_aborts_and_deletes_nothing(&MEM).await;
}

#[tokio::test]
async fn inv4_sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing() {
    rows::sweep_with_a_failed_protect_callback_aborts_and_deletes_nothing(&MEM).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn inv4_sweep_is_single_flight_the_contended_second_aborts_as_data() {
    rows::sweep_is_single_flight_the_contended_second_aborts_as_data(&MEM).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn inv4_sweep_racing_a_pinned_put_never_deletes_the_protected_entry() {
    rows::sweep_racing_a_pinned_put_never_deletes_the_protected_entry(&MEM).await;
}

#[tokio::test]
async fn inv5_list_correctness_is_observable_through_sweep_outcomes() {
    rows::list_correctness_is_observable_through_sweep_outcomes(&MEM).await;
}

#[tokio::test]
async fn inv6_stat_reports_len_and_algorithm_without_fetching_content() {
    rows::stat_reports_len_and_algorithm_without_fetching_content(&MEM).await;
}

#[tokio::test]
async fn constructor_mem_only_accepts_every_length_on_both_put_paths() {
    rows::mem_only_mode_accepts_every_length_on_both_put_paths(&MEM).await;
}

#[tokio::test]
async fn constructor_kv_only_rejects_over_threshold_puts_at_put_time() {
    rows::kv_only_mode_rejects_over_threshold_puts_at_put_time(&KV_ONLY).await;
}

// ------------------------------------------------------------------
// The factory contract's structural shapes (binding.rs' demonstrations
// carried to the real column factories): fresh isolated opens and
// idempotent teardown safe against a panicking row. These are not
// conformance rows — no backlog entry — they are the factory
// contract's own wiring-level assertions.
// ------------------------------------------------------------------

/// Every `open()` vends a fresh, isolated store: two stores live
/// concurrently without observable shared state, and a third open sees
/// neither island's entries.
#[tokio::test]
async fn factory_open_vends_fresh_isolated_stores() {
    let a = MEM.open().await.expect("open vends a fresh store");
    let b = MEM.open().await.expect("open vends another fresh store");

    let content_a = b"island a's content".to_vec();
    let content_b = b"island b's content".to_vec();
    let (key_a, _) = a
        .put(
            Some(content_a.len() as u64),
            Cursor::new(content_a.as_slice()),
        )
        .await
        .expect("island a's put publishes");
    let (key_b, _) = b
        .put(None, Cursor::new(content_b.as_slice()))
        .await
        .expect("island b's put publishes");

    assert!(a.has(&key_a.as_bytes()).await.unwrap());
    assert!(
        !a.has(&key_b.as_bytes()).await.unwrap(),
        "island a sees nothing of island b"
    );
    assert!(b.has(&key_b.as_bytes()).await.unwrap());
    assert!(
        !b.has(&key_a.as_bytes()).await.unwrap(),
        "island b sees nothing of island a"
    );

    let c = MEM.open().await.expect("a third open vends fresh");
    assert!(
        !c.has(&key_a.as_bytes()).await.unwrap(),
        "a third open shares nothing with the first"
    );
    assert!(
        !c.has(&key_b.as_bytes()).await.unwrap(),
        "a third open shares nothing with the second"
    );
}

/// A failing assertion panics the row before teardown runs — and
/// teardown must make sense over the abandoned instance anyway: it is
/// idempotent against the panicked row's path, and the next open vends
/// a fresh store with no residue (README §The factory contract).
#[tokio::test]
async fn factory_teardown_is_safe_against_a_panicking_row() {
    let content = b"put before the assertion fails".to_vec();
    let abandoned = Key::derive(&content, content.len() as u64);
    let factory = Arc::new(MemOnlyFactory);

    let task = {
        let factory = Arc::clone(&factory);
        tokio::spawn(async move {
            let store = factory.open().await.expect("open vends the store");
            let (_, pin) = store
                .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
                .await
                .expect("the put publishes");
            drop(pin);
            panic!("a failing assertion panics the row (the factory contract's early exit)");
        })
    };
    assert!(
        task.await.is_err(),
        "the row's panic surfaced as its failure"
    );

    factory
        .teardown()
        .await
        .expect("teardown makes sense over the abandoned instance");
    factory.teardown().await.expect("teardown is idempotent");

    let store = factory
        .open()
        .await
        .expect("the next open still vends a store");
    assert!(
        !store.has(&abandoned.as_bytes()).await.unwrap(),
        "no residue from the panicked row — isolation is factory-owned"
    );
}