linkrs-storage 0.1.0

use super::*;

#[test]
fn test_create_and_list_spaces() {
    let mut storage = create_test_storage();

    let mut space1 = SpaceInfo::new("space1".to_string()).with_vid_type(DataType::BigInt);
    let mut space2 = SpaceInfo::new("space2".to_string()).with_vid_type(DataType::String);
    storage.create_space(&mut space1).unwrap();
    storage.create_space(&mut space2).unwrap();

    let spaces = storage.list_spaces().unwrap();
    assert_eq!(spaces.len(), 2);
    assert!(storage.space_exists("space1"));
    assert!(storage.space_exists("space2"));
    assert!(!storage.space_exists("space3"));

    assert_eq!(storage.get_space_id("space1").unwrap(), 1);
}

#[test]
fn test_drop_space_cleans_tags_and_edge_types() {
    let mut storage = create_test_storage();
    setup_space(&mut storage);
    setup_person_tag(&mut storage);
    setup_knows_edge(&mut storage);

    storage.drop_space("test_space").unwrap();
    assert!(!storage.space_exists("test_space"));
}

#[test]
fn test_create_and_get_tag() {
    let mut storage = create_test_storage();
    setup_space(&mut storage);

    let tag_id = setup_person_tag(&mut storage);
    assert!(tag_id > 0);

    let tag = storage.get_tag("test_space", "Person").unwrap();
    assert!(tag.is_some());
    assert_eq!(tag.as_ref().unwrap().tag_name, "Person");
    // id (primary-key mirror) + name + age.
    assert_eq!(tag.as_ref().unwrap().properties.len(), 3);

    let tags = storage.list_tags("test_space").unwrap();
    assert_eq!(tags.len(), 1);
}

#[test]
fn test_drop_tag_removes_tag() {
    let mut storage = create_test_storage();
    setup_space(&mut storage);
    setup_person_tag(&mut storage);

    storage.drop_tag("test_space", "Person").unwrap();
    assert!(storage.get_tag("test_space", "Person").unwrap().is_none());
}

#[test]
fn test_create_and_get_edge_type() {
    let mut storage = create_test_storage();
    setup_space(&mut storage);
    setup_person_tag(&mut storage);

    let edge_id = setup_knows_edge(&mut storage);
    assert!(edge_id > 0);

    let edge = storage.get_edge_type("test_space", "KNOWS").unwrap();
    assert!(edge.is_some());
    assert_eq!(edge.as_ref().unwrap().edge_type_name, "KNOWS");

    let edges = storage.list_edge_types("test_space").unwrap();
    assert_eq!(edges.len(), 1);
}

