moli/moli-streams/src/transform.rs

//! Runtime-independent `TransformStream` coordination.
//!
//! The renderer owns JavaScript chunks, reasons, callbacks, promises, and the
//! controller's finish promise/resolver pair. This module coordinates start,
//! write, backpressure, and admission to a shared finish operation. The renderer
//! applies cancel/abort/flush settlement in the corresponding promise reaction.

use crate::readable::ReadableState;
use crate::strategy::StrategySnapshot;
use crate::writable::WritableState;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformMode {
    Identity,
    Callback,
    TextEncoder,
    TextDecoder,
}

impl TransformMode {
    #[must_use]
    pub fn from_storage(mode: Option<&str>, has_transformer: bool) -> Self {
        match mode {
            Some("text-encoder") => Self::TextEncoder,
            Some("text-decoder") => Self::TextDecoder,
            _ if has_transformer => Self::Callback,
            _ => Self::Identity,
        }
    }

    #[must_use]
    pub const fn write_algorithm(self) -> TransformWriteAlgorithm {
        match self {
            Self::Identity => TransformWriteAlgorithm::Identity,
            Self::Callback => TransformWriteAlgorithm::Callback,
            Self::TextEncoder => TransformWriteAlgorithm::TextEncoder,
            Self::TextDecoder => TransformWriteAlgorithm::TextDecoder,
        }
    }

    #[must_use]
    pub const fn flush_algorithm(self) -> TransformFlushAlgorithm {
        match self {
            Self::Callback => TransformFlushAlgorithm::Callback,
            Self::TextDecoder => TransformFlushAlgorithm::TextDecoder,
            Self::Identity | Self::TextEncoder => TransformFlushAlgorithm::None,
        }
    }

