alma 0.1.1

use super::resources::{
    workspace_observe_task_resource_error, workspace_observe_task_resource_unavailable,
};
use crate::plugin::{
    PluginHostImportError, PluginHostImportSession, PluginHostResourceKind, PluginIdentity,
    PluginResourceHandleRejectionReason, PluginWorkspaceObserveOutcome,
    PluginWorkspaceObserveOutcomeShape, PluginWorkspaceObserveTaskCancellationReport,
    PluginWorkspaceObserveTaskEnqueueError, PluginWorkspaceObserveTaskEnqueueReport,
    PluginWorkspaceObserveTaskHandle, PluginWorkspaceObserveTaskPoll,
    PluginWorkspaceObserveTaskQueue, PluginWorkspaceObserveTaskQueueLimit,
    PluginWorkspaceObserveTaskTake, SealedPluginWorkspaceObserveTaskBatch, WorkspacePath,
    abi::bindings::alma::editor::host as wit_host,
    host::PluginWorkspaceObserveTaskResourceAuthority,
};
use std::num::NonZeroUsize;
use wasmtime::component::{Resource, ResourceTable, ResourceTableError};

/// Store-local workspace observation task state.
#[derive(Debug)]
pub(super) struct WasmtimeWorkspaceObserveTasks {
    /// Host-owned task queue for filesystem-policy execution and retained outcomes.
    queue: PluginWorkspaceObserveTaskQueue,
    /// Persistent task resources visible to this component instance.
    resources: WasmtimeWorkspaceObserveTaskResources,
    /// Task resources returned during the active update.
    resources_started_in_update: WasmtimeStartedWorkspaceObserveTaskResources,
}

impl WasmtimeWorkspaceObserveTasks {
    /// Creates an empty task queue with a matching persistent-resource cap.
    pub(super) fn new(limit: PluginWorkspaceObserveTaskQueueLimit) -> Self {
        let queue = PluginWorkspaceObserveTaskQueue::from_validated_limit(limit);
        let resources = WasmtimeWorkspaceObserveTaskResources::new(queue.limit());
        Self {
            queue,
            resources,
            resources_started_in_update: WasmtimeStartedWorkspaceObserveTaskResources::new(),
        }
    }

    /// Returns the queue owner needed by the active host-import session.
    pub(super) const fn queue_mut(&mut self) -> &mut PluginWorkspaceObserveTaskQueue {
        &mut self.queue
    }

    /// Enqueues sealed task work produced by a successful update.
    pub(super) fn enqueue(
        &mut self,
        batch: SealedPluginWorkspaceObserveTaskBatch,
    ) -> Result<PluginWorkspaceObserveTaskEnqueueReport, PluginWorkspaceObserveTaskEnqueueError>
    {
        batch.enqueue(&mut self.queue)
    }

    /// Executes all pending task reads through filesystem policy.
    pub(super) fn execute_all(
        &mut self,
        filesystem: &crate::fs_utils::FilesystemConfig,
    ) -> Vec<crate::plugin::PluginWorkspaceObserveTaskCompletion> {
        self.queue.execute_all(filesystem)
    }

    /// Executes at most `max_tasks` pending task reads through filesystem policy.
    pub(super) fn execute_up_to(
        &mut self,
        filesystem: &crate::fs_utils::FilesystemConfig,
        max_tasks: NonZeroUsize,
    ) -> Vec<crate::plugin::PluginWorkspaceObserveTaskCompletion> {
        self.queue.execute_up_to(filesystem, max_tasks)
    }

    /// Executes the next pending task read through filesystem policy.
    pub(super) fn execute_next(
        &mut self,
        filesystem: &crate::fs_utils::FilesystemConfig,
    ) -> Option<crate::plugin::PluginWorkspaceObserveTaskCompletion> {
        self.queue.execute_next(filesystem)
    }

    /// Polls a task handle without exposing retained bytes.
    pub(super) fn poll(
        &self,
        handle: &PluginWorkspaceObserveTaskHandle,
    ) -> PluginWorkspaceObserveTaskPoll {
        self.queue.poll(handle)
    }

