//! Red architectural contracts for complete targeted lazy topology coverage.
//!
//! Phase 3 intentionally selects the eager oracle for mutable topology and for
//! targeted algorithms other than clean-CSR BFS. These contracts keep Phase 4
//! honest: representation-specific cursors must be owned across PostgreSQL
//! visibility probes, and each migrated algorithm must retain its established
//! ordering and admission semantics.

use std::fs;
use std::path::Path;

fn crate_source(relative: &str) -> String {
    fs::read_to_string(Path::new(env!("CARGO_MANIFEST_DIR")).join(relative))
        .unwrap_or_else(|error| panic!("failed to read {relative}: {error}"))
}

fn function_body<'a>(source: &'a str, signature: &str) -> &'a str {
    let start = source
        .find(signature)
        .unwrap_or_else(|| panic!("missing function `{signature}`"));
    let remaining = &source[start..];
    let open = remaining.find('{').expect("function must have a body");
    let mut depth = 0usize;
    for (offset, byte) in remaining[open..].bytes().enumerate() {
        match byte {
            b'{' => depth += 1,
            b'}' => {
                depth -= 1;
                if depth == 0 {
                    return &remaining[..open + offset + 1];
                }
            }
            _ => {}
        }
    }
    panic!("function `{signature}` has unbalanced braces")
}

#[test]
fn bounded_classic_edge_overlays_use_owned_cursors_instead_of_eager_fallbacks() {
    let engine = crate_source("src/engine.rs");
    let neighbors = crate_source("src/projection/neighbors.rs");
    let prepare = function_body(&engine, "fn prepare_resumable_traversal");

    assert!(
        !prepare.contains("!self.edge_buffer.is_empty()"),
        "bounded committed edge overlays must not force a table-wide eager visibility scan"
    );
    assert!(
        neighbors.contains("OwnedNeighborCursor"),
        "classic overlays need an owned cursor that can cross the ENGINE/SPI boundary without retaining projection borrows"
    );
    for required in [
        "resumable_bfs_matches_eager_with_committed_overlay",
        "resumable_bfs_matches_eager_with_transaction_delta",
    ] {
        assert!(
            engine.contains(required),
            "P4 mutable-topology differential corpus is missing `{required}`"
        );
    }
}

#[test]
fn durable_layers_require_a_bounded_owned_cursor_before_lazy_bfs_selection() {
    let engine = crate_source("src/engine.rs");
    let bfs = crate_source("src/bfs.rs");
    let neighbors = crate_source("src/projection/neighbors.rs");
    let layered = crate_source("src/projection/layered.rs");
    let prepare = function_body(&engine, "fn prepare_resumable_traversal");
    let layered_neighbor_impl = function_body(&layered, "impl NeighborSource for LayeredNeighbors");
    let directional_impl = function_body(
        &layered,
        "impl NeighborSource for DirectionalLayeredNeighbors",
    );

    assert!(
        neighbors.contains("Layered {") || layered.contains("OwnedLayeredNeighborCursor"),
        "P4.2 needs an owned layered cursor that records k-way base, durable-segment, committed-overlay, and transaction-delta positions"
    );
    assert!(
        layered_neighbor_impl.contains("fn fill_neighbors")
            && directional_impl.contains("fn fill_neighbors"),
        "layered sources must override the replaying default pager before they can cross the ENGINE/SPI boundary"
    );
    for required in [
        "owned_layered_cursor_matches_current_order_in_both_directions",
        "owned_layered_cursor_preserves_tombstone_precedence_and_parallel_relationships",
        "owned_layered_cursor_yields_progress_after_zero_output_raw_page",
        "owned_layered_cursor_bounds_examined_raw_rows",
        "owned_layered_cursor_matches_eager_for_generated_mutation_sequences",
        "owned_layered_cursor_preserves_parallel_base_relationship_identity_order",
        "owned_layered_cursor_pages_base_chunk_replacements_without_base_leakage",
        "owned_layered_cursor_collapses_duplicate_full_base_keys_across_pages",
        "owned_layered_cursor_preserves_durable_and_frozen_overlay_precedence",
        "owned_layered_cursor_preserves_inbound_durable_and_frozen_overlay_precedence",
    ] {
        assert!(
            layered.contains(required),
            "P4.2 layered-cursor behavioral corpus is missing `{required}`"
        );
    }
    assert!(
        !prepare.contains("layered_neighbors()?.is_some()"),
        "segment-backed targeted BFS must stop selecting eager only after the bounded layered cursor contracts are green"
    );
    assert!(
        bfs.contains("resumable_bfs_epoch_rejects_same_cardinality_topology_substitution"),
        "resumable durable traversal must reject same-cardinality overlay substitutions across policy SPI"
    );
}

