frame-state 0.2.0

use std::sync::Arc;

use frame_core::capability::{Capability, CapabilityChecker, CheckVerdict};
use frame_core::component::ComponentId;
use haematite::{ConflictPolicy, Hash, checkout};

use crate::codec::entity_key;
use crate::engine::{Storage, entities_at, work_name};
use crate::error::StateError;
use crate::store::{Operation, Shared, operation_error, require_active};
use crate::types::{
    CrossComponentRef, EntityId, MergePolicy, MergeRecord, MetaState, ReferenceTargetState,
};

const SHARD: usize = 0;

pub(crate) trait ResolutionChecker {
    fn component_id(&self) -> ComponentId;

    fn check(
        &self,
        capability: &Capability,
    ) -> Result<CheckVerdict, frame_core::capability::CapabilityCheckError>;
}

impl ResolutionChecker for CapabilityChecker {
    fn component_id(&self) -> ComponentId {
        self.component_id()
    }

    fn check(
        &self,
        capability: &Capability,
    ) -> Result<CheckVerdict, frame_core::capability::CapabilityCheckError> {
        self.check(capability)
    }
}

/// Component capability scoped to one Namespace branch and one incarnation.
///
/// The handle exposes no branch-name argument, registry, node store, or arbitrary
/// root checkout. It can name a foreign component only through capability-checked
/// read resolution: naming is not authority; a matching grant row must be present
/// at resolution time. Foreign writes remain inexpressible. Every consuming
/// operation validates its captured incarnation against durable meta state.
pub struct ComponentStoreHandle {
    shared: Arc<Shared>,
    component: ComponentId,
    incarnation: u64,
}

impl ComponentStoreHandle {
    pub(crate) const fn new(shared: Arc<Shared>, component: ComponentId, incarnation: u64) -> Self {
        Self {
            shared,
            component,
            incarnation,
        }
    }

    /// Returns the identity to which this capability is permanently scoped.
    #[must_use]
    pub const fn component_id(&self) -> ComponentId {
        self.component
    }

    /// Returns the incarnation captured when this handle was issued.
    #[must_use]
    pub const fn incarnation(&self) -> u64 {
        self.incarnation
    }

    /// Buffers a content-addressed entity in this component's Namespace.
    ///
    /// Call [`Self::commit`] to durably materialize a batch. Reads through this
    /// handle observe its buffer immediately.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed branch failures.
    pub fn put(&self, entity: &[u8]) -> Result<EntityId, StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                storage.put_entity(self.component, entity)
            })
    }

    /// Reads an entity through this component's buffered branch view.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed branch failures.
    pub fn get(&self, entity: EntityId) -> Result<Option<Vec<u8>>, StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                storage.get_entity(self.component, entity)
            })
    }

    /// Resolves one link against the target's current committed root.
    ///
    /// The supplied checker must belong to this handle's component. Resolution
    /// follows the consuming-act law in [`frame_core::capability`]: it performs
    /// exactly one fresh target grant check before inspecting target state, then
    /// creates, reads, and drops the checkout view internally. Only owned bytes
    /// leave this call. Naming a target is not authority, self-targets receive no
    /// special case, and ordinary own-namespace reads should use [`Self::get`].
    ///
    /// # Errors
    ///
    /// Returns a typed binding refusal before checking when identities differ; a
    /// verbatim capability denial; a stale-reader refusal; a dangling target with
    /// its durable state; or a typed check, synchronization, branch, or checkout
    /// failure. An absent entity in an active target returns `Ok(None)`.
    pub fn resolve_cross_component(
        &self,
        checker: &CapabilityChecker,
        reference: CrossComponentRef,
    ) -> Result<Option<Vec<u8>>, StateError> {
        self.resolve_cross_component_with(checker, reference)
    }

    pub(crate) fn resolve_cross_component_with(
        &self,
        checker: &impl ResolutionChecker,
        reference: CrossComponentRef,
    ) -> Result<Option<Vec<u8>>, StateError> {
        if checker.component_id() != self.component {
            return Err(StateError::CapabilityBinding {
                handle_component: self.component,
                checker_component: checker.component_id(),
            });
        }
        let capability = Capability::CrossComponentRead {
            target: reference.target,
        };
        match checker.check(&capability)? {
            CheckVerdict::Allowed => {
                self.shared
                    .resolve_cross_component(self.component, self.incarnation, reference)
            }
            CheckVerdict::Denied(denial) => Err(StateError::CapabilityDenied(denial)),
        }
    }

    /// Buffers deletion of an entity from this component's Namespace.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed branch failures.
    pub fn delete(&self, entity: EntityId) -> Result<(), StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                storage.delete_entity(self.component, entity)
            })
    }

    /// Enumerates committed entities in fixed digest order.
    ///
    /// Pending puts/deletes become enumerable after [`Self::commit`]; direct
    /// [`Self::get`] calls still observe them before commit.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates checkout/corruption errors.
    pub fn enumerate(&self) -> Result<Vec<(EntityId, Vec<u8>)>, StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                storage.enumerate_entities(self.component)
            })
    }

    /// Durably commits all pending mutations as one Namespace atomicity unit.
    ///
    /// The commit's fsync barrier is shared-store-wide, even though only this
    /// branch's dirty shards are materialized.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed commit failures.
    pub fn commit(&self) -> Result<Hash, StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                storage.commit_namespace(self.component)
            })
    }

    /// Creates a durable Work-kind child inside this Namespace lineage.
    ///
    /// `name` is hex-encoded into the durable branch name; the returned handle
    /// cannot name or widen to another Work or Namespace branch.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional/concurrent operations, duplicate branches,
    /// and every engine policy or ref-store failure.
    pub fn fork_work(&self, name: &str) -> Result<WorkStoreHandle, StateError> {
        self.shared
            .fork_work(self.component, self.incarnation, name)
    }

    /// Merges an owned Work branch under the schema-declared policy and writes
    /// its audit record into this Namespace before removing the Work ref.
    ///
    /// # Errors
    ///
    /// Refuses foreign/stale/concurrent handles and preserves typed engine
    /// policy, merge, commit, and ref errors.
    pub fn merge_work(&self, work: &WorkStoreHandle) -> Result<MergeRecord, StateError> {
        self.shared
            .merge_work(self.component, self.incarnation, work)
    }

    /// Lists committed merge audit records in sequence order.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed decoding errors.
    pub fn merge_records(&self) -> Result<Vec<MergeRecord>, StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                storage.merge_records(self.component)
            })
    }

    /// Returns this Namespace's current committed root.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed storage failures.
    pub fn current_root(&self) -> Result<Hash, StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                Ok(storage.namespace(self.component)?.current_root())
            })
    }

    /// Performs one root-hash comparison against committed state.
    ///
    /// Equal hashes prove unchanged; movement does not prove a logical change.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed storage failures.
    pub fn has_changed(&self, since_root: Hash) -> Result<bool, StateError> {
        self.current_root().map(|root| root != since_root)
    }
}