    /// Takes a completed task result.
    pub(super) fn take(
        &mut self,
        handle: &PluginWorkspaceObserveTaskHandle,
    ) -> PluginWorkspaceObserveTaskTake {
        self.queue.take(handle)
    }

    /// Adds a persistent task resource returned by the active update.
    pub(super) fn push_resource_started_in_update(
        &mut self,
        table: &mut ResourceTable,
        resource_authority: PluginWorkspaceObserveTaskResourceAuthority,
    ) -> Result<Resource<PluginWorkspaceObserveTaskHandle>, PluginHostImportError> {
        let resource_authority = resource_authority.into_parts();
        let resource = self.resources.push(
            table,
            resource_authority.handle,
            resource_authority.authority_path,
        )?;
        self.resources_started_in_update.record(&resource);
        Ok(resource)
    }

    /// Resolves a persistent task resource after revalidating workspace authority.
    pub(super) fn resolve_authorized_resource(
        &mut self,
        table: &ResourceTable,
        session: &PluginHostImportSession,
        resource: &Resource<PluginWorkspaceObserveTaskHandle>,
    ) -> Result<PluginWorkspaceObserveTaskHandle, PluginHostImportError> {
        self.resources.resolve_authorized(table, session, resource)
    }

    /// Drops one guest-owned task resource.
    pub(super) fn delete_resource(
        &mut self,
        table: &mut ResourceTable,
        resource: Resource<PluginWorkspaceObserveTaskHandle>,
    ) -> Result<WasmtimeWorkspaceObserveTaskResourceDrop, PluginHostImportError> {
        let rep = WasmtimeWorkspaceObserveTaskResourceRep::from_resource(&resource);
        let dropped = self.resources.delete(table, resource)?;
        self.forget_started_resource(rep);
        Ok(dropped)
    }

    /// Stops tracking a task resource as uncommitted after it is dropped.
    pub(super) fn forget_started_resource(&mut self, rep: WasmtimeWorkspaceObserveTaskResourceRep) {
        self.resources_started_in_update.forget(rep);
    }

    /// Clears per-update bookkeeping after successful guest return.
    pub(super) fn commit_started_in_update(&mut self) {
        self.resources_started_in_update.commit();
    }

    /// Tombstones task resources returned by an update that failed before commit.
    pub(super) fn invalidate_started_in_update(&mut self) -> usize {
        let reps = self.resources_started_in_update.take_for_invalidation();
        self.resources.invalidate_reps(&reps)
    }

    /// Cancels queue-owned task state and stale task resources for a revoked identity.
    pub(super) fn cancel_identity(
        &mut self,
        identity: &PluginIdentity,
    ) -> PluginWorkspaceObserveTaskCancellationReport {
        let report = self.queue.cancel_identity(identity);
        let _invalidated_resources = self.resources.cancel_identity(identity);
        self.resources_started_in_update
            .retain_not_belonging_to(&self.resources, identity);
        report
    }

    /// Cancels all queue-owned task state and tombstones all persistent task resources.
    pub(super) fn cancel_all(&mut self) -> Vec<PluginWorkspaceObserveTaskCancellationReport> {
        let reports = self.queue.cancel_all();
        let _invalidated_resources = self.resources.cancel_all();
        self.resources_started_in_update.clear();
        reports
    }

    /// Returns the number of reads waiting for filesystem policy.
    #[cfg(test)]
    pub(super) fn pending_len(&self) -> usize {
        self.queue.pending_len()
    }

    /// Returns pending queue slots reserved by unsealed update batches.
    #[cfg(test)]
    pub(super) fn reserved_pending_len(&self) -> usize {
        self.queue.reserved_pending_len()
    }

    /// Returns the number of completed outcomes retained for polling.
    #[cfg(test)]
    pub(super) fn retained_len(&self) -> usize {
        self.queue.retained_len()
    }

    /// Returns the number of active persistent task resources.
    #[cfg(test)]
    pub(super) fn live_resource_len(&self) -> usize {
        self.resources.live_len()
    }

