//! Contract tests for the durable projection build sequence.
//!
//! These tests record the required behavior and fixture call sites that durable
//! projection modules must turn green. Implemented contracts pass; future
//! contracts fail by default so phase progress is visible in the normal suite.

use super::test_fixtures::{
    assert_full_csr_equivalence, edge_store_from_tuples, NormalizedMutation, ProjectionArtifactDir,
};
use crate::projection::ingest::{ProjectionIngester, ProjectionSyncRow};
use crate::projection::layered::LayeredNeighbors;
use crate::projection::manifest::{
    ManifestChunkRef, ManifestFileRef, ManifestSegmentRef, ProjectionManifest,
    ProjectionManifestStore, VALIDATION_STATUS_VALID,
};
use crate::projection::neighbors::CsrNeighbors;
use crate::projection::normalize::{MutationBufferLimits, MutationOperation};
use crate::projection::segment::{
    DeltaSegment, SegmentEdge, SegmentEdgeWeight, SegmentFilterValue, SegmentKind,
    SegmentNodeState, SegmentResolution, SegmentTenant,
};
use crate::types::TraversalDirection;

#[test]
fn projection_manifest_roundtrips_base_only_generation() {
    let dir = ProjectionArtifactDir::new("projection_manifest_roundtrips_base_only_generation");
    let manifest_path = dir.manifest_path(1);
    let manifest = ProjectionManifest::base_only(
        1,
        manifest_path.to_string_lossy(),
        "xxh3:base",
        2,
        42,
        1_700_000,
    );

    let json = manifest.to_pretty_json().expect("manifest encodes");
    let decoded = ProjectionManifest::from_json(&json).expect("manifest decodes");

    assert_eq!(decoded, manifest);
    assert_eq!(decoded.validation_status, VALIDATION_STATUS_VALID);
}

#[test]
fn delta_segment_roundtrips_edge_topology_weight_and_delete_sections() {
    let dir = ProjectionArtifactDir::new(
        "delta_segment_roundtrips_edge_topology_weight_and_delete_sections",
    );
    let segment_path = dir.segment_path(1, 0);
    let weighted = NormalizedMutation {
        generation_id: 1,
        direction: TraversalDirection::Out,
        source: 0,
        target: 1,
        type_id: crate::types::EdgeTypeId::test_v6(2),
        schema_reversed: false,
        weight: Some(7),
        tombstone: false,
    };
    let delete = NormalizedMutation {
        tombstone: true,
        ..weighted.clone()
    };
    let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 42)
        .expect("segment constructs");
    segment.edge_inserts.push(SegmentEdge {
        source: weighted.source,
        target: weighted.target,
        type_id: weighted.type_id,
        schema_reversed: false,
        relationship_id: None,
    });
    segment.edge_weights.push(SegmentEdgeWeight {
        source: weighted.source,
        target: weighted.target,
        type_id: weighted.type_id,
        relationship_id: None,
        weight: weighted.weight.expect("fixture has weight"),
        schema_reversed: false,
    });
    segment.edge_deletes.push(SegmentEdge {
        source: delete.source,
        target: delete.target,
        type_id: delete.type_id,
        schema_reversed: false,
        relationship_id: None,
    });

    segment
        .write_to_path(&segment_path)
        .expect("segment writes");
    let decoded = DeltaSegment::read_from_path(&segment_path).expect("segment reads");

    assert_eq!(decoded.edge_inserts, segment.edge_inserts);
    assert_eq!(decoded.edge_weights, segment.edge_weights);
    assert_eq!(decoded.edge_deletes, segment.edge_deletes);
}

