//! WS1 fusion equivalence property test (`U_WS1_9_FUSION`, DESIGN §2.4
//! of `docs/development/graph-fidelity-planner-correctness/02_DESIGN-graph-fidelity-planner-correctness.md`).
//!
//! # What it proves
//!
//! For every well-formed [`sqry_db::QueryDb`] and every well-formed
//! [`sqry_db::planner::QueryPlan`] generated by [`plan_gen::arbitrary_plan_tree`],
//! the planner produces **byte-identical** results on the two execution
//! paths:
//!
//! 1. **Unfused path** — direct `sqry_db::planner::execute_plan(&plan, &db)`.
//!    Bypasses the fuser entirely; the executor walks the IR tree
//!    top-down with an empty shared-node cache.
//! 2. **Fused path** — `fuse_single(plan)` → `execute_batch(&batch, &db)`.
//!    The fuser promotes shared subtrees / subqueries, pre-populates the
//!    executor's shared-node cache, and dispatches the prefix exactly
//!    once per fusion group.
//!
//! Both paths must yield the same `Vec<NodeId>` (already sorted /
//! deduplicated by [`sqry_db::planner::execute::dedup_sort`]).
//!
//! Additionally, the test asserts the [`FusionStats`] invariants
//! prescribed by the DAG acceptance for `U_WS1_9_FUSION`:
//!
//! * `stats.scans_eliminated <= stats.total_plans` (≤ "scans original").
//! * `stats.fusion_groups <= stats.total_plans`.
//! * `stats.scans_eliminated == stats.total_plans - stats.fusion_groups`.
//!
//! The lower bound `scans_eliminated >= 0` is satisfied at the type
//! system level (`scans_eliminated: usize`) so we assert it via a
//! documented type-system witness rather than a runtime check that could
//! never fail.
//!
//! # Why the equivalence holds (and why this test still matters)
//!
//! Today, [`sqry_db::planner::execute_plan`] does not invoke the fuser —
//! single-plan execution is identical to unfused execution. The fuser
//! activates via the batch API (`fuse_plans` + `execute_batch`). This
//! property test:
//!
//! 1. Locks down the contract that running ANY plan through the fused
//!    batch path produces the same output as direct execution — even
//!    when batch fusion exercises shared-node priming, subquery
//!    pre-evaluation, and chain-prefix sharing for what is structurally
//!    a single-plan batch.
//! 2. Catches future regressions if production dispatch is wired to
//!    route single plans through `fuse_single` (e.g. for telemetry).
//! 3. Doubles as a fuzz / smoke for the IR's `Hash`/`Eq` contract on
//!    every operator and predicate variant the generator emits.
//!
//! # `QueryDbConfig::disable_fusion` exposure
//!
//! Per DAG `critical_decisions` for `U_WS1_9_FUSION`, the test-only
//! `QueryDbConfig::disable_fusion()` method is exposed (DESIGN §2.4) so
//! the comparison-side `QueryDb` has a single, documented config
//! surface even though the equivalence harness drives the two execution
//! paths directly. See `sqry-db/src/config.rs::QueryDbConfig::fusion_disabled`
//! for the gating + semantics docstring.
//!
//! # Acceptance criteria (DAG verbatim, `U_WS1_9_FUSION`)
//!
//! * Fused and unfused configs produce byte-identical results for an
//!   arbitrary plan.
//! * Plan tree generator bounded depth (`<= 6`) and arity (`<= 4`) — see
//!   [`plan_gen::MAX_DEPTH`] / [`plan_gen::MAX_ARITY`].
//! * `FusionStats` invariants hold on every case
//!   (`scans_eliminated >= 0`, `scans_eliminated <= scans_original`).

#![allow(clippy::needless_pass_by_value)]

use std::sync::Arc;

use proptest::prelude::*;
use proptest::test_runner::Config;

use sqry_core::graph::unified::node::id::NodeId;

use sqry_db::QueryDb;
use sqry_db::QueryDbConfig;
use sqry_db::planner::{FusionStats, QueryPlan, execute_batch, execute_plan, fuse_single};

#[path = "graph_gen.rs"]
#[allow(unused_imports)]
mod graph_gen;

#[path = "plan_gen.rs"]
#[allow(unused_imports)]
mod plan_gen;