    /// Returns tracked resource table reps for aliasing regressions.
    #[cfg(test)]
    pub(super) fn tracked_resource_reps(&self) -> Vec<WasmtimeWorkspaceObserveTaskResourceRep> {
        self.resources
            .resources
            .iter()
            .map(|resource| resource.rep)
            .collect()
    }
}

/// Task resource reps returned by the active update and not yet committed.
#[derive(Debug)]
pub(super) struct WasmtimeStartedWorkspaceObserveTaskResources {
    /// Persistent-resource reps whose pending task requests are still discardable.
    reps: Vec<WasmtimeWorkspaceObserveTaskResourceRep>,
}

impl WasmtimeStartedWorkspaceObserveTaskResources {
    /// Starts with no active-update resources.
    pub(super) const fn new() -> Self {
        Self { reps: Vec::new() }
    }

    /// Records a task resource returned to the guest during the active update.
    pub(super) fn record(&mut self, resource: &Resource<PluginWorkspaceObserveTaskHandle>) {
        self.reps
            .push(WasmtimeWorkspaceObserveTaskResourceRep::from_resource(
                resource,
            ));
    }

    /// Stops treating a dropped task resource as uncommitted update work.
    pub(super) fn forget(&mut self, rep: WasmtimeWorkspaceObserveTaskResourceRep) {
        self.reps.retain(|started| *started != rep);
    }

    /// Successful guest return promotes started resources to persistent resources.
    pub(super) fn commit(&mut self) {
        self.reps.clear();
    }

    /// Failed or abandoned guest return tombstones every still-uncommitted resource.
    pub(super) fn take_for_invalidation(&mut self) -> Vec<WasmtimeWorkspaceObserveTaskResourceRep> {
        std::mem::take(&mut self.reps)
    }

    /// Keeps only started resources whose identity did not just lose runtime-owned task state.
    pub(super) fn retain_not_belonging_to(
        &mut self,
        resources: &WasmtimeWorkspaceObserveTaskResources,
        identity: &PluginIdentity,
    ) {
        self.reps
            .retain(|rep| !resources.resource_belongs_to_identity(*rep, identity));
    }

    /// Removes all active-update resource bookkeeping.
    pub(super) fn clear(&mut self) {
        self.reps.clear();
    }
}

/// Store-local occupied-resource cap for persistent workspace observation task resources.
#[derive(Debug)]
pub(super) struct WasmtimeWorkspaceObserveTaskResources {
    /// Guest-owned resources that may remain live or tombstoned across updates.
    pub(super) resources: Vec<WasmtimeWorkspaceObserveTaskResource>,
    /// Cap derived from queue-owned pending and retention bounds.
    limit: WasmtimeWorkspaceObserveTaskResourceLimit,
}

impl WasmtimeWorkspaceObserveTaskResources {
    /// Creates an empty resource tracker.
    pub(super) const fn new(limit: PluginWorkspaceObserveTaskQueueLimit) -> Self {
        Self {
            resources: Vec::new(),
            limit: WasmtimeWorkspaceObserveTaskResourceLimit::from_queue_limit(limit),
        }
    }

    /// Adds a task resource only while the occupied-resource cap has room.
    pub(super) fn push(
        &mut self,
        table: &mut ResourceTable,
        handle: PluginWorkspaceObserveTaskHandle,
        authority_path: WorkspacePath,
    ) -> Result<Resource<PluginWorkspaceObserveTaskHandle>, PluginHostImportError> {
        if self.resources.len() >= self.limit.get() {
            return Err(PluginHostImportError::ResourceHandle {
                kind: PluginHostResourceKind::WorkspaceObserveTask,
                reason: PluginResourceHandleRejectionReason::Unavailable,
            });
        }
        let resource = table
            .push(handle.clone())
            .map_err(|source| workspace_observe_task_resource_error(&source))?;
        let rep = WasmtimeWorkspaceObserveTaskResourceRep::from_resource(&resource);
        if let Some(index) = self.resource_index(rep) {
            let _invalidated = self.resources[index].invalidate();
            let _handle = table
                .delete(resource)
                .map_err(|source| workspace_observe_task_resource_error(&source))?;
            return Err(workspace_observe_task_resource_unavailable());
        }
        self.resources
            .push(WasmtimeWorkspaceObserveTaskResource::new(
                rep,
                handle,
                authority_path,
            ));
        Ok(resource)
    }