#[test]
fn segment_backed_any_requires_a_shared_precedence_cursor_before_lazy_selection() {
    let engine = crate_source("src/engine.rs");
    let layered = crate_source("src/projection/layered.rs");
    let prepare = function_body(&engine, "fn prepare_resumable_traversal");

    for required in [
        "owned_layered_any_cursor_preserves_shared_map_precedence_and_order",
        "owned_layered_any_cursor_yields_after_bounded_zero_output_work",
        "owned_layered_any_cursor_matches_eager_for_generated_direction_pairs",
    ] {
        assert!(
            layered.contains(required),
            "P4.2 Any cursor behavioral corpus is missing `{required}`"
        );
    }
    assert!(
        prepare.contains("let segment_backed_any")
            && prepare.contains("any_direction_overlays = segment_backed_any.then"),
        "segment-backed Any must freeze both directional overlays before the first visibility yield"
    );
    assert!(
        engine.contains("assert_resumable_bfs_matches_eager")
            && engine.contains("TraversalDirection::Any"),
        "segment-backed Any eligibility needs an engine-level eager/resumable differential"
    );
}

#[test]
fn targeted_bfs_workflow_inventory_is_explicit() {
    let workflow = crate_source("src/sql_facade/workflow.rs");

    let expand = function_body(&workflow, "fn expand");
    let find_related = function_body(&workflow, "fn find_related");
    let neighborhood = function_body(&workflow, "fn neighborhood");
    assert!(
        expand.contains("StatementBfsVisibility") && expand.contains("execute_rows"),
        "expand is a direct targeted BFS workflow and needs eager/lazy differential coverage"
    );
    assert!(
        find_related
            .matches("traverse_search_rows_with_statement_bfs_visibility")
            .count()
            >= 2,
        "find_related is the resolver-sharing workflow because its filtered and broad-count traversals must reuse one statement-local oracle"
    );
    assert!(
        neighborhood.contains("traverse_search_rows_with_statement_bfs_visibility"),
        "neighborhood is a targeted BFS workflow and needs exact grouped-row eager/lazy parity"
    );
}

