alma 0.1.1

//! FIFO worker queue for sealed workspace I/O.

use super::{
    PluginWorkspaceIoDiscardReport, PluginWorkspaceIoReport, SealedPluginWorkspaceIoBatch,
};
use crate::{
    fs_utils::FilesystemConfig,
    plugin::{PluginIdentity, PluginOperationalEvent, PluginOperationalQueue},
};
use std::{
    collections::VecDeque,
    fmt::{Debug, Display, Formatter},
    num::NonZeroUsize,
};

/// Worker queue limit fields with stable diagnostic spellings.
#[derive(Clone, Copy, Eq, PartialEq)]
pub enum PluginWorkspaceIoWorkerQueueLimitField {
    /// Maximum sealed workspace I/O batches waiting for filesystem policy.
    MaxBatches,
}

impl PluginWorkspaceIoWorkerQueueLimitField {
    /// Stable field name used in diagnostics.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::MaxBatches => "max_batches",
        }
    }
}

impl Display for PluginWorkspaceIoWorkerQueueLimitField {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.as_str())
    }
}

impl Debug for PluginWorkspaceIoWorkerQueueLimitField {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// Bounded worker queue cap.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PluginWorkspaceIoWorkerQueueLimit {
    /// Maximum queued sealed batches.
    max_batches: NonZeroUsize,
}

impl PluginWorkspaceIoWorkerQueueLimit {
    /// Creates a worker queue limit.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoWorkerQueueError::ZeroLimit`] when `max_batches` is zero.
    pub const fn try_new(max_batches: usize) -> Result<Self, PluginWorkspaceIoWorkerQueueError> {
        match NonZeroUsize::new(max_batches) {
            Some(max_batches) => Ok(Self { max_batches }),
            None => Err(PluginWorkspaceIoWorkerQueueError::ZeroLimit {
                field: PluginWorkspaceIoWorkerQueueLimitField::MaxBatches,
            }),
        }
    }

    /// Returns the queued sealed-batch cap.
    #[must_use]
    pub const fn max_batches(self) -> usize {
        self.max_batches.get()
    }

    /// Returns the queued sealed-batch cap as a non-zero proof for diagnostics.
    #[must_use]
    const fn max_batches_limit(self) -> NonZeroUsize {
        self.max_batches
    }
}

impl Default for PluginWorkspaceIoWorkerQueueLimit {
    fn default() -> Self {
        Self::try_new(1024).expect("default workspace queue limit should be non-zero")
    }
}

/// FIFO queue for sealed workspace I/O.
pub struct PluginWorkspaceIoWorkerQueue {
    /// Worker queue cap.
    limit: PluginWorkspaceIoWorkerQueueLimit,
    /// Sealed batches awaiting filesystem execution.
    batches: VecDeque<SealedPluginWorkspaceIoBatch>,
}

impl PluginWorkspaceIoWorkerQueue {
    /// Creates an empty worker queue.
    #[must_use]
    pub const fn with_limit(limit: PluginWorkspaceIoWorkerQueueLimit) -> Self {
        Self::from_validated_limit(limit)
    }

    /// Creates an empty worker queue from an already-validated limit proof.
    #[must_use]
    pub(in crate::plugin) const fn from_validated_limit(
        limit: PluginWorkspaceIoWorkerQueueLimit,
    ) -> Self {
        Self {
            limit,
            batches: VecDeque::new(),
        }
    }

    /// Queues sealed workspace I/O.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoWorkerEnqueueError`] when the queue is full.
    pub fn push(
        &mut self,
        batch: SealedPluginWorkspaceIoBatch,
    ) -> Result<(), PluginWorkspaceIoWorkerEnqueueError> {
        self.admit(batch)?.publish();
        Ok(())
    }

