alma 0.1.1

//! Deferred workspace I/O batches.

use super::{
    AuthorizedWorkspaceReadAccess, AuthorizedWorkspaceWriteAccess, PluginWorkspaceIoBudget,
    PluginWorkspaceIoCommitError, PluginWorkspaceIoCompletion,
    PluginWorkspaceIoIdentityMismatchSource, PluginWorkspaceIoReport,
    PluginWorkspaceIoWorkerEnqueueError, PluginWorkspaceIoWorkerQueue,
    PluginWorkspaceObserveOutcome, PluginWorkspaceReadByteLimit, PluginWorkspaceReadSuccess,
    PluginWorkspaceWriteByteLimit, PluginWorkspaceWriteSuccess,
};
use super::{
    access::{PluginWorkspaceReadToken, PluginWorkspaceWriteToken},
    budget::{PluginWorkspaceIoRequestLedger, PluginWorkspaceNonDeferredReadPermit},
    report::{PluginWorkspaceReadCompletion, PluginWorkspaceWriteCompletion},
};
use crate::{
    fs_utils::{EscapedDisplayText, FilesystemConfig},
    plugin::{PluginIdentity, WorkspacePath},
};
use std::fmt::{Debug, Formatter};

/// V1 workspace I/O execution policy.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PluginWorkspaceIoExecutionModel {
    /// Host imports queue bounded workspace requests during a guest update, then execute them
    /// synchronously through `fs_utils` only after the guest returns successfully.
    DeferredSynchronous,
}

/// Workspace I/O accepted during one guest update but not yet committed to filesystem policy.
#[derive(Eq, PartialEq)]
pub struct PendingPluginWorkspaceIoBatch {
    /// Shared request budget for deferred and task-style workspace work.
    request_ledger: PluginWorkspaceIoRequestLedger,
    /// Accepted workspace operations.
    requests: Vec<PendingPluginWorkspaceIo>,
}

impl PendingPluginWorkspaceIoBatch {
    /// Creates an empty pending workspace I/O batch.
    ///
    /// The only v1 execution model is deferred synchronous I/O: requests are accepted while the
    /// guest update is running, but no filesystem operation is performed until the batch is sealed
    /// after successful guest return.
    #[must_use]
    pub const fn new(identity: PluginIdentity, budget: PluginWorkspaceIoBudget) -> Self {
        Self::from_validated_budget(identity, budget)
    }

    /// Creates an empty pending workspace I/O batch from an already-validated budget proof.
    #[must_use]
    pub(in crate::plugin) const fn from_validated_budget(
        identity: PluginIdentity,
        budget: PluginWorkspaceIoBudget,
    ) -> Self {
        Self {
            request_ledger: PluginWorkspaceIoRequestLedger::new(identity, budget),
            requests: Vec::new(),
        }
    }

    /// Returns the v1 execution model for this batch.
    #[must_use]
    pub const fn execution_model(&self) -> PluginWorkspaceIoExecutionModel {
        PluginWorkspaceIoExecutionModel::DeferredSynchronous
    }

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

    /// Returns the validated identity retained before the successful-return boundary.
    #[must_use]
    pub const fn identity_proof(&self) -> &PluginIdentity {
        self.request_ledger.identity_proof()
    }

    /// Queues a workspace read that has already passed plugin capability authorization.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoCommitError`] when the batch is full.
    pub fn push_read(
        &mut self,
        access: AuthorizedWorkspaceReadAccess,
    ) -> Result<(), PluginWorkspaceIoCommitError> {
        self.ensure_access_identity(access.identity_proof())?;
        let token = access.into_policy_token();
        self.push(PendingPluginWorkspaceIo::Read(token))
    }

    /// Queues a workspace write that has already passed plugin capability authorization.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoCommitError`] when the payload exceeds the write budget or the
    /// batch is full.
    pub fn push_write(
        &mut self,
        access: AuthorizedWorkspaceWriteAccess,
        bytes: impl Into<Vec<u8>>,
    ) -> Result<(), PluginWorkspaceIoCommitError> {
        self.ensure_access_identity(access.identity_proof())?;
        let token = access.into_policy_token();
        let payload = PluginWorkspaceWritePayload::try_new(
            bytes,
            self.request_ledger.max_write_bytes_limit(),
        )?;
        self.push(PendingPluginWorkspaceIo::Write { token, payload })
    }