#[test]
fn dfs_and_reverse_traversal_freeze_order_and_visited_timing_before_migration() {
    let bfs = crate_source("src/bfs.rs");
    let engine = crate_source("src/engine.rs");
    let traversal = crate_source("src/sql_traversal.rs");
    let pg_traversal = crate_source("src/pg_tests/traversal_api.rs");

    for required in [
        "resumable_dfs_matches_eager_reversed_push_order_and_visited_timing",
        "resumable_dfs_matches_eager_out_in_any_across_overlay_durable_and_tx",
        "resumable_dfs_marks_visited_on_push_for_duplicates_cycles_and_parallel_edges",
        "resumable_dfs_hidden_node_relationship_and_intermediate_match_eager",
        "resumable_dfs_preserves_max_nodes_frontier_depth_and_truncation",
        "resumable_dfs_yields_bounded_owned_pages_and_rejects_epoch_change",
    ] {
        assert!(
            bfs.contains(required) || engine.contains(required),
            "P4.3 DFS differential corpus is missing `{required}`"
        );
    }
    assert!(
        bfs.contains("ResumableDfsMachine")
            && bfs.contains("OwnedNeighborCursor")
            && bfs.contains("neighbors_reversed"),
        "DFS must own its stack, visited timing, reversed-neighbor cursor, parents, and outputs before policy probes can run outside ENGINE borrows"
    );
    assert!(
        traversal.contains("execute_lazy_dfs_candidates")
            && traversal.contains("resolve_bfs_visibility_batch"),
        "targeted DFS must resolve bounded node and relationship candidates outside ENGINE borrows through the established PostgreSQL oracle"
    );
    for required in [
        "dfs_lazy_matches_eager_rls_out_in_any_and_hidden_intermediate",
        "dfs_lazy_multiseed_matches_eager_and_selects_lazy",
        "dfs_lazy_traverse_search_matches_eager_and_selects_lazy",
        "dfs_lazy_relationship_rls_blocks_hidden_parallel_edge",
        "dfs_lazy_cancellation_and_policy_error_drop_state_then_retry",
        "dfs_no_rls_fast_path_has_zero_visibility_spi",
    ] {
        assert!(
            pg_traversal.contains(required),
            "P4.3 PostgreSQL DFS corpus is missing `{required}`"
        );
    }
}

#[test]
fn targeted_unweighted_paths_freeze_level_meeting_and_endpoint_semantics_before_migration() {
    let paths = crate_source("src/path_finder.rs");
    let engine = crate_source("src/engine.rs");
    let traversal = crate_source("src/sql_facade/traversal.rs");
    let postgres_paths = crate_source("src/pg_tests/traversal_paths.rs");

    for required in [
        "resumable_single_direction_bfs_matches_eager_target_discovery_and_ties",
        "resumable_bidirectional_bfs_matches_eager_meeting_node_selection",
        "resumable_bidirectional_bfs_completes_chosen_level_before_selecting_meeting",
        "resumable_unweighted_paths_match_eager_hidden_nodes_relationships_and_endpoints",
        "resumable_unweighted_paths_match_eager_overlay_durable_and_tx",
        "resumable_unweighted_paths_match_eager_for_generated_graphs_and_visibility_masks",
        "resumable_unweighted_paths_preserve_max_depth_work_caps_errors_and_epoch_rejection",
    ] {
        assert!(
            paths.contains(required) || engine.contains(required),
            "P4.4 unweighted-path differential corpus is missing `{required}`"
        );
    }
    assert!(
        paths.contains("ResumableSingleDirectionBfs")
            && paths.contains("ResumableBidirectionalBfs"),
        "single-direction and bidirectional paths need separate owned resumable state so every intermediate can be policy-resolved outside ENGINE borrows"
    );
    assert!(
        traversal.contains("execute_lazy_shortest_path_rows")
            && traversal.contains("resolve_bfs_visibility_batch"),
        "graph.shortest_path must resolve bounded node and relationship candidates through the established PostgreSQL policy oracle"
    );
    let shortest_path_rows = function_body(&traversal, "fn shortest_path_rows_governed");
    assert!(
        shortest_path_rows.contains("execute_lazy_shortest_path_rows")
            && !shortest_path_rows.contains("prepare_eager_visibility"),
        "the direct unweighted path API must select its resumable executor instead of unconditionally preparing the eager oracle"
    );

    for required in [
        "unweighted_paths_lazy_match_eager_single_bidirectional_ties_and_order",
        "unweighted_paths_lazy_node_and_relationship_rls_choose_visible_route",
        "unweighted_paths_lazy_source_target_visibility_preserves_error_ordering",
        "unweighted_paths_lazy_overlay_durable_and_tx_match_eager",
        "unweighted_paths_lazy_resource_cap_and_missing_identity_fail_closed",
        "unweighted_paths_lazy_cancellation_policy_error_drop_state_then_retry",
        "unweighted_paths_no_rls_fast_path_has_zero_visibility_spi",
    ] {
        assert!(
            postgres_paths.contains(required),
            "P4.4 PostgreSQL unweighted-path corpus is missing `{required}`"
        );
    }
}