#[test]
fn test_same_schema_names_are_isolated_by_space() {
    let mut storage = create_test_storage();
    let mut alpha = SpaceInfo::new("alpha".to_string()).with_vid_type(DataType::BigInt);
    let mut beta = SpaceInfo::new("beta".to_string()).with_vid_type(DataType::BigInt);
    storage.create_space(&mut alpha).unwrap();
    storage.create_space(&mut beta).unwrap();

    let tag = linkrs_core::types::TagInfo::new("Person".to_string()).with_properties(vec![
        PropertyDef::new("id".into(), DataType::BigInt),
        PropertyDef::new("name".into(), DataType::String),
    ]);
    let alpha_tag_id = storage.create_tag("alpha", &tag).unwrap();
    let beta_tag_id = storage.create_tag("beta", &tag).unwrap();
    assert_ne!(alpha_tag_id, beta_tag_id);

    let edge_type = EdgeTypeInfo::new("KNOWS".to_string())
        .with_src_tag("Person".to_string())
        .with_dst_tag("Person".to_string());
    let alpha_edge_id = storage.create_edge_type("alpha", &edge_type).unwrap();
    let beta_edge_id = storage.create_edge_type("beta", &edge_type).unwrap();
    assert_ne!(alpha_edge_id, beta_edge_id);

    storage
        .insert_vertex(
            "alpha",
            Vertex::new(
                VertexId::try_from_int64(1).expect("test vertex id"),
                Tag::new(
                    "Person".to_string(),
                    vec![("name".into(), Value::string("Alice"))]
                        .into_iter()
                        .collect(),
                ),
            ),
        )
        .unwrap();
    storage
        .insert_vertex(
            "beta",
            Vertex::new(
                VertexId::try_from_int64(1).expect("test vertex id"),
                Tag::new(
                    "Person".to_string(),
                    vec![("name".into(), Value::string("Bob"))]
                        .into_iter()
                        .collect(),
                ),
            ),
        )
        .unwrap();
    storage
        .insert_vertex(
            "alpha",
            Vertex::new(
                VertexId::try_from_int64(2).expect("test vertex id"),
                Tag::new(
                    "Person".to_string(),
                    vec![("name".into(), Value::string("Carol"))]
                        .into_iter()
                        .collect(),
                ),
            ),
        )
        .unwrap();
    storage
        .insert_vertex(
            "beta",
            Vertex::new(
                VertexId::try_from_int64(2).expect("test vertex id"),
                Tag::new(
                    "Person".to_string(),
                    vec![("name".into(), Value::string("Dave"))]
                        .into_iter()
                        .collect(),
                ),
            ),
        )
        .unwrap();

    storage
        .insert_edge(
            "alpha",
            Edge::new(
                VertexId::try_from_int64(1).expect("test vertex id"),
                VertexId::try_from_int64(2).expect("test vertex id"),
                "KNOWS".to_string(),
                0,
                std::collections::HashMap::new(),
            ),
        )
        .unwrap();
    storage
        .insert_edge(
            "beta",
            Edge::new(
                VertexId::try_from_int64(1).expect("test vertex id"),
                VertexId::try_from_int64(2).expect("test vertex id"),
                "KNOWS".to_string(),
                0,
                std::collections::HashMap::new(),
            ),
        )
        .unwrap();

    let alpha_vertex = storage
        .get_vertex(
            "alpha",
            "Person",
            &VertexId::try_from_int64(1).expect("test vertex id"),
        )
        .unwrap()
        .unwrap();
    let beta_vertex = storage
        .get_vertex(
            "beta",
            "Person",
            &VertexId::try_from_int64(1).expect("test vertex id"),
        )
        .unwrap()
        .unwrap();
    assert_eq!(
        alpha_vertex.property_value("name"),
        Some(Value::string("Alice"))
    );
    assert_eq!(
        beta_vertex.property_value("name"),
        Some(Value::string("Bob"))
    );

    assert_eq!(
        storage
            .scan_vertices_by_tag("alpha", "Person")
            .unwrap()
            .len(),
        2
    );
    assert_eq!(
        storage
            .scan_vertices_by_tag("beta", "Person")
            .unwrap()
            .len(),
        2
    );
    assert_eq!(
        storage.scan_edges_by_type("alpha", "KNOWS").unwrap().len(),
        1
    );
    assert_eq!(
        storage.scan_edges_by_type("beta", "KNOWS").unwrap().len(),
        1
    );
}

#[test]
fn test_drop_edge_type() {
    let mut storage = create_test_storage();
    setup_space(&mut storage);
    setup_person_tag(&mut storage);
    setup_knows_edge(&mut storage);

    storage.drop_edge_type("test_space", "KNOWS").unwrap();
    assert!(storage
        .get_edge_type("test_space", "KNOWS")
        .unwrap()
        .is_none());
}

#[test]
fn test_update_edge_endpoints() {
    let mut storage = create_test_storage();
    setup_space(&mut storage);
    setup_person_tag(&mut storage);
    setup_knows_edge(&mut storage);

    assert!(storage
        .update_edge_endpoints("test_space", "KNOWS", "Person", "Person")
        .unwrap());
    let edge = storage
        .get_edge_type("test_space", "KNOWS")
        .unwrap()
        .expect("edge type exists");
    assert_eq!(edge.src_tag_name, "Person");
    assert_eq!(edge.dst_tag_name, "Person");

    // Clearing restores the unconstrained state.
    assert!(storage
        .update_edge_endpoints("test_space", "KNOWS", "", "")
        .unwrap());
    let edge = storage
        .get_edge_type("test_space", "KNOWS")
        .unwrap()
        .expect("edge type exists");
    assert!(edge.src_tag_name.is_empty());
    assert!(edge.dst_tag_name.is_empty());

    // Unknown endpoint tags are rejected.
    assert!(storage
        .update_edge_endpoints("test_space", "KNOWS", "Ghost", "Person")
        .is_err());
    // Unknown edge types are rejected like other schema alters.
    assert!(storage
        .update_edge_endpoints("test_space", "GHOST", "", "")
        .is_err());
}