use graph_gen::{GeneratedGraph, well_formed_graph};
use plan_gen::arbitrary_plan_tree;

// ---------------------------------------------------------------------------
// Proptest config — PROPTEST_CASES env override (default 256, PR 10k, nightly 100k)
// ---------------------------------------------------------------------------

/// Reads `PROPTEST_CASES` from the environment, defaulting to 256.
/// Matches the WS1_3 self-test / WS1_4 family convention (DESIGN §2.3,
/// §2.4).
fn cases_from_env() -> u32 {
    std::env::var("PROPTEST_CASES")
        .ok()
        .and_then(|v| v.parse::<u32>().ok())
        .unwrap_or(256)
}

fn proptest_config() -> Config {
    Config {
        cases: cases_from_env(),
        max_shrink_iters: 10_000,
        ..Config::default()
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Builds a default-config [`QueryDb`] (fusion enabled) over the
/// generated graph's snapshot. Used for the fused execution path.
fn build_db_fused(graph: &GeneratedGraph) -> QueryDb {
    let snapshot = Arc::new(graph.graph.snapshot());
    QueryDb::new(snapshot, QueryDbConfig::default())
}

/// Builds a [`QueryDb`] with [`QueryDbConfig::disable_fusion`] applied.
/// Used for the unfused execution path. While `execute_plan` today does
/// not consult the flag, the test explicitly constructs the disabled-
/// fusion config so the harness has a witness for the
/// `disable_fusion()`-exists surface and so a future production hookup
/// gets equivalence pinned across both code paths.
fn build_db_unfused(graph: &GeneratedGraph) -> QueryDb {
    let snapshot = Arc::new(graph.graph.snapshot());
    QueryDb::new(snapshot, QueryDbConfig::default().disable_fusion())
}

/// Runs `plan` through the **fused** path:
/// `fuse_single(plan)` → `execute_batch(&batch, &db)`. Returns
/// `(results, stats)` so callers can assert both the equivalence
/// property and the [`FusionStats`] invariants.
fn run_fused(db: &QueryDb, plan: &QueryPlan) -> (Vec<NodeId>, FusionStats) {
    let batch = fuse_single(plan.clone());
    let stats = *batch.stats();
    let mut per_plan = execute_batch(&batch, db);
    debug_assert_eq!(
        per_plan.len(),
        1,
        "fuse_single must yield a 1-plan batch (total_plans={})",
        stats.total_plans
    );
    let results = per_plan.pop().unwrap_or_default();
    (results, stats)
}

/// Runs `plan` through the **unfused** path: direct
/// [`execute_plan`]. Bypasses every fusion-side code path.
fn run_unfused(db: &QueryDb, plan: &QueryPlan) -> Vec<NodeId> {
    execute_plan(plan, db)
}

/// Asserts the [`FusionStats`] invariants required by DAG acceptance for
/// `U_WS1_9_FUSION`.
///
/// * `scans_eliminated <= total_plans`. The DESIGN spells the upper
///   bound as `scans_eliminated <= scans_original`; for a single-plan
///   batch, `scans_original` is the input `total_plans` count (one scan
///   per plan if no fusion happens).
/// * `fusion_groups <= total_plans`.
/// * `scans_eliminated == total_plans - fusion_groups` (the algebraic
///   identity the fuser maintains).
///
/// The lower bound `scans_eliminated >= 0` is enforced by the
/// `usize`-typing of [`FusionStats::scans_eliminated`]; the assertion
/// below is a documented witness (`stats.scans_eliminated == stats.scans_eliminated`)
/// rather than a runtime check that can never fail.
fn assert_fusion_stats_invariants(stats: &FusionStats) -> Result<(), TestCaseError> {
    // Type-system check: `scans_eliminated: usize` cannot go negative.
    // This line documents the contract DAG mentions explicitly
    // ("`scans_eliminated >= 0`" — type-system check if u-typed).
    let _: usize = stats.scans_eliminated;

    prop_assert!(
        stats.fusion_groups <= stats.total_plans,
        "fusion_groups ({}) > total_plans ({}) — fuser must never produce more groups than inputs",
        stats.fusion_groups,
        stats.total_plans,
    );

    prop_assert!(
        stats.scans_eliminated <= stats.total_plans,
        "scans_eliminated ({}) > total_plans ({}) — eliminated cannot exceed original-scan count",
        stats.scans_eliminated,
        stats.total_plans,
    );

    // Algebraic identity: scans_eliminated = total_plans - fusion_groups.
    // Use saturating_sub for symmetry with the fuser implementation.
    let expected_eliminated = stats.total_plans.saturating_sub(stats.fusion_groups);
    prop_assert_eq!(
        stats.scans_eliminated,
        expected_eliminated,
        "scans_eliminated invariant: total_plans({}) - fusion_groups({}) = {}, got {}",
        stats.total_plans,
        stats.fusion_groups,
        expected_eliminated,
        stats.scans_eliminated,
    );

    // Subquery counters must respect their own ordering.
    prop_assert!(
        stats.subqueries_unique <= stats.subqueries_total,
        "subqueries_unique ({}) > subqueries_total ({})",
        stats.subqueries_unique,
        stats.subqueries_total,
    );

    Ok(())
}

// ---------------------------------------------------------------------------
// Equivalence runner
// ---------------------------------------------------------------------------

/// Runs the equivalence comparison for a single (graph, plan) pair.
///
/// Separated from the `proptest!` body so the inline doc-comment +
/// failure diagnostics live next to the assertion logic. Returns
/// `Result<(), TestCaseError>` so `prop_assert*` macros propagate.
fn run_fusion_equivalence(graph: &GeneratedGraph, plan: &QueryPlan) -> Result<(), TestCaseError> {
    let db_fused = build_db_fused(graph);
    let db_unfused = build_db_unfused(graph);

    let (fused_results, fused_stats) = run_fused(&db_fused, plan);
    let unfused_results = run_unfused(&db_unfused, plan);

    // The DAG's "byte-identical" wording maps to Rust's `Vec<NodeId>`
    // equality: both sides return sorted+deduped vectors so structural
    // equality is sufficient.
    prop_assert_eq!(
        &fused_results,
        &unfused_results,
        "fused vs unfused divergence: |fused|={}, |unfused|={}, plan={:?}",
        fused_results.len(),
        unfused_results.len(),
        plan,
    );

    assert_fusion_stats_invariants(&fused_stats)?;

    Ok(())
}

// ---------------------------------------------------------------------------
// Proptest body
// ---------------------------------------------------------------------------

proptest! {
    #![proptest_config(proptest_config())]

    /// Core equivalence property: fused == unfused for arbitrary
    /// well-formed plans on arbitrary well-formed graphs.
    ///
    /// Acceptance: DAG `U_WS1_9_FUSION` —
    /// "fused and unfused configs produce byte-identical results for
    /// arbitrary plan".
    #[test]
    fn fusion_preserves_results(
        graph in well_formed_graph(),
        plan in arbitrary_plan_tree(),
    ) {
        run_fusion_equivalence(&graph, &plan)?;
    }
}

// ---------------------------------------------------------------------------
// Static smoke — pins the `disable_fusion()` API and the FusionStats
// invariant helper against trivial inputs. Acts as a fast feedback gate
// before the proptest harness runs; failures here are programmer errors
// in the harness, not the proptest input space.
// ---------------------------------------------------------------------------

#[test]
fn smoke_disable_fusion_method_exists_and_is_idempotent() {
    let cfg = QueryDbConfig::default().disable_fusion();
    assert!(
        cfg.fusion_disabled,
        "QueryDbConfig::disable_fusion() must set fusion_disabled = true"
    );
    let cfg2 = cfg.clone().disable_fusion();
    assert!(
        cfg2.fusion_disabled,
        "disable_fusion() must be idempotent (already-disabled config stays disabled)"
    );
}

#[test]
fn smoke_fusion_stats_invariants_on_default() {
    // Default FusionStats has all-zero counters and must satisfy every
    // invariant trivially. This guards against an accidental tightening
    // of the assertions that would reject the default value.
    let stats = FusionStats::default();
    // Wrap in a closure so we can use `prop_assert!` machinery uniformly.
    let result: Result<(), TestCaseError> = (|| {
        assert_fusion_stats_invariants(&stats)?;
        Ok(())
    })();
    assert!(
        result.is_ok(),
        "default FusionStats must satisfy invariants: {result:?}"
    );
}