    #[must_use]
    pub const fn cancel_algorithm(self) -> TransformCancelAlgorithm {
        match self {
            Self::Callback => TransformCancelAlgorithm::Callback,
            Self::Identity | Self::TextEncoder | Self::TextDecoder => {
                TransformCancelAlgorithm::None
            }
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformWriteAlgorithm {
    Identity,
    Callback,
    TextEncoder,
    TextDecoder,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformFlushAlgorithm {
    None,
    Callback,
    TextDecoder,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformCancelAlgorithm {
    None,
    Callback,
}

#[derive(Clone, Copy, Debug, PartialEq)]
pub struct TransformReadableSnapshot {
    state: ReadableState,
    pending_read_count: usize,
    pipe_registered: bool,
    strategy: StrategySnapshot,
}

impl TransformReadableSnapshot {
    #[must_use]
    pub const fn new(
        state: ReadableState,
        pending_read_count: usize,
        pipe_registered: bool,
        strategy: StrategySnapshot,
    ) -> Self {
        Self {
            state,
            pending_read_count,
            pipe_registered,
            strategy,
        }
    }

    #[must_use]
    pub const fn state(self) -> ReadableState {
        self.state
    }

    #[must_use]
    pub const fn pending_read_count(self) -> usize {
        self.pending_read_count
    }

    #[must_use]
    pub const fn pipe_registered(self) -> bool {
        self.pipe_registered
    }

    #[must_use]
    pub fn backpressure(self) -> TransformBackpressure {
        if !matches!(self.state, ReadableState::Readable) {
            return TransformBackpressure::Terminal;
        }
        if self.pending_read_count > 0 || self.pipe_registered || self.strategy.has_capacity() {
            TransformBackpressure::Ready
        } else {
            TransformBackpressure::Backpressured
        }
    }

    #[must_use]
    pub fn can_accept_chunk(self) -> bool {
        matches!(self.backpressure(), TransformBackpressure::Ready)
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformBackpressure {
    Ready,
    Backpressured,
    Terminal,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FinishResidenceState {
    Available,
    Claimed,
}

impl FinishResidenceState {
    #[must_use]
    pub const fn from_storage(has_finish_promise: bool) -> Self {
        if has_finish_promise {
            Self::Claimed
        } else {
            Self::Available
        }
    }
}

#[derive(Clone, Copy, Debug, PartialEq)]
pub struct TransformSnapshot {
    readable: TransformReadableSnapshot,
    writable_state: WritableState,
    mode: TransformMode,
    start_pending: bool,
    pending_operation_count: usize,
    finish: FinishResidenceState,
}

impl TransformSnapshot {
    #[must_use]
    pub const fn new(
        readable: TransformReadableSnapshot,
        writable_state: WritableState,
        mode: TransformMode,
        start_pending: bool,
        pending_operation_count: usize,
        finish: FinishResidenceState,
    ) -> Self {
        Self {
            readable,
            writable_state,
            mode,
            start_pending,
            pending_operation_count,
            finish,
        }
    }

    #[must_use]
    pub const fn readable(self) -> TransformReadableSnapshot {
        self.readable
    }

    #[must_use]
    pub const fn writable_state(self) -> WritableState {
        self.writable_state
    }

    #[must_use]
    pub const fn mode(self) -> TransformMode {
        self.mode
    }

    #[must_use]
    pub fn plan_write_admission(self) -> TransformWriteAdmissionPlan {
        if matches!(self.mode, TransformMode::Callback)
            || self.start_pending
            || self.pending_operation_count > 0
            || !self.readable.can_accept_chunk()
        {
            TransformWriteAdmissionPlan::Queue
        } else {
            TransformWriteAdmissionPlan::Run(self.mode.write_algorithm())
        }
    }

    #[must_use]
    pub const fn plan_queued_write_algorithm(self) -> TransformWriteAlgorithm {
        self.mode.write_algorithm()
    }

    #[must_use]
    pub const fn plan_close_admission(self) -> TransformCloseAdmissionPlan {
        if self.start_pending || self.pending_operation_count > 0 {
            TransformCloseAdmissionPlan::Queue
        } else {
            TransformCloseAdmissionPlan::Run
        }
    }

    #[must_use]
    pub const fn plan_finish(self, operation: FinishOperation) -> FinishClaimPlan {
        if matches!(self.finish, FinishResidenceState::Claimed) {
            return FinishClaimPlan::Reuse;
        }
        let algorithm = match operation {
            FinishOperation::WritableClose => FinishAlgorithm::Flush(self.mode.flush_algorithm()),
            FinishOperation::ReadableCancel | FinishOperation::WritableAbort => {
                FinishAlgorithm::Cancel(self.mode.cancel_algorithm())
            }
        };
        FinishClaimPlan::Claim { algorithm }
    }

    #[must_use]
    pub const fn plan_start_settlement(self, outcome: AlgorithmOutcome) -> StartSettlementPlan {
        match outcome {
            AlgorithmOutcome::Fulfilled => StartSettlementPlan::ClearPendingAndPump,
            AlgorithmOutcome::Rejected => StartSettlementPlan::ClearPendingAndErrorBoth,
        }
    }

    #[must_use]
    pub const fn plan_write_settlement(
        self,
        outcome: AlgorithmOutcome,
        direct_write_size: bool,
    ) -> WriteSettlementPlan {
        match outcome {
            AlgorithmOutcome::Fulfilled => WriteSettlementPlan::Fulfill {
                finish_direct_write: direct_write_size,
                drain_pipe: direct_write_size,
            },
            AlgorithmOutcome::Rejected => WriteSettlementPlan::Reject {
                finish_direct_write: direct_write_size,
                error: self.plan_error(),
            },
        }
    }

    #[must_use]
    pub const fn plan_writable_close_settlement(
        self,
        outcome: AlgorithmOutcome,
    ) -> WritableCloseSettlementPlan {
        match outcome {
            AlgorithmOutcome::Fulfilled => WritableCloseSettlementPlan::MarkClosed,
            AlgorithmOutcome::Rejected => {
                WritableCloseSettlementPlan::ClearInFlightAndErrorWritable
            }
        }
    }

    #[must_use]
    pub const fn plan_error(self) -> TransformErrorPlan {
        match self.readable.state {
            ReadableState::Readable => TransformErrorPlan {
                reason: ErrorReasonSource::Provided,
                readable: ReadableErrorAction::Error,
            },
            ReadableState::Closed => TransformErrorPlan {
                reason: ErrorReasonSource::Provided,
                readable: ReadableErrorAction::Keep,
            },
            ReadableState::Errored => TransformErrorPlan {
                reason: ErrorReasonSource::ReadableStored,
                readable: ReadableErrorAction::Keep,
            },
        }
    }

    #[must_use]
    pub const fn plan_terminate(self) -> TerminatePlan {
        TerminatePlan {
            readable: if matches!(self.readable.state, ReadableState::Readable) {
                ReadableTerminateAction::Close
            } else {
                ReadableTerminateAction::Keep
            },
        }
    }

    #[must_use]
    pub const fn plan_finish_setup_failure(
        self,
        operation: FinishOperation,
    ) -> FinishSetupFailurePlan {
        match operation {
            FinishOperation::ReadableCancel => {
                FinishSetupFailurePlan::ErrorWritableWithOriginalReasonAndReject
            }
            FinishOperation::WritableAbort => {
                FinishSetupFailurePlan::ErrorReadableWithOriginalReasonAndReject
            }
            FinishOperation::WritableClose => {
                FinishSetupFailurePlan::ErrorReadableWithUndefinedAndReject
            }
        }
    }

    #[must_use]
    pub const fn plan_enqueue_failure(
        self,
        failure: TransformEnqueueFailure,
    ) -> TransformEnqueueFailurePlan {
        match failure {
            TransformEnqueueFailure::ClosedOrErrored => TransformEnqueueFailurePlan {
                returned_error: EnqueueErrorSource::SynthesizedTypeError,
                propagation: self.plan_error(),
            },
            TransformEnqueueFailure::Strategy => {
                let propagation = self.plan_error();
                TransformEnqueueFailurePlan {
                    returned_error: match propagation.reason {
                        ErrorReasonSource::Provided => EnqueueErrorSource::Provided,
                        ErrorReasonSource::ReadableStored => EnqueueErrorSource::ReadableStored,
                    },
                    propagation,
                }
            }
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformWriteAdmissionPlan {
    Queue,
    Run(TransformWriteAlgorithm),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformCloseAdmissionPlan {
    Queue,
    Run,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FinishOperation {
    ReadableCancel,
    WritableAbort,
    WritableClose,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FinishAlgorithm {
    Cancel(TransformCancelAlgorithm),
    Flush(TransformFlushAlgorithm),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FinishClaimPlan {
    Reuse,
    Claim { algorithm: FinishAlgorithm },
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum AlgorithmOutcome {
    Fulfilled,
    Rejected,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StartSettlementPlan {
    ClearPendingAndPump,
    ClearPendingAndErrorBoth,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WriteSettlementPlan {
    Fulfill {
        finish_direct_write: bool,
        drain_pipe: bool,
    },
    Reject {
        finish_direct_write: bool,
        error: TransformErrorPlan,
    },
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WritableCloseSettlementPlan {
    MarkClosed,
    ClearInFlightAndErrorWritable,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ErrorReasonSource {
    Provided,
    ReadableStored,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReadableErrorAction {
    Keep,
    Error,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TransformErrorPlan {
    reason: ErrorReasonSource,
    readable: ReadableErrorAction,
}

impl TransformErrorPlan {
    #[must_use]
    pub const fn reason(self) -> ErrorReasonSource {
        self.reason
    }

    #[must_use]
    pub const fn readable(self) -> ReadableErrorAction {
        self.readable
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ReadableTerminateAction {
    Keep,
    Close,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TerminatePlan {
    readable: ReadableTerminateAction,
}

impl TerminatePlan {
    #[must_use]
    pub const fn readable(self) -> ReadableTerminateAction {
        self.readable
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FinishSetupFailurePlan {
    ErrorWritableWithOriginalReasonAndReject,
    ErrorReadableWithOriginalReasonAndReject,
    ErrorReadableWithUndefinedAndReject,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransformEnqueueFailure {
    ClosedOrErrored,
    Strategy,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum EnqueueErrorSource {
    SynthesizedTypeError,
    Provided,
    ReadableStored,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct TransformEnqueueFailurePlan {
    returned_error: EnqueueErrorSource,
    propagation: TransformErrorPlan,
}

impl TransformEnqueueFailurePlan {
    #[must_use]
    pub const fn returned_error(self) -> EnqueueErrorSource {
        self.returned_error
    }

    #[must_use]
    pub const fn propagation(self) -> TransformErrorPlan {
        self.propagation
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn readable(
        state: ReadableState,
        pending_reads: usize,
        pipe_registered: bool,
        high_water_mark: f64,
        total_size: f64,
    ) -> TransformReadableSnapshot {
        TransformReadableSnapshot::new(
            state,
            pending_reads,
            pipe_registered,
            StrategySnapshot::new(high_water_mark, total_size),
        )
    }

    fn snapshot(
        readable: TransformReadableSnapshot,
        writable_state: WritableState,
        mode: TransformMode,
    ) -> TransformSnapshot {
        TransformSnapshot::new(
            readable,
            writable_state,
            mode,
            false,
            0,
            FinishResidenceState::Available,
        )
    }

    #[test]
    fn storage_mode_owns_write_flush_and_cancel_routing() {
        let cases = [
            (
                None,
                false,
                TransformMode::Identity,
                TransformWriteAlgorithm::Identity,
                TransformFlushAlgorithm::None,
                TransformCancelAlgorithm::None,
            ),
            (
                None,
                true,
                TransformMode::Callback,
                TransformWriteAlgorithm::Callback,
                TransformFlushAlgorithm::Callback,
                TransformCancelAlgorithm::Callback,
            ),
            (
                Some("text-encoder"),
                true,
                TransformMode::TextEncoder,
                TransformWriteAlgorithm::TextEncoder,
                TransformFlushAlgorithm::None,
                TransformCancelAlgorithm::None,
            ),
            (
                Some("text-decoder"),
                false,
                TransformMode::TextDecoder,
                TransformWriteAlgorithm::TextDecoder,
                TransformFlushAlgorithm::TextDecoder,
                TransformCancelAlgorithm::None,
            ),
        ];
        for (stored, has_transformer, mode, write, flush, cancel) in cases {
            let decoded = TransformMode::from_storage(stored, has_transformer);
            assert_eq!(decoded, mode);
            assert_eq!(decoded.write_algorithm(), write);
            assert_eq!(decoded.flush_algorithm(), flush);
            assert_eq!(decoded.cancel_algorithm(), cancel);
        }
    }

    #[test]
    fn readable_backpressure_combines_lifecycle_demand_pipe_and_strategy() {
        assert_eq!(
            readable(ReadableState::Closed, 1, true, 1.0, 0.0).backpressure(),
            TransformBackpressure::Terminal
        );
        assert_eq!(
            readable(ReadableState::Readable, 1, false, 0.0, 0.0).backpressure(),
            TransformBackpressure::Ready
        );
        assert_eq!(
            readable(ReadableState::Readable, 0, true, 0.0, 0.0).backpressure(),
            TransformBackpressure::Ready
        );
        assert_eq!(
            readable(ReadableState::Readable, 0, false, 2.0, 1.0).backpressure(),
            TransformBackpressure::Ready
        );
        assert_eq!(
            readable(ReadableState::Readable, 0, false, 1.0, 1.0).backpressure(),
            TransformBackpressure::Backpressured
        );
    }

    #[test]
    fn write_and_close_admission_cover_callback_start_queue_and_capacity() {
        let ready = readable(ReadableState::Readable, 1, false, 0.0, 0.0);
        assert_eq!(
            snapshot(ready, WritableState::Writable, TransformMode::Identity)
                .plan_write_admission(),
            TransformWriteAdmissionPlan::Run(TransformWriteAlgorithm::Identity)
        );
        assert_eq!(
            snapshot(ready, WritableState::Writable, TransformMode::Callback)
                .plan_write_admission(),
            TransformWriteAdmissionPlan::Queue
        );

        for (start_pending, pending_count, backpressured) in
            [(true, 0, false), (false, 1, false), (false, 0, true)]
        {
            let readable = if backpressured {
                readable(ReadableState::Readable, 0, false, 0.0, 0.0)
            } else {
                ready
            };
            let current = TransformSnapshot::new(
                readable,
                WritableState::Writable,
                TransformMode::Identity,
                start_pending,
                pending_count,
                FinishResidenceState::Available,
            );
            assert_eq!(
                current.plan_write_admission(),
                TransformWriteAdmissionPlan::Queue
            );
            assert_eq!(
                current.plan_close_admission(),
                if start_pending || pending_count > 0 {
                    TransformCloseAdmissionPlan::Queue
                } else {
                    TransformCloseAdmissionPlan::Run
                }
            );
        }
    }

    #[test]
    fn finish_residence_is_claimed_once_with_operation_specific_algorithm() {
        let available = snapshot(
            readable(ReadableState::Readable, 0, false, 1.0, 0.0),
            WritableState::Writable,
            TransformMode::Callback,
        );
        assert_eq!(
            available.plan_finish(FinishOperation::ReadableCancel),
            FinishClaimPlan::Claim {
                algorithm: FinishAlgorithm::Cancel(TransformCancelAlgorithm::Callback),
            }
        );
        assert_eq!(
            available.plan_finish(FinishOperation::WritableClose),
            FinishClaimPlan::Claim {
                algorithm: FinishAlgorithm::Flush(TransformFlushAlgorithm::Callback),
            }
        );
        let claimed = TransformSnapshot::new(
            available.readable(),
            available.writable_state(),
            available.mode(),
            false,
            0,
            FinishResidenceState::Claimed,
        );
        assert_eq!(
            claimed.plan_finish(FinishOperation::WritableAbort),
            FinishClaimPlan::Reuse
        );
    }

    #[test]
    fn transform_error_preserves_the_first_readable_error() {
        let plan = snapshot(
            readable(ReadableState::Errored, 0, false, 1.0, 0.0),
            WritableState::Writable,
            TransformMode::Identity,
        )
        .plan_error();
        assert_eq!(plan.reason(), ErrorReasonSource::ReadableStored);
        assert_eq!(plan.readable(), ReadableErrorAction::Keep);

        let plan = snapshot(
            readable(ReadableState::Readable, 0, false, 1.0, 0.0),
            WritableState::Writable,
            TransformMode::Identity,
        )
        .plan_error();
        assert_eq!(plan.reason(), ErrorReasonSource::Provided);
        assert_eq!(plan.readable(), ReadableErrorAction::Error);
    }

    #[test]
    fn setup_failures_and_enqueue_failures_preserve_error_ownership() {
        let current = snapshot(
            readable(ReadableState::Readable, 0, false, 1.0, 0.0),
            WritableState::Writable,
            TransformMode::Identity,
        );
        assert_eq!(
            current.plan_finish_setup_failure(FinishOperation::ReadableCancel),
            FinishSetupFailurePlan::ErrorWritableWithOriginalReasonAndReject
        );
        assert_eq!(
            current.plan_finish_setup_failure(FinishOperation::WritableClose),
            FinishSetupFailurePlan::ErrorReadableWithUndefinedAndReject
        );
        let closed = current.plan_enqueue_failure(TransformEnqueueFailure::ClosedOrErrored);
        assert_eq!(
            closed.returned_error(),
            EnqueueErrorSource::SynthesizedTypeError
        );

        let errored = snapshot(
            readable(ReadableState::Errored, 0, false, 1.0, 0.0),
            WritableState::Writable,
            TransformMode::Identity,
        )
        .plan_enqueue_failure(TransformEnqueueFailure::Strategy);
        assert_eq!(errored.returned_error(), EnqueueErrorSource::ReadableStored);
        assert_eq!(
            errored.propagation().reason(),
            ErrorReasonSource::ReadableStored
        );
    }

    #[test]
    fn start_write_and_terminate_plans_are_explicit() {
        let current = snapshot(
            readable(ReadableState::Readable, 0, false, 1.0, 0.0),
            WritableState::Writable,
            TransformMode::Identity,
        );
        assert_eq!(
            current.plan_start_settlement(AlgorithmOutcome::Fulfilled),
            StartSettlementPlan::ClearPendingAndPump
        );
        assert_eq!(
            current.plan_start_settlement(AlgorithmOutcome::Rejected),
            StartSettlementPlan::ClearPendingAndErrorBoth
        );
        assert_eq!(
            current.plan_write_settlement(AlgorithmOutcome::Fulfilled, true),
            WriteSettlementPlan::Fulfill {
                finish_direct_write: true,
                drain_pipe: true,
            }
        );
        assert_eq!(
            current.plan_write_settlement(AlgorithmOutcome::Rejected, false),
            WriteSettlementPlan::Reject {
                finish_direct_write: false,
                error: current.plan_error(),
            }
        );
        assert_eq!(
            current.plan_writable_close_settlement(AlgorithmOutcome::Fulfilled),
            WritableCloseSettlementPlan::MarkClosed
        );
        assert_eq!(
            current.plan_writable_close_settlement(AlgorithmOutcome::Rejected),
            WritableCloseSettlementPlan::ClearInFlightAndErrorWritable
        );
        assert_eq!(
            current.plan_terminate().readable(),
            ReadableTerminateAction::Close
        );
    }
}