#[test]
fn delta_segment_roundtrips_node_resolution_filter_tenant_sections() {
    let dir = ProjectionArtifactDir::new(
        "delta_segment_roundtrips_node_resolution_filter_tenant_sections",
    );
    let segment_path = dir.segment_path(1, 1);
    let mut segment = DeltaSegment::new(SegmentKind::Node, 0, TraversalDirection::Any, 0, 4, 43)
        .expect("segment constructs");
    segment.node_states.push(SegmentNodeState {
        node_idx: 1,
        active: true,
    });
    segment.resolutions.push(SegmentResolution {
        table_oid: 100,
        pk_hash: crate::resolution_index::ResolutionIndexBuilder::hash_pk("node-1"),
        primary_key: Some("node-1".to_string()),
        node_idx: 1,
        tombstone: false,
    });
    segment.filters.push(SegmentFilterValue {
        node_idx: 1,
        column_id: 2,
        value: crate::filter_index::PersistedFilterValue::Uuid(
            0x1234_5678_90ab_cdef_1234_5678_90ab_cdef,
        ),
        tombstone: false,
    });
    segment.tenants.push(SegmentTenant {
        node_idx: 1,
        tenant_hash: xxhash_rust::xxh3::xxh3_64(b"tenant-1"),
        tenant: Some("tenant-1".to_string()),
        tombstone: true,
    });

    segment
        .write_to_path(&segment_path)
        .expect("segment writes");
    let decoded = DeltaSegment::read_from_path(&segment_path).expect("segment reads");

    assert_eq!(decoded.node_states, segment.node_states);
    assert_eq!(decoded.resolutions, segment.resolutions);
    assert_eq!(decoded.filters, segment.filters);
    assert_eq!(decoded.tenants, segment.tenants);
}

#[test]
fn projection_ingest_committed_edge_insert_publishes_l0_manifest() {
    let dir = ProjectionArtifactDir::new("projection_ingest_committed_edge_contract");
    std::fs::write(dir.path().join("base.pggraph"), b"base").expect("base artifact writes");
    ProjectionManifestStore::new(dir.path())
        .publish(&ProjectionManifest::base_only(
            1,
            "base.pggraph",
            "crc32:00000000",
            1,
            0,
            1,
        ))
        .expect("base manifest publishes");
    let ingester = ProjectionIngester::new(dir.path(), "base.pggraph", "crc32:00000000", 1);
    let row = ProjectionSyncRow {
        sync_id: 1,
        generation_id: 1,
        committed: true,
        operation: MutationOperation::InsertEdge,
        direction: TraversalDirection::Out,
        source: 0,
        target: 1,
        type_id: crate::types::EdgeTypeId::test_v6(2),
        edge_type_label: None,
        weight: None,
        relationship_identity: None,
        table_oid: None,
        pk_hash: None,
        primary_key: None,
        node_idx: None,
        filter_column_id: None,
        filter_value: None,
        tenant_hash: None,
        tenant: None,
        schema_reversed: false,
    };

    let result = ingester
        .ingest_committed_rows(&[row], MutationBufferLimits::new(10, 10_000))
        .expect("ingestion publishes");
    let manifest = result.manifest.expect("manifest published");
    let segment = DeltaSegment::read_from_path(&dir.path().join(&manifest.segments[0].path))
        .expect("segment reads");

    assert_eq!(manifest.sync_watermark, 1);
    assert_eq!(manifest.segments.len(), 1);
    assert_eq!(
        segment.edge_inserts,
        vec![SegmentEdge {
            source: 0,
            target: 1,
            type_id: crate::types::EdgeTypeId::test_v6(2),
            schema_reversed: false,
            relationship_id: None,
        }]
    );
}

#[test]
fn layered_neighbors_equal_full_rebuild_for_insert_delete_sequence() {
    let base = edge_store_from_tuples(4, &[(0, 1, 1), (0, 2, 1)]);
    let mut insert = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 1)
        .expect("insert segment");
    insert.edge_inserts.push(SegmentEdge {
        source: 0,
        target: 3,
        type_id: crate::types::EdgeTypeId::test_v6(1),
        schema_reversed: false,
        relationship_id: None,
    });
    let mut delete = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 4, 2)
        .expect("delete segment");
    delete.edge_deletes.push(SegmentEdge {
        source: 0,
        target: 1,
        type_id: crate::types::EdgeTypeId::test_v6(1),
        schema_reversed: false,
        relationship_id: None,
    });
    let full_rebuild = edge_store_from_tuples(4, &[(0, 2, 1), (0, 3, 1)]);
    let expected = CsrNeighbors::new(&full_rebuild);
    let layered = LayeredNeighbors::new(&base, vec![insert, delete]);

    assert_full_csr_equivalence(4, &expected, &layered);
}