#[test]
fn targeted_weighted_paths_freeze_heap_and_tie_order_before_migration() {
    let paths = crate_source("src/path_finder.rs");
    let engine = crate_source("src/engine.rs");
    let traversal = crate_source("src/sql_facade/traversal.rs");
    let visibility = crate_source("src/sql_visibility.rs");
    let postgres_paths = crate_source("src/pg_tests/traversal_paths.rs");

    for required in [
        "ResumableDijkstra",
        "materialize_resumable_dijkstra_batch",
        "apply_resumable_dijkstra_verdicts",
        "resumable_dijkstra_matches_eager_heap_ties_stale_entries_and_target_pop",
        "resumable_dijkstra_matches_eager_generated_visibility_and_weights",
        "resumable_dijkstra_preserves_weighted_step_metadata_and_edge_type_filters",
        "resumable_dijkstra_pages_one_popped_node_at_a_time",
        "resumable_dijkstra_zero_output_pages_are_bounded",
        "resumable_dijkstra_rejects_epoch_work_memory_and_oversized_keys",
    ] {
        assert!(
            paths.contains(required),
            "P4.5 weighted-path state/differential corpus is missing `{required}`"
        );
    }

    let preparation = function_body(&visibility, "fn prepare_bfs_visibility");
    assert!(
        !preparation.contains("edge_store"),
        "lazy visibility preparation must not scan projected edges to discover dynamic labels"
    );

    for required in [
        "prepare_resumable_dijkstra",
        "materialize_resumable_dijkstra_batch",
        "admit_resumable_dijkstra_batch",
        "finish_resumable_dijkstra",
    ] {
        assert!(
            engine.contains(required),
            "P4.5 engine integration is missing `{required}`"
        );
    }

    let weighted = function_body(&traversal, "fn weighted_shortest_path");
    let weighted_typed = function_body(&traversal, "fn weighted_shortest_path_typed");
    for (name, body) in [
        ("weighted_shortest_path", weighted),
        ("weighted_shortest_path_typed", weighted_typed),
    ] {
        assert!(
            body.contains("weighted_shortest_path_rows_governed"),
            "{name} must route through one governed weighted-path executor"
        );
        assert!(
            !body.contains("prepare_eager_visibility"),
            "{name} must not unconditionally prepare the eager whole-table oracle"
        );
    }

    for required in [
        "weighted_paths_lazy_match_eager_rls_ties_filters_and_metadata",
        "weighted_paths_lazy_durable_segments_match_eager",
        "weighted_paths_durable_unseen_dynamic_label_remains_pg018",
        "weighted_paths_pending_edge_overlay_remains_pg018",
        "weighted_paths_tx_node_state_falls_back_eager",
        "weighted_paths_lazy_resource_identity_cancellation_and_retry",
        "weighted_paths_no_rls_fast_path_has_zero_visibility_spi",
    ] {
        assert!(
            postgres_paths.contains(required),
            "P4.5 PostgreSQL weighted-path corpus is missing `{required}`"
        );
    }
}

