frame-state 0.2.0

use haematite::{
    BranchCommitError, BranchKind, BranchPolicyError, BranchRefStore, BranchRegistry,
    CommitDurability, CommitRequest, ConflictPolicy, DiskStore, LeafNode, Node, NodeStore,
    SnapshotRegistry, commit_branch, create_branch_with_kind, current_timestamp,
    fork_from_with_kind, merge,
};

use super::{TestDirectory, TestResult, component, install};
use crate::store::Operation;
use crate::{ComponentSchema, MergePolicy, StateError};

const SHARD: usize = 0;

#[test]
fn work_merge_uses_declared_policy_records_roots_and_removes_ref() -> TestResult {
    let directory = TestDirectory::new("work-merge")?;
    let store = crate::ComponentStateStore::open(directory.path())?;
    let id = component("merge");
    let namespace = install(&store, id)?;
    let before = namespace.current_root()?;
    let work = namespace.fork_work("candidate")?;
    let entity = work.put(b"landed")?;
    let source_root = work.commit()?;

    let record = namespace.merge_work(&work)?;
    assert_eq!(record.sequence, 0);
    assert_eq!(record.policy, MergePolicy::Lww);
    assert_eq!(record.source_root, source_root);
    assert_eq!(record.destination_root_before, before);
    assert_ne!(namespace.current_root()?, before);
    assert_eq!(namespace.get(entity)?, Some(b"landed".to_vec()));
    assert_eq!(namespace.merge_records()?, vec![record.clone()]);

    let refs = BranchRefStore::open(directory.path().join("refs"))?;
    assert!(refs.get(&record.source_branch).is_none());
    Ok(())
}

#[test]
fn raw_engine_refuses_namespace_nesting_fork_and_cross_namespace_merge_with_names_and_kinds()
-> TestResult {
    let directory = TestDirectory::new("raw-policy")?;
    let (mut nodes, mut refs, registry, root) = raw_engine(directory.path())?;
    let namespace_a = create_branch_with_kind(
        "namespace-a",
        [(SHARD, root)],
        BranchKind::Namespace,
        &mut refs,
        &registry,
        current_timestamp(),
    )?;
    let namespace_b = create_branch_with_kind(
        "namespace-b",
        [(SHARD, root)],
        BranchKind::Namespace,
        &mut refs,
        &registry,
        current_timestamp(),
    )?;
    let work_a = fork_from_with_kind(
        &namespace_a,
        "work-a-17",
        BranchKind::Work,
        &mut refs,
        &registry,
        current_timestamp(),
    )?;

    let fork_error = fork_from_with_kind(
        &namespace_a,
        "namespace-b-child",
        BranchKind::Namespace,
        &mut refs,
        &registry,
        current_timestamp(),
    )
    .expect_err("Namespace-in-Namespace fork must be engine-refused");
    // haematite 0.6.x kind-marker matrix contract: `check_fork_into` refuses
    // every fork whose destination kind is Namespace upfront as
    // `NamespaceFork` (Namespace branches are direct-create-only), before the
    // namespace-boundary check runs. `ForkAcrossNamespaceBoundary` is not
    // reachable through `fork_from_with_kind`: Work children inherit the
    // parent's resolved lineage, so the public route can never present a
    // differing destination lineage — haematite's own matrix covers those
    // cells synthetically against the crate-private `check_fork_into`.
    assert!(matches!(
        fork_error,
        BranchCommitError::Policy(BranchPolicyError::NamespaceFork {
            source,
            source_kind: BranchKind::Namespace,
            destination,
            destination_kind: BranchKind::Namespace,
        }) if source == "namespace-a" && destination == "namespace-b-child"
    ));

    let merge_error = merge(
        &mut nodes,
        &work_a,
        &namespace_b,
        &refs,
        &ConflictPolicy::Lww,
    )
    .expect_err("cross-Namespace raw merge must be engine-refused");
    let rendered = merge_error.to_string();
    assert!(matches!(
        merge_error,
        BranchCommitError::Policy(BranchPolicyError::MergeAcrossNamespaceBoundary {
            source,
            source_kind: BranchKind::Work,
            destination,
            destination_kind: BranchKind::Namespace,
            ..
        }) if source == "work-a-17" && destination == "namespace-b"
    ));
    assert!(rendered.contains("'work-a-17' (Work"));
    assert!(rendered.contains("'namespace-b' (Namespace"));
    Ok(())
}

#[test]
fn nested_work_fork_inherits_namespace_lineage_and_engine_accepts_merge() -> TestResult {
    let directory = TestDirectory::new("nested-work")?;
    let (mut nodes, mut refs, registry, root) = raw_engine(directory.path())?;
    let namespace = create_branch_with_kind(
        "namespace-nested",
        [(SHARD, root)],
        BranchKind::Namespace,
        &mut refs,
        &registry,
        current_timestamp(),
    )?;
    let first = fork_from_with_kind(
        &namespace,
        "work-parent",
        BranchKind::Work,
        &mut refs,
        &registry,
        current_timestamp(),
    )?;
    first.put(SHARD, b"parent", b"committed")?;
    commit_durable(&first, &mut nodes, &registry, &mut refs)?;
    let nested = fork_from_with_kind(
        &first,
        "work-child",
        BranchKind::Work,
        &mut refs,
        &registry,
        current_timestamp(),
    )?;
    nested.put(SHARD, b"child", b"committed")?;
    commit_durable(&nested, &mut nodes, &registry, &mut refs)?;

    let nested_record = refs.get("work-child").expect("nested durable ref");
    assert_eq!(nested_record.kind, BranchKind::Work);
    assert_eq!(
        nested_record.resolved_namespace_lineage(),
        Some("namespace-nested")
    );
    let report = merge(&mut nodes, &nested, &namespace, &refs, &ConflictPolicy::Lww)?;
    assert_ne!(report.merged_root, namespace.current_root());
    let merged = haematite::checkout(&nodes, report.merged_root);
    // The nested fork anchor already contains the parent's unmerged write.
    // Against the Namespace destination, 0.5.0 treats that absence as a
    // destination-side deletion; only the nested child's own delta lands.
    assert_eq!(merged.get(b"parent")?, None);
    assert_eq!(merged.get(b"child")?, Some(b"committed".to_vec()));
    Ok(())
}