    /// Resolves a task resource after revalidating its originating workspace authority.
    pub(super) fn resolve_authorized(
        &mut self,
        table: &ResourceTable,
        session: &PluginHostImportSession,
        resource: &Resource<PluginWorkspaceObserveTaskHandle>,
    ) -> Result<PluginWorkspaceObserveTaskHandle, PluginHostImportError> {
        let rep = WasmtimeWorkspaceObserveTaskResourceRep::from_resource(resource);
        let Some(index) = self.resource_index(rep) else {
            return Err(workspace_observe_task_resource_error(
                &ResourceTableError::NotPresent,
            ));
        };
        if !self.resources[index].is_active() {
            return Err(workspace_observe_task_resource_error(
                &ResourceTableError::NotPresent,
            ));
        }
        if let Err(error) = session.authorize_workspace_observe_task_resource(
            self.resources[index].authority_path.as_ref(),
        ) {
            let _invalidated = self.resources[index].invalidate();
            return Err(error);
        }

        let handle = match table.get(resource) {
            Ok(handle) => handle.clone(),
            Err(source) => {
                let _invalidated = self.resources[index].invalidate();
                return Err(workspace_observe_task_resource_error(&source));
            }
        };
        if handle != self.resources[index].handle {
            let _invalidated = self.resources[index].invalidate();
            return Err(workspace_observe_task_resource_unavailable());
        }
        Ok(handle)
    }

    /// Drops one guest-owned task resource.
    pub(super) fn delete(
        &mut self,
        table: &mut ResourceTable,
        resource: Resource<PluginWorkspaceObserveTaskHandle>,
    ) -> Result<WasmtimeWorkspaceObserveTaskResourceDrop, PluginHostImportError> {
        let rep = WasmtimeWorkspaceObserveTaskResourceRep::from_resource(&resource);
        match table.delete(resource) {
            Ok(handle) => {
                let Some(tracked) = self.remove_rep(rep) else {
                    return Err(workspace_observe_task_resource_unavailable());
                };
                if handle != tracked.handle {
                    return Err(workspace_observe_task_resource_unavailable());
                }
                Ok(WasmtimeWorkspaceObserveTaskResourceDrop::from_tracked(
                    tracked,
                ))
            }
            Err(ResourceTableError::NotPresent) => Ok(self.remove_rep(rep).map_or(
                WasmtimeWorkspaceObserveTaskResourceDrop::Inactive,
                WasmtimeWorkspaceObserveTaskResourceDrop::from_tracked,
            )),
            Err(source) => {
                if let Some(index) = self.resource_index(rep) {
                    let _invalidated = self.resources[index].invalidate();
                }
                Err(workspace_observe_task_resource_error(&source))
            }
        }
    }

    /// Invalidates all live task resources for a revoked identity.
    pub(super) fn cancel_identity(&mut self, identity: &PluginIdentity) -> usize {
        let mut invalidated = 0;
        for resource in &mut self.resources {
            if resource.handle.identity_proof() == identity && resource.invalidate() {
                invalidated += 1;
            }
        }
        invalidated
    }

    /// Invalidates every live task resource owned by this store.
    pub(super) fn cancel_all(&mut self) -> usize {
        let mut invalidated = 0;
        for resource in &mut self.resources {
            if resource.invalidate() {
                invalidated += 1;
            }
        }
        invalidated
    }

    /// Invalidates task resources returned by an update that failed before commit.
    pub(super) fn invalidate_reps(
        &mut self,
        reps: &[WasmtimeWorkspaceObserveTaskResourceRep],
    ) -> usize {
        let mut invalidated = 0;
        for rep in reps {
            let Some(index) = self.resource_index(*rep) else {
                continue;
            };
            if self.resources[index].invalidate() {
                invalidated += 1;
            }
        }
        invalidated
    }