    /// Validates queue admission and retains the admitted batch with the target queue.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoWorkerEnqueueError`] when the queue is full.
    pub(crate) fn admit(
        &mut self,
        batch: SealedPluginWorkspaceIoBatch,
    ) -> Result<PluginWorkspaceIoWorkerQueueAdmission<'_>, PluginWorkspaceIoWorkerEnqueueError>
    {
        if batch.is_empty() {
            return Ok(PluginWorkspaceIoWorkerQueueAdmission::new(
                batch,
                &mut self.batches,
            ));
        }
        if self.batches.len() >= self.limit.max_batches() {
            let source = PluginWorkspaceIoWorkerQueueError::TooManyBatches {
                identity: batch.identity_proof().clone(),
                limit: self.limit.max_batches_limit(),
            };
            return Err(PluginWorkspaceIoWorkerEnqueueError::new(source, batch));
        }
        Ok(PluginWorkspaceIoWorkerQueueAdmission::new(
            batch,
            &mut self.batches,
        ))
    }

    /// Cancels queued work for a revoked instance.
    #[must_use]
    pub fn cancel_identity(
        &mut self,
        identity: &PluginIdentity,
    ) -> PluginWorkspaceIoCancellationReport {
        let mut retained = VecDeque::with_capacity(self.batches.len());
        let mut report = PluginWorkspaceIoCancellationReport::new(identity.clone());
        while let Some(batch) = self.batches.pop_front() {
            if batch.identity_proof() == identity {
                report.record_batch(batch.request_count());
            } else {
                retained.push_back(batch);
            }
        }
        self.batches = retained;
        report
    }

    /// Cancels all queued work before filesystem execution.
    ///
    /// Reports are emitted in first-seen queue order and grouped by plugin identity. Retryable
    /// batches held outside the queue remain owned by their caller.
    #[must_use]
    pub fn cancel_all(&mut self) -> Vec<PluginWorkspaceIoCancellationReport> {
        let mut reports = PluginWorkspaceIoCancellationReports::new();
        while let Some(batch) = self.batches.pop_front() {
            reports.record_batch(batch.identity_proof(), batch.request_count());
        }
        reports.into_vec()
    }

    /// Executes the next queued batch.
    pub fn execute_next(
        &mut self,
        filesystem: &FilesystemConfig,
    ) -> Option<PluginWorkspaceIoReport> {
        self.batches
            .pop_front()
            .map(|batch| batch.execute_synchronous(filesystem))
    }

    /// Executes queued batches in FIFO order.
    pub fn execute_all(&mut self, filesystem: &FilesystemConfig) -> Vec<PluginWorkspaceIoReport> {
        let mut reports = Vec::with_capacity(self.batches.len());
        while let Some(report) = self.execute_next(filesystem) {
            reports.push(report);
        }
        reports
    }

    /// Returns the queue cap.
    #[must_use]
    pub const fn limit(&self) -> PluginWorkspaceIoWorkerQueueLimit {
        self.limit
    }

    /// Returns the queued batch count.
    #[must_use]
    pub fn len(&self) -> usize {
        self.batches.len()
    }

    /// Returns whether the queue is empty.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.batches.is_empty()
    }
}

impl Default for PluginWorkspaceIoWorkerQueue {
    fn default() -> Self {
        Self::from_validated_limit(PluginWorkspaceIoWorkerQueueLimit::default())
    }
}

impl Debug for PluginWorkspaceIoWorkerQueue {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginWorkspaceIoWorkerQueue")
            .field("limit", &self.limit)
            .field("batch_count", &self.batches.len())
            .field(
                "batch_shapes",
                &self
                    .batches
                    .iter()
                    .map(|batch| PluginWorkspaceIoWorkerBatchShape {
                        identity: batch.identity_proof().clone(),
                        request_count: batch.request_count(),
                    })
                    .collect::<Vec<_>>(),
            )
            .finish()
    }
}

/// Sealed workspace I/O batch paired with the worker queue that admitted it.
#[must_use]
pub struct PluginWorkspaceIoWorkerQueueAdmission<'queue> {
    /// Admitted workspace I/O batch.
    batch: SealedPluginWorkspaceIoBatch,
    /// Queue storage that produced the admission proof.
    batches: &'queue mut VecDeque<SealedPluginWorkspaceIoBatch>,
}