/// Mutable capability for one speculative Work branch in one incarnation.
pub struct WorkStoreHandle {
    shared: Arc<Shared>,
    component: ComponentId,
    incarnation: u64,
    branch: String,
}

impl WorkStoreHandle {
    pub(crate) const fn new(
        shared: Arc<Shared>,
        component: ComponentId,
        incarnation: u64,
        branch: String,
    ) -> Self {
        Self {
            shared,
            component,
            incarnation,
            branch,
        }
    }

    pub(crate) const fn component(&self) -> ComponentId {
        self.component
    }

    pub(crate) const fn incarnation(&self) -> u64 {
        self.incarnation
    }

    pub(crate) fn branch(&self) -> &str {
        &self.branch
    }

    pub(crate) fn shared_ref(&self) -> &Shared {
        &self.shared
    }

    /// Buffers a content-addressed entity on this Work branch.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed branch failures.
    pub fn put(&self, entity: &[u8]) -> Result<EntityId, StateError> {
        let id = EntityId::of(entity);
        self.with_work(|storage, branch| {
            storage
                .work_branch(branch)?
                .put(SHARD, entity_key(id), entity)?;
            Ok(id)
        })
    }

    /// Reads an entity through this Work branch's buffered view.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed branch failures.
    pub fn get(&self, entity: EntityId) -> Result<Option<Vec<u8>>, StateError> {
        self.with_work(|storage, branch| {
            Ok(storage
                .work_branch(branch)?
                .get(SHARD, &storage.nodes, &entity_key(entity))?)
        })
    }

    /// Buffers deletion of an entity from this Work branch.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed branch failures.
    pub fn delete(&self, entity: EntityId) -> Result<(), StateError> {
        self.with_work(|storage, branch| {
            storage
                .work_branch(branch)?
                .delete(SHARD, entity_key(entity))?;
            Ok(())
        })
    }

    /// Enumerates this Work branch's committed entities.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates checkout/corruption errors.
    pub fn enumerate(&self) -> Result<Vec<(EntityId, Vec<u8>)>, StateError> {
        self.with_work(|storage, branch| {
            let root = storage.work_branch(branch)?.current_root();
            entities_at(&storage.nodes, root)
        })
    }

    /// Durably commits pending speculative mutations and returns the Work root.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed commit failures.
    pub fn commit(&self) -> Result<Hash, StateError> {
        self.with_work(Storage::commit_work)
    }

    /// Returns this Work branch's current committed root.
    ///
    /// # Errors
    ///
    /// Refuses stale/transitional handles and propagates typed branch failures.
    pub fn current_root(&self) -> Result<Hash, StateError> {
        self.with_work(|storage, branch| Ok(storage.work_branch(branch)?.current_root()))
    }