    /// Returns the number of live task resources tracked by this store.
    #[must_use]
    #[cfg(test)]
    pub(super) fn live_len(&self) -> usize {
        self.resources
            .iter()
            .filter(|resource| resource.is_active())
            .count()
    }

    /// Removes the tracked resource matching a table rep.
    pub(super) fn remove_rep(
        &mut self,
        rep: WasmtimeWorkspaceObserveTaskResourceRep,
    ) -> Option<WasmtimeWorkspaceObserveTaskResource> {
        let index = self.resource_index(rep)?;
        Some(self.resources.remove(index))
    }

    /// Returns the tracked index for a table resource rep.
    pub(super) fn resource_index(
        &self,
        rep: WasmtimeWorkspaceObserveTaskResourceRep,
    ) -> Option<usize> {
        self.resources
            .iter()
            .position(|resource| resource.rep == rep)
    }

    /// Returns whether a tracked resource rep belongs to an identity.
    pub(super) fn resource_belongs_to_identity(
        &self,
        rep: WasmtimeWorkspaceObserveTaskResourceRep,
        identity: &PluginIdentity,
    ) -> bool {
        self.resource_index(rep)
            .is_some_and(|index| self.resources[index].handle.identity_proof() == identity)
    }
}

/// Effect of dropping a workspace observation task resource.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(super) enum WasmtimeWorkspaceObserveTaskResourceDrop {
    /// A live task resource was dropped and any matching uncommitted request should be removed.
    Active(PluginWorkspaceObserveTaskHandle),
    /// A missing or already-invalidated task resource was cleaned up.
    Inactive,
}

impl WasmtimeWorkspaceObserveTaskResourceDrop {
    /// Classifies a tracked resource removed from the table.
    pub(super) fn from_tracked(resource: WasmtimeWorkspaceObserveTaskResource) -> Self {
        if resource.is_active() {
            Self::Active(resource.handle)
        } else {
            Self::Inactive
        }
    }
}

/// Non-zero occupied-resource cap for guest-owned workspace observation task resources.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) struct WasmtimeWorkspaceObserveTaskResourceLimit(NonZeroUsize);

impl WasmtimeWorkspaceObserveTaskResourceLimit {
    /// Derives the resource cap from queue-owned pending and retained task bounds.
    pub(super) const fn from_queue_limit(limit: PluginWorkspaceObserveTaskQueueLimit) -> Self {
        Self(limit.max_resource_slots_limit())
    }

    /// Returns the occupied resource slot cap.
    pub(super) const fn get(self) -> usize {
        self.0.get()
    }
}

/// Typed Wasmtime table rep for guest-owned workspace observation task resources.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) struct WasmtimeWorkspaceObserveTaskResourceRep(u32);

impl WasmtimeWorkspaceObserveTaskResourceRep {
    /// Captures the table rep for a workspace observation task resource.
    pub(super) fn from_resource<T>(resource: &Resource<T>) -> Self {
        Self(resource.rep())
    }

    /// Raw table rep for fixture assertions.
    #[cfg(test)]
    pub(super) const fn raw(self) -> u32 {
        self.0
    }

    /// Rebuilds a borrowed task resource for host-import fixture calls.
    #[cfg(test)]
    pub(super) fn borrowed_task(self) -> Resource<PluginWorkspaceObserveTaskHandle> {
        Resource::new_borrow(self.0)
    }

    /// Rebuilds an owned task resource for canonical-drop fixture calls.
    #[cfg(test)]
    pub(super) fn owned_task(self) -> Resource<PluginWorkspaceObserveTaskHandle> {
        Resource::new_own(self.0)
    }
}