    /// Returns the number of queued workspace operations.
    #[must_use]
    pub const fn request_count(&self) -> usize {
        self.requests.len()
    }

    /// Returns requests charged against the cap but tracked outside this I/O batch.
    #[must_use]
    pub const fn non_deferred_request_count(&self) -> usize {
        self.request_ledger.non_deferred_request_count()
    }

    /// Returns all workspace requests accepted during this update session.
    #[must_use]
    pub const fn accepted_request_count(&self) -> usize {
        self.request_ledger
            .accepted_request_count(self.request_count())
    }

    /// Returns the read byte cap shared by deferred reads and direct guest observations.
    #[must_use]
    pub const fn max_read_bytes(&self) -> u64 {
        self.request_ledger.max_read_bytes()
    }

    /// Charges a non-queued direct observation against the per-update request cap.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoCommitError`] when the session has already spent its request
    /// budget.
    pub(in crate::plugin::host) fn charge_direct_observation(
        &mut self,
    ) -> Result<PluginWorkspaceReadByteLimit, PluginWorkspaceIoCommitError> {
        Ok(self.prepare_non_deferred_read()?.commit())
    }

    /// Reads a workspace file directly through filesystem policy for guest delivery.
    ///
    /// Direct observations are not deferred I/O work, but they still belong to this update's
    /// workspace request ledger. The access proof identity is checked before charging budget or
    /// touching filesystem policy.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoCommitError`] when the access proof belongs to another plugin
    /// or the session has already spent its request budget.
    pub(in crate::plugin::host) fn observe_existing(
        &mut self,
        access: AuthorizedWorkspaceReadAccess,
        filesystem: &FilesystemConfig,
    ) -> Result<PluginWorkspaceObserveOutcome, PluginWorkspaceIoCommitError> {
        self.ensure_access_identity(access.identity_proof())?;
        let max_read_bytes = self.charge_direct_observation()?;
        let outcome = access
            .into_policy_token()
            .read_existing(filesystem, max_read_bytes)
            .map(|file| {
                PluginWorkspaceReadSuccess::new(
                    EscapedDisplayText::from_path(&file.path),
                    file.bytes,
                )
            });
        Ok(PluginWorkspaceObserveOutcome::from_read_result(outcome))
    }

    /// Prepares a read request whose durable work is tracked by another pending batch.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoCommitError`] when the session has already spent its request
    /// budget.
    pub(in crate::plugin::host) fn prepare_non_deferred_read(
        &mut self,
    ) -> Result<PluginWorkspaceNonDeferredReadPermit<'_>, PluginWorkspaceIoCommitError> {
        self.request_ledger
            .prepare_non_deferred_read(self.requests.len())
    }

    /// Seals workspace I/O after a guest update returns successfully.
    #[must_use]
    pub fn seal(self) -> SealedPluginWorkspaceIoBatch {
        let Self {
            request_ledger,
            requests,
        } = self;
        let ledger = request_ledger.into_parts();
        SealedPluginWorkspaceIoBatch {
            identity: ledger.identity,
            budget: ledger.budget,
            requests,
        }
    }

    /// Discards workspace I/O after a guest update fails before successful return.
    #[must_use]
    pub fn discard(self) -> PluginWorkspaceIoDiscardReport {
        let Self {
            request_ledger,
            requests,
        } = self;
        let ledger = request_ledger.into_parts();
        PluginWorkspaceIoDiscardReport {
            identity: ledger.identity,
            discarded_requests: requests.len(),
        }
    }

    /// Queues one workspace operation.
    fn push(
        &mut self,
        request: PendingPluginWorkspaceIo,
    ) -> Result<(), PluginWorkspaceIoCommitError> {
        self.request_ledger
            .ensure_deferred_request_budget(self.requests.len())?;
        self.requests.push(request);
        Ok(())
    }

    /// Checks that a workspace access proof belongs to this batch's plugin.
    fn ensure_access_identity(
        &self,
        access_identity: &PluginIdentity,
    ) -> Result<(), PluginWorkspaceIoCommitError> {
        if access_identity != self.identity_proof() {
            return Err(PluginWorkspaceIoCommitError::IdentityMismatch {
                mismatch_source: PluginWorkspaceIoIdentityMismatchSource::AccessProof,
                batch_identity: self.identity_proof().clone(),
                provided_identity: access_identity.clone(),
            });
        }
        Ok(())
    }
}