impl<'queue> PluginWorkspaceIoWorkerQueueAdmission<'queue> {
    /// Pairs an admitted batch with its target queue storage.
    const fn new(
        batch: SealedPluginWorkspaceIoBatch,
        batches: &'queue mut VecDeque<SealedPluginWorkspaceIoBatch>,
    ) -> Self {
        Self { batch, batches }
    }

    /// Publishes the admitted batch.
    pub(crate) fn publish(self) {
        let Self { batch, batches } = self;
        if !batch.is_empty() {
            batches.push_back(batch);
        }
    }
}

/// Redacted queued-batch shape.
#[derive(Clone, Debug, Eq, PartialEq)]
struct PluginWorkspaceIoWorkerBatchShape {
    /// Validated identity retained only for redacted queue diagnostics.
    identity: PluginIdentity,
    /// Number of queued operations, not their paths or payloads.
    request_count: usize,
}

/// Deterministic all-identity cancellation report accumulator.
struct PluginWorkspaceIoCancellationReports {
    /// Reports in first-seen queue order.
    reports: Vec<PluginWorkspaceIoCancellationReport>,
}

impl PluginWorkspaceIoCancellationReports {
    /// Starts an empty cancellation report set.
    const fn new() -> Self {
        Self {
            reports: Vec::new(),
        }
    }

    /// Records one canceled sealed batch for an identity.
    fn record_batch(&mut self, identity: &PluginIdentity, request_count: usize) {
        self.report_for(identity).record_batch(request_count);
    }

    /// Returns accumulated reports in deterministic order.
    fn into_vec(self) -> Vec<PluginWorkspaceIoCancellationReport> {
        self.reports
    }

    /// Returns the report for an identity, creating it at the first observed queue position.
    fn report_for(
        &mut self,
        identity: &PluginIdentity,
    ) -> &mut PluginWorkspaceIoCancellationReport {
        if let Some(index) = self
            .reports
            .iter()
            .position(|report| report.identity_proof() == identity)
        {
            return &mut self.reports[index];
        }

        let index = self.reports.len();
        self.reports
            .push(PluginWorkspaceIoCancellationReport::new(identity.clone()));
        &mut self.reports[index]
    }
}

/// Revocation cancellation report.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PluginWorkspaceIoCancellationReport {
    /// Revoked plugin identity.
    identity: PluginIdentity,
    /// Batches removed from the queue.
    canceled_batches: usize,
    /// Requests removed from the queue.
    canceled_requests: usize,
}

/// Failed admission of a sealed workspace I/O batch.
#[derive(Debug, Eq, PartialEq, thiserror::Error)]
#[error("{source}")]
pub struct PluginWorkspaceIoWorkerEnqueueError {
    /// Admission rejection.
    source: PluginWorkspaceIoWorkerQueueError,
    /// Still-sealed workspace I/O batch retained for retry or explicit drop.
    batch: SealedPluginWorkspaceIoBatch,
}

/// Retryable workspace I/O worker enqueue failure split by queue boundary.
pub struct PluginWorkspaceIoWorkerEnqueueErrorParts {
    /// Closed queue rejection.
    pub queue_error: PluginWorkspaceIoWorkerQueueError,
    /// Still-sealed workspace I/O batch retained for retry.
    pub batch: SealedPluginWorkspaceIoBatch,
}

impl PluginWorkspaceIoWorkerEnqueueError {
    /// Builds a retry-safe enqueue error.
    const fn new(
        source: PluginWorkspaceIoWorkerQueueError,
        batch: SealedPluginWorkspaceIoBatch,
    ) -> Self {
        Self { source, batch }
    }

    /// Returns the closed queue rejection.
    #[must_use]
    pub const fn queue_error(&self) -> &PluginWorkspaceIoWorkerQueueError {
        &self.source
    }

    /// Returns the sealed batch retained for retry.
    #[must_use]
    pub const fn batch(&self) -> &SealedPluginWorkspaceIoBatch {
        &self.batch
    }

    /// Returns named queue rejection and retryable sealed-batch parts.
    #[must_use]
    pub fn into_parts(self) -> PluginWorkspaceIoWorkerEnqueueErrorParts {
        PluginWorkspaceIoWorkerEnqueueErrorParts {
            queue_error: self.source,
            batch: self.batch,
        }
    }