#[test]
fn test_schema_wal_replays_create_and_alter_after_restart() {
    let temp_dir = tempfile::TempDir::new().expect("Failed to create temp dir");
    let work_dir = temp_dir.path().to_path_buf();

    {
        let mut storage = GraphStorage::new_with_path(work_dir.clone())
            .expect("Failed to create persistent GraphStorage");
        setup_space(&mut storage);
        storage
            .save_to_disk()
            .expect("Failed to persist base schema");

        let tag = linkrs_core::types::TagInfo::new("Person".to_string()).with_properties(vec![
            PropertyDef::new("name".into(), DataType::String),
            PropertyDef::new("age".into(), DataType::BigInt),
        ]);
        storage
            .create_tag("test_space", &tag)
            .expect("Failed to create tag");

        let edge = EdgeTypeInfo::new("KNOWS".to_string())
            .with_src_tag("Person".to_string())
            .with_dst_tag("Person".to_string())
            .with_properties(vec![PropertyDef::new("since".into(), DataType::Int)]);
        storage
            .create_edge_type("test_space", &edge)
            .expect("Failed to create edge type");

        storage
            .alter_tag(
                "test_space",
                "Person",
                vec![PropertyDef::new("email".into(), DataType::String)],
                vec!["age".to_string()],
                None,
            )
            .expect("Failed to alter tag");
        storage
            .alter_edge_type(
                "test_space",
                "KNOWS",
                vec![PropertyDef::new("weight".into(), DataType::Double)],
                vec!["since".to_string()],
                None,
            )
            .expect("Failed to alter edge type");

        storage.flush().expect("Failed to sync WAL");
    }

    let storage = GraphStorage::open(work_dir).expect("Failed to reopen persistent GraphStorage");

    let tag = storage
        .get_tag("test_space", "Person")
        .expect("Failed to load tag")
        .expect("Tag should exist after recovery");
    let tag_props: Vec<String> = tag
        .properties
        .iter()
        .map(|prop| prop.name.clone())
        .collect();
    assert!(tag_props.contains(&"name".to_string()));
    assert!(tag_props.contains(&"email".to_string()));
    assert!(!tag_props.contains(&"age".to_string()));

    let edge = storage
        .get_edge_type("test_space", "KNOWS")
        .expect("Failed to load edge type")
        .expect("Edge type should exist after recovery");
    let edge_props: Vec<String> = edge
        .properties
        .iter()
        .map(|prop| prop.name.clone())
        .collect();
    assert!(edge_props.contains(&"weight".to_string()));
    assert!(!edge_props.contains(&"since".to_string()));
}

#[test]
fn test_schema_wal_replays_drop_after_restart() {
    let temp_dir = tempfile::TempDir::new().expect("Failed to create temp dir");
    let work_dir = temp_dir.path().to_path_buf();

    {
        let mut storage = GraphStorage::new_with_path(work_dir.clone())
            .expect("Failed to create persistent GraphStorage");
        setup_space(&mut storage);
        storage
            .save_to_disk()
            .expect("Failed to persist base schema");

        setup_person_tag(&mut storage);
        let edge = EdgeTypeInfo::new("KNOWS".to_string())
            .with_src_tag("Person".to_string())
            .with_dst_tag("Person".to_string())
            .with_properties(vec![PropertyDef::new("since".into(), DataType::Int)]);
        storage
            .create_edge_type("test_space", &edge)
            .expect("Failed to create edge type");

        storage
            .drop_edge_type("test_space", "KNOWS")
            .expect("Failed to drop edge type");
        storage
            .drop_tag("test_space", "Person")
            .expect("Failed to drop tag");

        storage.flush().expect("Failed to sync WAL");
    }

    let storage = GraphStorage::open(work_dir).expect("Failed to reopen persistent GraphStorage");

    assert!(storage
        .get_tag("test_space", "Person")
        .expect("Failed to load tag")
        .is_none());
    assert!(storage
        .get_edge_type("test_space", "KNOWS")
        .expect("Failed to load edge type")
        .is_none());
}

#[test]
fn test_space_wal_replays_create_and_clear_after_restart() {
    let temp_dir = tempfile::TempDir::new().expect("Failed to create temp dir");
    let work_dir = temp_dir.path().to_path_buf();

    {
        let mut storage = GraphStorage::new_with_path(work_dir.clone())
            .expect("Failed to create persistent GraphStorage");
        setup_space(&mut storage);
        setup_person_tag(&mut storage);
        setup_knows_edge(&mut storage);
        storage.flush().expect("Failed to sync WAL");
    }

    {
        let mut storage = GraphStorage::open(work_dir.clone()).expect("Failed to reopen storage");
        // The replayed schema must still resolve the space id consistently.
        assert_eq!(
            storage
                .get_space_id("test_space")
                .expect("space id should exist"),
            storage
                .get_space("test_space")
                .expect("Failed to load space")
                .expect("space should exist after replay")
                .space_id
        );

        assert!(storage.space_exists("test_space"));
        assert_eq!(
            storage
                .list_tags("test_space")
                .expect("Failed to list tags")
                .len(),
            1
        );
        assert_eq!(
            storage
                .list_edge_types("test_space")
                .expect("Failed to list edge types")
                .len(),
            1
        );

        storage
            .save_to_disk()
            .expect("Failed to persist recovered schema");

        storage
            .clear_space("test_space")
            .expect("Failed to clear space");
        storage.flush().expect("Failed to sync WAL");
    }

    let storage = GraphStorage::open(work_dir).expect("Failed to reopen persistent GraphStorage");

    let space = storage
        .get_space("test_space")
        .expect("Failed to load space")
        .expect("Space should still exist after clear");
    assert_eq!(space.comment, Some("test".to_string()));
    assert_eq!(
        storage
            .list_tags("test_space")
            .expect("Failed to list tags")
            .len(),
        0
    );
    assert_eq!(
        storage
            .list_edge_types("test_space")
            .expect("Failed to list edge types")
            .len(),
        0
    );
}