impl Debug for PendingPluginWorkspaceIoBatch {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PendingPluginWorkspaceIoBatch")
            .field("identity", &self.identity_proof())
            .field("execution_model", &self.execution_model())
            .field("request_count", &self.request_count())
            .field("request_ledger", &self.request_ledger)
            .field(
                "request_shapes",
                &self
                    .requests
                    .iter()
                    .map(PendingPluginWorkspaceIo::shape)
                    .collect::<Vec<_>>(),
            )
            .finish()
    }
}

/// Workspace I/O from a guest update that reached the successful-return boundary.
#[derive(Eq, PartialEq)]
pub struct SealedPluginWorkspaceIoBatch {
    /// Stable plugin identity for diagnostics.
    identity: PluginIdentity,
    /// Per-update workspace I/O budget.
    budget: PluginWorkspaceIoBudget,
    /// Accepted workspace operations.
    requests: Vec<PendingPluginWorkspaceIo>,
}

impl SealedPluginWorkspaceIoBatch {
    /// Returns the plugin identity.
    #[must_use]
    pub fn identity(&self) -> &str {
        self.identity_proof().as_str()
    }

    /// Returns the identity proof retained across the successful-return boundary.
    #[must_use]
    pub const fn identity_proof(&self) -> &PluginIdentity {
        &self.identity
    }

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

    /// Returns whether this sealed batch carries no filesystem-policy work.
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.requests.is_empty()
    }

    /// Enqueues sealed workspace I/O for host-owned filesystem-policy execution.
    ///
    /// Failed admission returns the still-sealed batch so callers can retry or explicitly discard
    /// it without losing successful-return work as a side effect of queue saturation.
    ///
    /// # Errors
    ///
    /// Returns [`PluginWorkspaceIoWorkerEnqueueError`] when the worker queue rejects admission.
    pub fn enqueue(
        self,
        queue: &mut PluginWorkspaceIoWorkerQueue,
    ) -> Result<(), PluginWorkspaceIoWorkerEnqueueError> {
        queue.push(self)
    }

    /// Discards sealed workspace I/O that will not be retried before filesystem policy execution.
    #[must_use]
    pub fn discard(self) -> PluginWorkspaceIoDiscardReport {
        let Self {
            identity,
            budget: _budget,
            requests,
        } = self;
        PluginWorkspaceIoDiscardReport {
            identity,
            discarded_requests: requests.len(),
        }
    }

    /// Executes sealed workspace I/O synchronously through `fs_utils`.
    ///
    /// This is the v1 stepping stone before an async worker exists. It preserves the observable
    /// async contract by keeping requests discardable until successful guest return and by routing
    /// all filesystem access through policy tokens.
    #[must_use]
    pub fn execute_synchronous(self, filesystem: &FilesystemConfig) -> PluginWorkspaceIoReport {
        let mut completions = Vec::with_capacity(self.requests.len());
        for request in self.requests {
            completions.push(match request {
                PendingPluginWorkspaceIo::Read(token) => {
                    let path = WorkspacePath::from_ref(token.workspace_path());
                    let outcome = token
                        .read_existing(filesystem, self.budget.max_read_bytes_limit())
                        .map(|file| {
                            PluginWorkspaceReadSuccess::new(
                                EscapedDisplayText::from_path(&file.path),
                                file.bytes,
                            )
                        });
                    PluginWorkspaceIoCompletion::Read(PluginWorkspaceReadCompletion::new(
                        path, outcome,
                    ))
                }
                PendingPluginWorkspaceIo::Write { token, payload } => {
                    let path = WorkspacePath::from_ref(token.workspace_path());
                    let outcome = token
                        .write_atomic(payload.as_bytes(), filesystem, payload.max_size())
                        .map(PluginWorkspaceWriteSuccess::new);
                    PluginWorkspaceIoCompletion::Write(PluginWorkspaceWriteCompletion::new(
                        path, outcome,
                    ))
                }
            });
        }
        PluginWorkspaceIoReport::new(self.identity, completions)
    }
}