#[test]
fn policy_vocabulary_is_refused_at_declaration_time_when_not_executable() -> TestResult {
    let directory = TestDirectory::new("policy-declaration")?;
    let store = crate::ComponentStateStore::open(directory.path())?;
    let vector_id = component("vector");
    assert!(matches!(
        store.declare_schema(
            vector_id,
            ComponentSchema {
                merge_policy: MergePolicy::VectorClock,
            },
        ),
        Err(StateError::UnsupportedPolicy {
            policy: MergePolicy::VectorClock,
        })
    ));
    let custom_id = component("custom-missing");
    assert!(matches!(
        store.declare_schema(
            custom_id,
            ComponentSchema {
                merge_policy: MergePolicy::Custom("absent".to_owned()),
            },
        ),
        Err(StateError::ResolverNotRegistered { name }) if name == "absent"
    ));

    store.register_custom_resolver("registered", prefer_branch)?;
    let registered_id = component("custom-registered");
    store.declare_schema(
        registered_id,
        ComponentSchema {
            merge_policy: MergePolicy::Custom("registered".to_owned()),
        },
    )?;
    let namespace = store.install_storage(registered_id)?;
    let work = namespace.fork_work("custom")?;
    work.put(b"custom-policy-value")?;
    let record = namespace.merge_work(&work)?;
    assert_eq!(record.policy, MergePolicy::Custom("registered".to_owned()));
    Ok(())
}

#[test]
fn concurrent_facade_operations_return_typed_in_progress_refusals() -> TestResult {
    let directory = TestDirectory::new("operation-claims")?;
    let store = std::sync::Arc::new(crate::ComponentStateStore::open(directory.path())?);
    let install_id = component("concurrent-install");
    store.declare_schema(install_id, ComponentSchema::lww())?;
    let storage_guard = store.lock_storage()?;
    let install_store = std::sync::Arc::clone(&store);
    let installing = std::thread::spawn(move || install_store.install_storage(install_id));
    wait_for_operation(&store, install_id, Operation::Install);
    assert!(matches!(
        store.install_storage(install_id),
        Err(StateError::InstallInProgress { component }) if component == install_id
    ));
    drop(storage_guard);
    let _installed = installing.join().expect("install thread")?;

    let storage_guard = store.lock_storage()?;
    let archive_store = std::sync::Arc::clone(&store);
    let archiving = std::thread::spawn(move || archive_store.archive_storage(install_id));
    wait_for_operation(&store, install_id, Operation::Archive);
    assert!(matches!(
        store.archive_storage(install_id),
        Err(StateError::ArchiveInProgress { component }) if component == install_id
    ));
    drop(storage_guard);
    let _archive = archiving.join().expect("archive thread")?;

    let work_id = component("concurrent-work");
    let work_owner = install(&store, work_id)?;
    let storage_guard = store.lock_storage()?;
    let forking = std::thread::spawn(move || work_owner.fork_work("held"));
    wait_for_operation(&store, work_id, Operation::Work);
    assert!(matches!(
        store.archive_storage(work_id),
        Err(StateError::WorkInProgress { component }) if component == work_id
    ));
    drop(storage_guard);
    let _work = forking.join().expect("fork thread")?;
    Ok(())
}

fn wait_for_operation(
    store: &crate::ComponentStateStore,
    component: frame_core::component::ComponentId,
    expected: Operation,
) {
    loop {
        let observed = store
            .shared
            .operations
            .lock()
            .expect("operation map")
            .get(&component)
            .copied();
        if observed.is_some_and(|active| {
            std::mem::discriminant(&active) == std::mem::discriminant(&expected)
        }) {
            return;
        }
        std::thread::yield_now();
    }
}

fn raw_engine(
    path: &std::path::Path,
) -> Result<(DiskStore, BranchRefStore, BranchRegistry, haematite::Hash), Box<dyn std::error::Error>>
{
    let mut nodes = DiskStore::new(path.join("nodes"))?;
    let root = nodes.put(&Node::Leaf(LeafNode::new(Vec::new())?))?;
    nodes.sync_dirty_dirs()?;
    let refs = BranchRefStore::open(path.join("refs"))?;
    Ok((nodes, refs, BranchRegistry::new(), root))
}

fn commit_durable(
    branch: &haematite::BranchHandle,
    nodes: &mut DiskStore,
    registry: &BranchRegistry,
    refs: &mut BranchRefStore,
) -> Result<(), BranchCommitError> {
    commit_branch(
        branch,
        nodes,
        registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    Ok(())
}

fn prefer_branch(
    _key: &[u8],
    _ancestor: Option<&[u8]>,
    _parent: Option<&[u8]>,
    branch: Option<&[u8]>,
) -> Option<Vec<u8>> {
    branch.map(<[u8]>::to_vec)
}

#[test]
fn snapshot_registry_fixture_remains_openable() -> TestResult {
    let directory = TestDirectory::new("snapshot-open")?;
    let snapshots = SnapshotRegistry::open(directory.path().join("snapshots"))?;
    assert!(snapshots.is_empty());
    Ok(())
}