#[test]
fn test_space_wal_replays_drop_after_restart() {
    let temp_dir = tempfile::TempDir::new().expect("Failed to create temp dir");
    let work_dir = temp_dir.path().to_path_buf();

    {
        let mut storage = GraphStorage::new_with_path(work_dir.clone())
            .expect("Failed to create persistent GraphStorage");
        setup_space(&mut storage);
        setup_person_tag(&mut storage);
        setup_knows_edge(&mut storage);
        storage
            .drop_space("test_space")
            .expect("Failed to drop space");
        storage.flush().expect("Failed to sync WAL");
    }

    let storage = GraphStorage::open(work_dir).expect("Failed to reopen persistent GraphStorage");

    assert!(!storage.space_exists("test_space"));
    assert!(storage
        .list_spaces()
        .expect("Failed to list spaces")
        .is_empty());
}

#[test]
fn test_create_and_drop_tag_index() {
    let mut storage = create_test_storage();
    setup_space(&mut storage);
    setup_person_tag(&mut storage);

    let index = Index::new(IndexConfig {
        id: 1,
        name: "person_name_idx".to_string(),
        space_id: 1,
        schema_name: "Person".to_string(),
        fields: vec![IndexField::new(
            "name".to_string(),
            Value::string(""),
            false,
        )],
        properties: vec![],
        index_type: IndexType::TagIndex,
        is_unique: false,
        covering: false,
        partial_condition: None,
    });
    storage.create_tag_index("test_space", &index).unwrap();

    let indexes = storage.list_tag_indexes("test_space").unwrap();
    assert_eq!(indexes.len(), 1);

    storage
        .drop_tag_index("test_space", "person_name_idx")
        .unwrap();
    let indexes = storage.list_tag_indexes("test_space").unwrap();
    assert_eq!(indexes.len(), 0);
}

#[test]
fn test_snapshot_admin_methods() {
    let (_temp_dir, storage) = create_persistent_storage();

    let initial_stats = storage.snapshot_stats();
    assert_eq!(initial_stats.snapshot_count, 0);
    assert_eq!(initial_stats.total_size_bytes, 0);
    assert_eq!(initial_stats.latest_snapshot_id, None);

    let checkpoint = storage
        .create_checkpoint()
        .expect("checkpoint should succeed")
        .expect("persistence should be enabled");

    assert!(checkpoint.snapshot_created);
    assert!(storage
        .verify_snapshot(checkpoint.checkpoint_id)
        .expect("snapshot verification should succeed"));

    let stats = storage.snapshot_stats();
    assert_eq!(stats.snapshot_count, 1);
    assert_eq!(stats.latest_snapshot_id, Some(checkpoint.checkpoint_id));

    let deleted = storage
        .cleanup_snapshots()
        .expect("snapshot cleanup should succeed");
    assert_eq!(deleted, 0);
}

#[test]
fn test_checkpoint_publishes_snapshot_watermark() {
    let (temp_dir, mut storage) = create_persistent_storage();
    setup_space(&mut storage);
    setup_person_tag(&mut storage);
    insert_test_vertex(&mut storage, 1, "Alice");

    let stats = storage
        .create_checkpoint()
        .expect("checkpoint should succeed")
        .expect("persistence should be enabled");
    assert!(
        stats.snapshot_timestamp > 0,
        "checkpoint must record its MVCC snapshot timestamp"
    );

    // The published manifest carries the same snapshot plus a reclaim LSN,
    // which is what watermark captures (and post-restart lazy loads) read.
    let manager = linkrs_sync::checkpoint_manifest::CheckpointManifestManager::new(
        temp_dir.path().join("checkpoint").join("manifests"),
    );
    let manifest = manager
        .load_latest()
        .expect("manifest load should succeed")
        .expect("a checkpoint manifest should be published");
    assert_eq!(manifest.snapshot_timestamp, Some(stats.snapshot_timestamp));
    assert_ne!(
        manifest.safe_lsn,
        linkrs_core::types::CommitLsn::ZERO,
        "a checkpoint over WAL writes must publish a reclaim LSN"
    );
}