#[test]
fn status_reports_manifest_watermark_segments_chunks_gc_and_repair() {
    let dir = ProjectionArtifactDir::new(
        "status_reports_manifest_watermark_segments_chunks_gc_and_repair_contract",
    );
    std::fs::write(dir.path().join("base.pggraph"), b"base").expect("base writes");
    let segment_path = dir.segment_path(1, 0);
    let mut segment = DeltaSegment::new(SegmentKind::Edge, 0, TraversalDirection::Out, 0, 2, 4)
        .expect("segment constructs");
    segment.edge_inserts.push(SegmentEdge {
        source: 0,
        target: 1,
        type_id: crate::types::EdgeTypeId::test_v6(1),
        schema_reversed: false,
        relationship_id: None,
    });
    segment.edge_deletes.push(SegmentEdge {
        source: 1,
        target: 0,
        type_id: crate::types::EdgeTypeId::test_v6(1),
        schema_reversed: false,
        relationship_id: None,
    });
    segment
        .write_to_path(&segment_path)
        .expect("segment writes");
    let chunk_path = dir.path().join("active.pggraph-chunk");
    segment
        .write_to_path(&chunk_path)
        .expect("chunk segment writes");
    let obsolete_path = dir.path().join("obsolete.pggraph-delta");
    std::fs::write(&obsolete_path, b"old").expect("obsolete writes");

    let mut manifest = ProjectionManifest::base_only(7, "base.pggraph", "crc32:base", 1, 4, 99);
    manifest.mark_ingestion();
    manifest.mark_compaction();
    manifest.mark_repair();
    manifest.segments.push(ManifestSegmentRef {
        path: relative_path(dir.path(), &segment_path),
        checksum: checksum_for_path(&segment_path),
        level: 0,
        source_start: 0,
        source_end: 2,
        sync_watermark: 4,
    });
    manifest.base_chunks.push(ManifestChunkRef {
        path: relative_path(dir.path(), &chunk_path),
        checksum: checksum_for_path(&chunk_path),
        source_start: 0,
        source_end: 2,
        dirty_source_count: 2,
        dirty_edge_count: 1,
    });
    manifest.obsolete_files.push(ManifestFileRef {
        path: relative_path(dir.path(), &obsolete_path),
        bytes: 3,
    });
    ProjectionManifestStore::new(dir.path())
        .publish(&manifest)
        .expect("manifest publishes");
    crate::projection::status::record_projection_gc(dir.path()).expect("GC timestamp records");

    let status = crate::projection::status::collect_projection_status(dir.path(), None, 9, 1, 0)
        .expect("status collects");

    assert_eq!(status.manifest_generation, Some(7));
    assert_eq!(status.manifest_watermark, Some(4));
    assert_eq!(status.pending_durable_rows, 5);
    assert_eq!(status.segment_count, 1);
    assert_eq!(status.l0_segment_count, 1);
    assert_eq!(status.edge_segment_count, 1);
    assert_eq!(status.dirty_chunk_count, 1);
    assert_eq!(status.obsolete_file_count, 1);
    assert_eq!(status.obsolete_bytes, 3);
    assert_eq!(status.active_generation_count, 1);
    assert_eq!(status.artifact_validation_state, "healthy");
    assert!(status.ingest_recommended);
    assert!(status.compaction_recommended);
    assert!(status.gc_recommended);
    assert!(!status.repair_recommended);
    assert_eq!(status.last_ingestion_unix_micros, Some(99));
    assert_eq!(status.last_compaction_unix_micros, Some(99));
    assert!(status.last_gc_unix_micros.is_some());
    assert_eq!(status.last_repair_unix_micros, Some(99));
}

fn relative_path(root: &std::path::Path, path: &std::path::Path) -> String {
    path.strip_prefix(root)
        .expect("path is under root")
        .to_string_lossy()
        .into_owned()
}

fn checksum_for_path(path: &std::path::Path) -> String {
    format!(
        "crc32:{:08x}",
        crc32fast::hash(&std::fs::read(path).expect("file reads"))
    )
}