/// Store-tracked workspace observation task resource.
#[derive(Clone, Debug, Eq, PartialEq)]
pub(super) struct WasmtimeWorkspaceObserveTaskResource {
    /// Wasmtime resource-table index.
    pub(super) rep: WasmtimeWorkspaceObserveTaskResourceRep,
    /// Identity-scoped task handle stored in the table.
    pub(super) handle: PluginWorkspaceObserveTaskHandle,
    /// Workspace path whose observation grant authorized this resource.
    pub(super) authority_path: WorkspacePath,
    /// Whether the guest-owned table slot still grants task access.
    pub(super) state: WasmtimeWorkspaceObserveTaskResourceState,
}

impl WasmtimeWorkspaceObserveTaskResource {
    /// Captures a live task resource table entry.
    pub(super) const fn new(
        rep: WasmtimeWorkspaceObserveTaskResourceRep,
        handle: PluginWorkspaceObserveTaskHandle,
        authority_path: WorkspacePath,
    ) -> Self {
        Self {
            rep,
            handle,
            authority_path,
            state: WasmtimeWorkspaceObserveTaskResourceState::Active,
        }
    }

    /// Returns whether the resource still authorizes queue access.
    pub(super) const fn is_active(&self) -> bool {
        matches!(
            self.state,
            WasmtimeWorkspaceObserveTaskResourceState::Active
        )
    }

    /// Tombstones a guest-owned resource without freeing its table rep.
    pub(super) const fn invalidate(&mut self) -> bool {
        if !self.is_active() {
            return false;
        }
        self.state = WasmtimeWorkspaceObserveTaskResourceState::Invalidated;
        true
    }
}

/// Resource-table slot state for a guest-owned workspace observation task.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum WasmtimeWorkspaceObserveTaskResourceState {
    /// Poll/take may revalidate authority and reach task state.
    Active,
    /// The table rep remains occupied until guest drop, but grants no access.
    Invalidated,
}

/// Converts a redacted poll result into the guest ABI.
pub(super) fn workspace_observe_poll_to_guest(
    poll: PluginWorkspaceObserveTaskPoll,
) -> wit_host::WorkspaceObserveTaskPoll {
    match poll {
        PluginWorkspaceObserveTaskPoll::Pending => wit_host::WorkspaceObserveTaskPoll::Pending,
        PluginWorkspaceObserveTaskPoll::Completed { outcome } => {
            wit_host::WorkspaceObserveTaskPoll::Completed(workspace_observe_shape_to_guest(outcome))
        }
        PluginWorkspaceObserveTaskPoll::Unknown => wit_host::WorkspaceObserveTaskPoll::Unknown,
    }
}

/// Converts a consuming task result into the guest ABI.
pub(super) fn workspace_observe_take_to_guest(
    take: PluginWorkspaceObserveTaskTake,
) -> wit_host::WorkspaceObserveTaskTake {
    match take {
        PluginWorkspaceObserveTaskTake::Pending => wit_host::WorkspaceObserveTaskTake::Pending,
        PluginWorkspaceObserveTaskTake::Completed(PluginWorkspaceObserveOutcome::Bytes(bytes)) => {
            wit_host::WorkspaceObserveTaskTake::Bytes(bytes.into_vec())
        }
        PluginWorkspaceObserveTaskTake::Completed(PluginWorkspaceObserveOutcome::Rejected) => {
            wit_host::WorkspaceObserveTaskTake::Rejected
        }
        PluginWorkspaceObserveTaskTake::Unknown => wit_host::WorkspaceObserveTaskTake::Unknown,
    }
}

/// Converts a redacted workspace observation outcome shape into the guest ABI.
fn workspace_observe_shape_to_guest(
    shape: PluginWorkspaceObserveOutcomeShape,
) -> wit_host::WorkspaceObserveOutcomeShape {
    match shape {
        PluginWorkspaceObserveOutcomeShape::Bytes { byte_len } => {
            wit_host::WorkspaceObserveOutcomeShape::Bytes(
                u64::try_from(byte_len).unwrap_or(u64::MAX),
            )
        }
        PluginWorkspaceObserveOutcomeShape::Rejected => {
            wit_host::WorkspaceObserveOutcomeShape::Rejected
        }
    }
}