frame-state 0.2.0

use std::collections::HashMap;
use std::path::Path;
use std::sync::{Arc, Mutex};

use frame_core::component::ComponentId;
use haematite::{BranchKind, CustomMergeFn, Hash, checkout};

use crate::engine::{Storage, archive_name, entities_at, namespace_name};
use crate::error::{ReconcileInconsistency, StateError};
use crate::handle::ComponentStoreHandle;
use crate::types::{
    Archive, ComponentSchema, EntityId, MetaRecord, MetaState, ReconcileAction, ReconcileReport,
};

/// Facade that exclusively owns haematite's mutable store and registries.
pub struct ComponentStateStore {
    pub(crate) shared: Arc<Shared>,
}

pub(crate) struct Shared {
    pub(crate) storage: Mutex<Storage>,
    pub(crate) operations: Mutex<HashMap<ComponentId, Operation>>,
    pub(crate) declarations: Mutex<HashMap<ComponentId, ComponentSchema>>,
    pub(crate) resolvers: Mutex<HashMap<String, CustomMergeFn>>,
}

#[derive(Clone, Copy)]
pub(crate) enum Operation {
    Install,
    Archive,
    Work,
}

impl ComponentStateStore {
    /// Opens or creates a durable state store rooted at `path`.
    ///
    /// Boot does not infer anything from frame-core's runtime registry. Call
    /// [`Self::reconcile`] explicitly before starting components.
    ///
    /// # Errors
    ///
    /// Returns typed filesystem, node, ref, snapshot, or branch errors. Corrupt
    /// durable records fail open rather than disappearing.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StateError> {
        Ok(Self {
            shared: Arc::new(Shared {
                storage: Mutex::new(Storage::open(path.as_ref())?),
                operations: Mutex::new(HashMap::new()),
                declarations: Mutex::new(HashMap::new()),
                resolvers: Mutex::new(HashMap::new()),
            }),
        })
    }

    /// Performs the ordered durable install: Installing meta, Namespace ref,
    /// then Active meta, and returns an incarnation-fenced scoped handle.
    ///
    /// The embedding host calls this after frame-core registration and before
    /// starting component processes; lifecycle events are not involved.
    ///
    /// # Errors
    ///
    /// Refuses undeclared schemas, active/archiving state, concurrent facade
    /// operations, and all typed substrate failures. A crash-left Installing
    /// record with its durable branch present is resumed using its recorded schema.
    pub fn install_storage(
        &self,
        component: ComponentId,
    ) -> Result<ComponentStoreHandle, StateError> {
        self.with_operation(component, Operation::Install, || {
            let mut storage = self.lock_storage()?;
            let incarnation = match storage.meta_record(component)? {
                None => 0,
                Some(mut record) => match record.state {
                    MetaState::Archived { .. } => {
                        record.incarnation.checked_add(1).ok_or_else(|| {
                            StateError::CorruptRecord {
                                detail: format!("component {component} incarnation overflow"),
                            }
                        })?
                    }
                    MetaState::Installing if storage.namespace_exists(component) => {
                        let incarnation = record.incarnation;
                        record.state = MetaState::Active;
                        storage.write_meta(&record)?;
                        drop(storage);
                        self.shared
                            .declarations
                            .lock()
                            .map_err(|_| StateError::SynchronizationPoisoned)?
                            .remove(&component);
                        return Ok(ComponentStoreHandle::new(
                            Arc::clone(&self.shared),
                            component,
                            incarnation,
                        ));
                    }
                    MetaState::Installing => {
                        return Err(StateError::InstallInProgress { component });
                    }
                    MetaState::Archiving { .. } => {
                        return Err(StateError::ArchiveInProgress { component });
                    }
                    MetaState::Active => return Err(StateError::AlreadyActive { component }),
                },
            };
            let schema = self
                .shared
                .declarations
                .lock()
                .map_err(|_| StateError::SynchronizationPoisoned)?
                .get(&component)
                .cloned()
                .ok_or(StateError::SchemaNotDeclared { component })?;
            let mut record = MetaRecord {
                component,
                incarnation,
                schema,
                state: MetaState::Installing,
            };
            storage.write_meta(&record)?;
            storage.create_namespace(component)?;
            record.state = MetaState::Active;
            storage.write_meta(&record)?;
            drop(storage);
            self.shared
                .declarations
                .lock()
                .map_err(|_| StateError::SynchronizationPoisoned)?
                .remove(&component);
            Ok(ComponentStoreHandle::new(
                Arc::clone(&self.shared),
                component,
                incarnation,
            ))
        })
    }

    /// Issues a scoped handle for an already-active namespace after boot.
    ///
    /// This issue-time check does not replace the consuming-act fence: every
    /// handle operation re-reads the durable incarnation before touching data.
    ///
    /// # Errors
    ///
    /// Refuses absent or transitional storage and propagates typed meta errors.
    pub fn component_handle(
        &self,
        component: ComponentId,
    ) -> Result<ComponentStoreHandle, StateError> {
        let storage = self.lock_storage()?;
        let record = require_active(&storage, component, None)?;
        Ok(ComponentStoreHandle::new(
            Arc::clone(&self.shared),
            component,
            record.incarnation,
        ))
    }

    /// Performs Archiving meta, snapshot pin + Work/Namespace ref removal, then
    /// Archived meta. Outstanding Work is committed for forensic reachability,
    /// recorded, and abandoned rather than blocking removal.
    ///
    /// The embedding host calls this only after frame-core stop + unregister.
    ///
    /// # Errors
    ///
    /// Refuses inactive/transitional or concurrent operations and propagates
    /// every durable branch/snapshot failure.
    pub fn archive_storage(&self, component: ComponentId) -> Result<Archive, StateError> {
        self.with_operation(component, Operation::Archive, || {
            let mut storage = self.lock_storage()?;
            let mut record = storage
                .meta_record(component)?
                .ok_or(StateError::NotActive { component })?;
            match record.state {
                MetaState::Active => {}
                MetaState::Installing => {
                    return Err(StateError::InstallInProgress { component });
                }
                MetaState::Archiving { .. } => {
                    return Err(StateError::ArchiveInProgress { component });
                }
                MetaState::Archived { .. } => return Err(StateError::NotActive { component }),
            }
            let generation = record.incarnation;
            storage.commit_namespace(component)?;
            let abandoned_work = storage.outstanding_work(component)?;
            record.state = MetaState::Archiving {
                generation,
                abandoned_work: abandoned_work.clone(),
            };
            storage.write_meta(&record)?;
            storage.pin_remove_archive(component, generation, &abandoned_work)?;
            record.state = MetaState::Archived {
                generation,
                abandoned_work: abandoned_work.clone(),
            };
            storage.finish_archive_meta(&record)?;
            let root = storage.archived_root(component, generation)?;
            Ok(Archive {
                generation,
                snapshot_name: archive_name(component, generation),
                root,
                abandoned_work,
            })
        })
    }

    /// Compares durable meta records with durable branch refs and completes every
    /// crash transition in the safe direction. It never consults a runtime
    /// registry and never silently deletes a namespace.
    ///
    /// # Errors
    ///
    /// Returns a typed, named inconsistency before repair when foreign branch
    /// interference is observed, or a typed substrate failure during repair.
    pub fn reconcile(&self) -> Result<ReconcileReport, StateError> {
        let mut storage = self.lock_storage()?;
        let records = storage.meta_records()?;
        for branch in storage.refs.list() {
            if branch.name == crate::engine::META_BRANCH {
                continue;
            }
            let owner_namespace = match branch.kind {
                BranchKind::Namespace => Some(branch.name.as_str()),
                BranchKind::Work => branch.resolved_namespace_lineage(),
            };
            let has_meta = owner_namespace.is_some_and(|namespace| {
                records
                    .iter()
                    .any(|record| namespace_name(record.component) == namespace)
            });
            if !has_meta {
                return Err(ReconcileInconsistency::BranchWithoutMeta {
                    branch: branch.name.clone(),
                    kind: branch.kind,
                }
                .into());
            }
        }
        let mut report = ReconcileReport::default();
        for mut record in records {
            let component = record.component;
            let exists = storage.namespace_exists(component);
            match &record.state {
                MetaState::Installing => {
                    if !exists {
                        storage.create_namespace(component)?;
                        report
                            .actions
                            .push(ReconcileAction::InstallBranchCreated { component });
                        record.state = MetaState::Active;
                        storage.write_meta(&record)?;
                    }
                }
                MetaState::Active if exists => {
                    report.healthy_active = report.healthy_active.saturating_add(1);
                }
                MetaState::Active => {
                    return Err(ReconcileInconsistency::ActiveWithoutBranch { component }.into());
                }
                MetaState::Archiving {
                    generation,
                    abandoned_work,
                } => {
                    let generation = *generation;
                    let abandoned_work = abandoned_work.clone();
                    if exists {
                        storage.pin_remove_archive(component, generation, &abandoned_work)?;
                        report.actions.push(ReconcileAction::OrphanedByCrash {
                            component,
                            generation,
                        });
                    } else {
                        let snapshot = archive_name(component, generation);
                        if storage.snapshots.get(&snapshot).is_none() {
                            return Err(ReconcileInconsistency::RemovedWithoutSnapshot {
                                component,
                                snapshot,
                            }
                            .into());
                        }
                        report.actions.push(ReconcileAction::ArchiveFinalized {
                            component,
                            generation,
                        });
                    }
                    record.state = MetaState::Archived {
                        generation,
                        abandoned_work,
                    };
                    storage.finish_archive_meta(&record)?;
                }
                MetaState::Archived { .. } if exists => {
                    return Err(ReconcileInconsistency::ArchivedWithBranch { component }.into());
                }
                MetaState::Archived { .. } => {}
            }
        }
        Ok(report)
    }

    /// Lists every pinned archive generation in monotonic order.
    ///
    /// # Errors
    ///
    /// Returns typed synchronization or archive-evidence decoding failures.
    pub fn list_archives(&self, component: ComponentId) -> Result<Vec<Archive>, StateError> {
        self.lock_storage()?.archives(component)
    }

    /// Reads one entity from a pinned archive generation through `ReadOnlyView`.
    ///
    /// # Errors
    ///
    /// Returns typed missing-archive, checkout, or synchronization failures.
    pub fn get_archived(
        &self,
        component: ComponentId,
        generation: u64,
        entity: EntityId,
    ) -> Result<Option<Vec<u8>>, StateError> {
        let storage = self.lock_storage()?;
        let root = storage.archived_root(component, generation)?;
        Ok(checkout(&storage.nodes, root).get(&crate::codec::entity_key(entity))?)
    }

    /// Enumerates entities from a pinned archive generation.
    ///
    /// # Errors
    ///
    /// Returns typed missing-archive, checkout, corruption, or synchronization failures.
    pub fn enumerate_archived(
        &self,
        component: ComponentId,
        generation: u64,
    ) -> Result<Vec<(EntityId, Vec<u8>)>, StateError> {
        let storage = self.lock_storage()?;
        entities_at(
            &storage.nodes,
            storage.archived_root(component, generation)?,
        )
    }

    /// Reads an entity at an explicitly recorded forensic root.
    ///
    /// This facade-only operation is intended for roots returned in
    /// [`Archive::abandoned_work`]; component handles cannot name arbitrary roots.
    ///
    /// # Errors
    ///
    /// Returns typed checkout or synchronization failures.
    pub fn get_at_root(&self, root: Hash, entity: EntityId) -> Result<Option<Vec<u8>>, StateError> {
        let storage = self.lock_storage()?;
        Ok(checkout(&storage.nodes, root).get(&crate::codec::entity_key(entity))?)
    }

    /// Returns the current committed namespace root.
    ///
    /// # Errors
    ///
    /// Refuses non-active state and propagates typed storage failures.
    pub fn current_root(&self, component: ComponentId) -> Result<Hash, StateError> {
        let mut storage = self.lock_storage()?;
        require_active(&storage, component, None)?;
        Ok(storage.namespace(component)?.current_root())
    }

    /// Compares one supplied root with the current root in constant time.
    ///
    /// Equal hashes prove unchanged. A moved hash does not prove a logical
    /// change, because content-addressed roots are history-independent.
    ///
    /// # Errors
    ///
    /// Refuses non-active state and propagates typed storage failures.
    pub fn has_changed(
        &self,
        component: ComponentId,
        since_root: Hash,
    ) -> Result<bool, StateError> {
        self.current_root(component).map(|root| root != since_root)
    }

    pub(crate) fn lock_storage(&self) -> Result<std::sync::MutexGuard<'_, Storage>, StateError> {
        self.shared
            .storage
            .lock()
            .map_err(|_| StateError::SynchronizationPoisoned)
    }

    fn with_operation<T>(
        &self,
        component: ComponentId,
        requested: Operation,
        operation: impl FnOnce() -> Result<T, StateError>,
    ) -> Result<T, StateError> {
        {
            let mut operations = self
                .shared
                .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 = operation();
        self.shared
            .operations
            .lock()
            .map_err(|_| StateError::SynchronizationPoisoned)?
            .remove(&component);
        result
    }
}

pub(crate) fn require_active(
    storage: &Storage,
    component: ComponentId,
    held: Option<u64>,
) -> Result<MetaRecord, StateError> {
    let record = storage
        .meta_record(component)?
        .ok_or(StateError::NotActive { component })?;
    if let Some(held) = held
        && held != record.incarnation
    {
        return Err(StateError::StaleHandle {
            component,
            held,
            current: record.incarnation,
        });
    }
    match record.state {
        MetaState::Active => Ok(record),
        MetaState::Installing => Err(StateError::InstallInProgress { component }),
        MetaState::Archiving { .. } => Err(StateError::ArchiveInProgress { component }),
        MetaState::Archived { .. } => Err(StateError::NotActive { component }),
    }
}

pub(crate) const fn operation_error(component: ComponentId, operation: Operation) -> StateError {
    match operation {
        Operation::Install => StateError::InstallInProgress { component },
        Operation::Archive => StateError::ArchiveInProgress { component },
        Operation::Work => StateError::WorkInProgress { component },
    }
}