impl Debug for SealedPluginWorkspaceIoBatch {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("SealedPluginWorkspaceIoBatch")
            .field("identity", &self.identity)
            .field(
                "execution_model",
                &PluginWorkspaceIoExecutionModel::DeferredSynchronous,
            )
            .field("budget", &self.budget)
            .field("request_count", &self.requests.len())
            .field(
                "request_shapes",
                &self
                    .requests
                    .iter()
                    .map(PendingPluginWorkspaceIo::shape)
                    .collect::<Vec<_>>(),
            )
            .finish()
    }
}

/// Report from discarding workspace I/O before filesystem policy execution.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PluginWorkspaceIoDiscardReport {
    /// Stable plugin identity for diagnostics.
    identity: PluginIdentity,
    /// Number of requests discarded before filesystem policy execution.
    discarded_requests: usize,
}

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

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

    /// Returns the number of discarded workspace I/O requests.
    #[must_use]
    pub const fn discarded_requests(&self) -> usize {
        self.discarded_requests
    }
}

/// Deferred workspace I/O operation.
#[derive(Eq, PartialEq)]
enum PendingPluginWorkspaceIo {
    /// Deferred workspace read.
    Read(PluginWorkspaceReadToken),
    /// Deferred workspace write.
    Write {
        /// Filesystem-policy bridge token.
        token: PluginWorkspaceWriteToken,
        /// Payload already checked against the write byte cap.
        payload: PluginWorkspaceWritePayload,
    },
}

impl PendingPluginWorkspaceIo {
    /// Returns a redacted diagnostic shape.
    fn shape(&self) -> PendingPluginWorkspaceIoShape {
        match self {
            Self::Read(token) => PendingPluginWorkspaceIoShape::Read {
                path_byte_len: token.workspace_relative_path().len(),
            },
            Self::Write { token, payload } => PendingPluginWorkspaceIoShape::Write {
                path_byte_len: token.workspace_relative_path().len(),
                byte_len: payload.len(),
            },
        }
    }
}

/// Workspace artifact write payload accepted into a pending batch.
#[derive(Eq, PartialEq)]
pub(in crate::plugin::host::workspace_io) struct PluginWorkspaceWritePayload {
    /// Payload bytes retained until the successful-return batch reaches filesystem policy.
    bytes: Vec<u8>,
    /// Cap that admitted these bytes.
    max_size: PluginWorkspaceWriteByteLimit,
}

impl PluginWorkspaceWritePayload {
    /// Builds a payload proof after checking the workspace write budget.
    pub(in crate::plugin::host::workspace_io) fn try_new(
        bytes: impl Into<Vec<u8>>,
        max_size: PluginWorkspaceWriteByteLimit,
    ) -> Result<Self, PluginWorkspaceIoCommitError> {
        let bytes = bytes.into();
        let size = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
        if size > max_size.get() {
            return Err(PluginWorkspaceIoCommitError::PayloadTooLarge {
                size,
                max_size: max_size.get(),
            });
        }
        Ok(Self { bytes, max_size })
    }

    /// Returns the admitted payload bytes.
    pub(in crate::plugin::host::workspace_io) fn as_bytes(&self) -> &[u8] {
        &self.bytes
    }

    /// Returns the admitted payload byte length.
    pub(in crate::plugin::host::workspace_io) const fn len(&self) -> usize {
        self.bytes.len()
    }

    /// Returns the write cap that admitted the payload.
    pub(in crate::plugin::host::workspace_io) const fn max_size(
        &self,
    ) -> PluginWorkspaceWriteByteLimit {
        self.max_size
    }
}

impl Debug for PluginWorkspaceWritePayload {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("PluginWorkspaceWritePayload")
            .field("byte_len", &self.len())
            .field("max_size", &self.max_size.get())
            .finish_non_exhaustive()
    }
}

/// Redacted workspace I/O request shape.
#[derive(Clone, Debug, Eq, PartialEq)]
enum PendingPluginWorkspaceIoShape {
    /// Deferred read request.
    Read {
        /// Path bytes.
        path_byte_len: usize,
    },
    /// Deferred write request.
    Write {
        /// Path bytes.
        path_byte_len: usize,
        /// Payload byte length for writes.
        byte_len: usize,
    },
}