#[test]
fn targeted_workflows_share_one_statement_local_resolver() {
    let workflow = crate_source("src/sql_facade/workflow.rs");
    let traversal = crate_source("src/sql_traversal.rs");
    let workflow_search_tests = crate_source("src/pg_tests/workflow_search_api.rs");
    let workflow_relationship_tests = crate_source("src/pg_tests/workflow_relationship_api.rs");
    let expand = function_body(&workflow, "fn expand");
    let find_related = function_body(&workflow, "fn find_related");
    let neighborhood = function_body(&workflow, "fn neighborhood");

    assert!(
        traversal.contains("struct StatementBfsVisibility")
            && traversal.contains("fn execute_candidates")
            && traversal.contains("fn execute_rows"),
        "P4.2 workflows need an owned statement-local BFS visibility resolver that survives multiple internal traversals without becoming transaction-scoped"
    );
    for (name, body) in [
        ("expand", expand),
        ("find_related", find_related),
        ("neighborhood", neighborhood),
    ] {
        assert!(
            body.contains("StatementBfsVisibility"),
            "{name} must create one statement-local resolver before its first targeted traversal"
        );
        assert_eq!(
            body.matches("StatementBfsVisibility").count(),
            1,
            "{name} must own one resolver, not prepare one resolver per internal traversal"
        );
        assert!(
            !body.contains("prepare_eager_visibility"),
            "{name} must select eager or resumable BFS through the shared statement resolver instead of forcing the eager oracle"
        );
    }
    assert!(
        find_related
            .matches("traverse_search_rows_with_statement_bfs_visibility")
            .count()
            >= 2,
        "find_related must retain both filtered and broad-count traversals so the shared-resolver test exercises actual reuse"
    );

    for required in [
        "workflow_expand_lazy_matches_eager_rows_hydration_and_truncation",
        "workflow_find_related_lazy_reuses_one_resolver_across_roots_and_counts",
        "workflow_lazy_visibility_cancellation_cleans_statement_and_retries",
        "workflow_no_rls_lazy_fast_path_has_zero_visibility_spi",
    ] {
        assert!(
            workflow_search_tests.contains(required),
            "P4.2 workflow search PostgreSQL corpus is missing `{required}`"
        );
    }
    assert!(
        workflow_relationship_tests
            .contains("workflow_neighborhood_lazy_matches_eager_groups_samples_and_caps"),
        "P4.2 neighborhood PostgreSQL corpus must compare exact grouped rows, deterministic samples, and truncation caps"
    );
}

#[test]
fn targeted_gql_and_cypher_lazy_postgresql_contract_is_complete() {
    let gql_tests = crate_source("src/pg_tests/gql.rs");
    let cypher_tests = crate_source("src/pg_tests/cypher.rs");

    for required in [
        "gql_identity_bounded_expansion_lazy_matches_eager_rls_optional_multipattern_order_and_caps",
        "gql_identity_bounded_write_match_lazy_uses_postgres_rls_and_matches_eager",
        "gql_identity_bounded_lazy_cancellation_policy_error_drop_state_then_retry",
        "gql_whole_source_scan_remains_eager_under_forced_lazy",
        "gql_no_rls_identity_bounded_fast_path_has_zero_resolver_spi",
        "gql_identity_filtered_join_eager_fallback_preserves_result_row_cap",
        "gql_identity_bounded_parallel_edge_after_distinct_node_cap_matches_eager_error",
        "gql_identity_bounded_acl_denial_precedes_null_missing_and_invalid_identity_values",
    ] {
        assert!(
            gql_tests.contains(required),
            "P4.6 GQL PostgreSQL corpus is missing `{required}`"
        );
    }
    assert!(
        cypher_tests
            .contains("cypher_identity_bounded_expansion_selects_lazy_and_matches_gql_under_rls"),
        "P4.6 openCypher compatibility coverage must exercise the same targeted lazy executor"
    );
}

#[test]
fn global_topology_work_remains_on_the_eager_oracle() {
    let components = crate_source("src/sql_facade/components.rs");
    let aggregation = crate_source("src/sql_aggregation.rs");

    for (source, signature) in [
        (&components, "fn connected_components"),
        (&components, "fn component_stats"),
    ] {
        let body = function_body(source, signature);
        assert!(
            body.contains("prepare_eager_visibility") && !body.contains("execute_lazy"),
            "global surface `{signature}` must retain the eager oracle until it has a purpose-built resumable executor"
        );
    }
    assert!(
        !aggregation.contains("ResumablePathCount"),
        "path-count analytics are explicitly outside the P4 targeted-lazy slice"
    );
}