    fn with_work<T>(
        &self,
        action: impl FnOnce(&mut Storage, &str) -> Result<T, StateError>,
    ) -> Result<T, StateError> {
        self.shared
            .with_active(self.component, self.incarnation, |storage| {
                action(storage, &self.branch)
            })
    }
}

impl Shared {
    pub(crate) fn with_active<T>(
        &self,
        component: ComponentId,
        incarnation: u64,
        action: impl FnOnce(&mut Storage) -> Result<T, StateError>,
    ) -> Result<T, StateError> {
        let mut storage = self
            .storage
            .lock()
            .map_err(|_| StateError::SynchronizationPoisoned)?;
        require_active(&storage, component, Some(incarnation))?;
        action(&mut storage)
    }

    fn resolve_cross_component(
        &self,
        reader: ComponentId,
        incarnation: u64,
        reference: CrossComponentRef,
    ) -> Result<Option<Vec<u8>>, StateError> {
        let mut storage = self
            .storage
            .lock()
            .map_err(|_| StateError::SynchronizationPoisoned)?;
        require_active(&storage, reader, Some(incarnation))?;
        let Some(record) = storage.meta_record(reference.target)? else {
            return Err(StateError::DanglingReference {
                target: reference.target,
                state: ReferenceTargetState::NeverInstalled,
            });
        };
        match record.state {
            MetaState::Active => {
                let root = storage.namespace(reference.target)?.current_root();
                Ok(checkout(&storage.nodes, root).get(&entity_key(reference.entity))?)
            }
            MetaState::Installing => Err(StateError::DanglingReference {
                target: reference.target,
                state: ReferenceTargetState::Installing,
            }),
            MetaState::Archiving { generation, .. } => Err(StateError::DanglingReference {
                target: reference.target,
                state: ReferenceTargetState::Archiving { generation },
            }),
            MetaState::Archived { generation, .. } => Err(StateError::DanglingReference {
                target: reference.target,
                state: ReferenceTargetState::Archived { generation },
            }),
        }
    }

    pub(crate) fn fork_work(
        self: &Arc<Self>,
        component: ComponentId,
        incarnation: u64,
        name: &str,
    ) -> Result<WorkStoreHandle, StateError> {
        let branch = work_name(component, name);
        self.with_facade_operation(component, Operation::Work, |storage| {
            require_active(storage, component, Some(incarnation))?;
            storage.fork_work(component, &branch)
        })?;
        Ok(WorkStoreHandle::new(
            Arc::clone(self),
            component,
            incarnation,
            branch,
        ))
    }

    pub(crate) fn merge_work(
        &self,
        component: ComponentId,
        incarnation: u64,
        work: &WorkStoreHandle,
    ) -> Result<MergeRecord, StateError> {
        if component != work.component()
            || incarnation != work.incarnation()
            || !std::ptr::eq(self, work.shared_ref())
        {
            return Err(StateError::ForeignWorkHandle);
        }
        let policy = {
            let storage = self
                .storage
                .lock()
                .map_err(|_| StateError::SynchronizationPoisoned)?;
            require_active(&storage, component, Some(incarnation))?
                .schema
                .merge_policy
        };
        let engine_policy = match &policy {
            MergePolicy::Lww => ConflictPolicy::Lww,
            MergePolicy::VectorClock => {
                return Err(StateError::UnsupportedPolicy { policy });
            }
            MergePolicy::Custom(name) => {
                let resolver = self
                    .resolvers
                    .lock()
                    .map_err(|_| StateError::SynchronizationPoisoned)?
                    .get(name)
                    .copied()
                    .ok_or_else(|| StateError::ResolverNotRegistered { name: name.clone() })?;
                ConflictPolicy::Custom(resolver)
            }
        };
        self.with_facade_operation(component, Operation::Work, |storage| {
            require_active(storage, component, Some(incarnation))?;
            storage.merge_work(component, work.branch(), &policy, &engine_policy)
        })
    }

    fn with_facade_operation<T>(
        &self,
        component: ComponentId,
        requested: Operation,
        action: impl FnOnce(&mut Storage) -> Result<T, StateError>,
    ) -> Result<T, StateError> {
        {
            let mut operations = self
                .operations
                .lock()
                .map_err(|_| StateError::SynchronizationPoisoned)?;
            if let Some(active) = operations.get(&component) {
                return Err(operation_error(component, *active));
            }
            operations.insert(component, requested);
        }
        let result = self
            .storage
            .lock()
            .map_err(|_| StateError::SynchronizationPoisoned)
            .and_then(|mut storage| action(&mut storage));
        self.operations
            .lock()
            .map_err(|_| StateError::SynchronizationPoisoned)?
            .remove(&component);
        result
    }
}