    /// Consumes the error and returns the retryable sealed batch.
    #[must_use]
    pub fn into_batch(self) -> SealedPluginWorkspaceIoBatch {
        self.batch
    }

    /// Discards the retained sealed batch instead of retrying admission.
    #[must_use]
    pub fn discard_batch(self) -> PluginWorkspaceIoDiscardReport {
        self.batch.discard()
    }
}

impl PluginWorkspaceIoCancellationReport {
    /// Starts a cancellation report for one plugin identity.
    const fn new(identity: PluginIdentity) -> Self {
        Self {
            identity,
            canceled_batches: 0,
            canceled_requests: 0,
        }
    }

    /// Adds one canceled sealed batch.
    const fn record_batch(&mut self, request_count: usize) {
        self.canceled_batches += 1;
        self.canceled_requests += request_count;
    }

    /// Returns the canceled plugin identity for display and serialization.
    #[must_use]
    pub fn identity(&self) -> &str {
        self.identity_proof().as_str()
    }

    /// Returns the validated identity revoked by this cancellation report.
    #[must_use]
    pub const fn identity_proof(&self) -> &PluginIdentity {
        &self.identity
    }

    /// Returns the canceled batch count.
    #[must_use]
    pub const fn canceled_batches(&self) -> usize {
        self.canceled_batches
    }

    /// Returns the canceled request count.
    #[must_use]
    pub const fn canceled_requests(&self) -> usize {
        self.canceled_requests
    }
}

/// Workspace I/O worker queue rejection class.
#[derive(Clone, Copy, Eq, PartialEq)]
pub enum PluginWorkspaceIoWorkerQueueErrorKind {
    /// A queue limit was zero.
    ZeroLimit,
    /// Sealed-batch capacity was exhausted.
    TooManyBatches,
}

impl PluginWorkspaceIoWorkerQueueErrorKind {
    /// Stable diagnostic spelling.
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::ZeroLimit => "zero-limit",
            Self::TooManyBatches => "too-many-batches",
        }
    }
}

impl Display for PluginWorkspaceIoWorkerQueueErrorKind {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.as_str())
    }
}

impl Debug for PluginWorkspaceIoWorkerQueueErrorKind {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(self.as_str())
    }
}

/// Worker queue rejection.
#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum PluginWorkspaceIoWorkerQueueError {
    /// Zero queue cap.
    ZeroLimit {
        /// Limit field name.
        field: PluginWorkspaceIoWorkerQueueLimitField,
    },
    /// Queue at cap.
    TooManyBatches {
        /// Rejected identity.
        identity: PluginIdentity,
        /// Queue cap.
        limit: NonZeroUsize,
    },
}

impl Display for PluginWorkspaceIoWorkerQueueError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ZeroLimit { .. } => {
                formatter.write_str("plugin workspace I/O queue limit must be non-zero")
            }
            Self::TooManyBatches { identity, limit } => {
                let identity = identity.as_str();
                let limit = limit.get();
                write!(
                    formatter,
                    "plugin {identity:?} exceeded workspace I/O queue limit of {limit}"
                )
            }
        }
    }
}

impl PluginWorkspaceIoWorkerQueueError {
    /// Closed rejection class for caller decisions and diagnostics.
    #[must_use]
    pub const fn kind(&self) -> PluginWorkspaceIoWorkerQueueErrorKind {
        match self {
            Self::ZeroLimit { .. } => PluginWorkspaceIoWorkerQueueErrorKind::ZeroLimit,
            Self::TooManyBatches { .. } => PluginWorkspaceIoWorkerQueueErrorKind::TooManyBatches,
        }
    }

    /// Redacted event for queue failure.
    #[must_use]
    pub fn operational_event(&self) -> Option<PluginOperationalEvent> {
        match self {
            Self::TooManyBatches { identity, limit } => {
                Some(PluginOperationalEvent::queue_saturated_for(
                    identity,
                    PluginOperationalQueue::WorkspaceIo,
                    *limit,
                ))
            }
            Self::ZeroLimit { .. } => None,
        }
    }
}