qubit-codec 0.17.4

// =============================================================================
//    Copyright (c) 2026 Haixing Hu.
//
//    SPDX-License-Identifier: Apache-2.0
//
//    Licensed under the Apache License, Version 2.0.
// =============================================================================
//! Buffered asynchronous output driver for streaming transcoders.

use core::fmt;
use std::collections::TryReserveError;
use std::future::poll_fn;
use std::io::Error;
use std::io::Result;
use std::pin::Pin;
use std::task::Context;
use std::task::Poll;

use qubit_io::AsyncBufferedOutput;
use qubit_io::AsyncOutput;
use qubit_io::Buffer;
use qubit_utils::SliceRange;
use qubit_utils::allocation_error;

use super::internal::capacity_error_to_invalid_data;
use super::internal::validate_encode_progress;
use crate::TranscodeEncoder;
use crate::TranscodeProgress;
use crate::TranscodeStatus;
use crate::Transcoder;

/// Buffers asynchronous transcoder output while preserving pending units.
///
/// This adapter is the asynchronous counterpart to
/// [`super::TranscodeEncodeOutput`]. It owns the unit-level buffer, so a
/// future cancelled while the wrapped output is pending retains every unit
/// that the transcoder has already produced. Each transcode operation returns
/// after exactly one encoder invocation, so its progress is the explicit
/// source-consumption commit boundary.
///
/// Cancellation and pending-output recovery assume that wrapped I/O calls
/// obey their normal return contracts. This adapter does not track poisoning
/// or indeterminate delivery after an inner I/O panic. If such a panic is
/// caught, reconcile external side effects before retrying or replaying
/// pending units.
///
/// # Type Parameters
///
/// * `O` - Wrapped asynchronous unit output.
///
/// # Examples
///
/// ```
/// use core::convert::Infallible;
/// use core::num::NonZeroUsize;
/// use std::future::Future;
/// use std::io::Result;
/// use std::pin::{pin, Pin};
/// use std::task::{Context, Poll, Waker};
///
/// use qubit_codec::{AsyncTranscodeEncodeOutput, Codec, CodecTranscodeEncoder};
/// use qubit_codec::{DecodeFailure, TranscodeEncoder};
/// use qubit_io::AsyncOutput;
///
/// struct ByteCodec;
/// impl Codec for ByteCodec {
///     type Value = u8;
///     type Unit = u8;
///     type DecodeError = Infallible;
///     type EncodeError = Infallible;
///     const MIN_UNITS_PER_VALUE: usize = 1;
///     const MAX_DECODE_UNITS_PER_VALUE: usize = 1;
///     const MAX_ENCODE_UNITS_PER_VALUE: usize = 1;
///     unsafe fn decode(&mut self, input: &[u8], index: usize)
///         -> std::result::Result<(u8, NonZeroUsize), DecodeFailure<Infallible>> {
///         Ok((input[index], NonZeroUsize::MIN))
///     }
///     unsafe fn encode(&mut self, value: &u8, output: &mut [u8], index: usize)
///         -> std::result::Result<usize, Infallible> {
///         output[index] = *value;
///         Ok(1)
///     }
/// }
///
/// #[derive(Default)]
/// struct Sink { bytes: Vec<u8>, flushes: usize }
/// impl AsyncOutput for Sink {
///     type Item = u8;
///     unsafe fn poll_write_unchecked(
///         self: Pin<&mut Self>, _: &mut Context<'_>, units: &[u8], index: usize, count: usize,
///     ) -> Poll<Result<usize>> {
///         self.get_mut().bytes.extend_from_slice(&units[index..index + count]);
///         Poll::Ready(Ok(count))
///     }
///     fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<()>> {
///         self.get_mut().flushes += 1;
///         Poll::Ready(Ok(()))
///     }
/// }
///
/// async fn run_example() -> Result<()> {
///     let mut encoder = CodecTranscodeEncoder::new(ByteCodec);
///     let mut output = AsyncTranscodeEncodeOutput::with_capacity(Sink::default(), 8);
///     let mut map_error = |_| std::io::Error::other("encode error");
///     output.reset_async(&mut encoder, &mut map_error).await?;
///     let values = [1_u8, 2, 3];
///     let mut offset = 0;
///     while offset < values.len() {
///         let progress = output.transcode_async(
///             &mut encoder, &mut map_error, &values, offset, values.len() - offset,
///         ).await?;
///         assert!(progress.read() > 0);
///         offset += progress.read();
///     }
///     output.finish_async(&mut encoder, &mut map_error).await?;
///     assert_eq!(output.pending_len(), values.len());
///     output.flush_async().await?;
///     let (sink, pending) = output.into_parts();
///     assert!(pending.is_empty());
///     assert_eq!(sink.bytes, vec![1, 2, 3]);
///     assert_eq!(sink.flushes, 1);
///     Ok(())
/// }
///
/// let mut future = pin!(run_example());
/// let mut context = Context::from_waker(Waker::noop());
/// match future.as_mut().poll(&mut context) {
///     Poll::Ready(result) => result.unwrap(),
///     Poll::Pending => panic!("the example sink always completes immediately"),
/// }
/// ```
///
/// `finish_async` commits final units to the adapter buffer. `flush_async`
/// delivers those units and flushes the wrapped sink.
#[must_use]
pub struct AsyncTranscodeEncodeOutput<O>
where
    O: AsyncOutput,
    O::Item: Clone + Default,
{
    /// Owned output buffer retaining units until the wrapped sink accepts them.
    output: AsyncBufferedOutput<O>,
    /// Minimum spare units requested by the previous encoder progress report.
    required_spare: usize,
}

impl<O> AsyncTranscodeEncodeOutput<O>
where
    O: AsyncOutput,
    O::Item: Clone + Default,
{
    /// Creates an asynchronous encoder output with the default unit capacity.
    ///
    /// # Parameters
    ///
    /// * `inner` - Unit output receiving transcoder data.
    ///
    /// # Returns
    ///
    /// Returns a buffered asynchronous output adapter.
    ///
    /// # Panics
    ///
    /// Panics if initializing the default buffer requires `O::Item::default()`
    /// or `O::Item::clone()` and either operation panics, or if the default
    /// backing length exceeds `Vec`'s supported capacity.
    pub fn new(inner: O) -> Self {
        Self {
            output: AsyncBufferedOutput::new(inner),
            required_spare: 1,
        }
    }

    /// Creates an asynchronous encoder output with at least `capacity` units.
    ///
    /// # Parameters
    ///
    /// * `inner` - Unit output receiving transcoder data.
    /// * `capacity` - Requested internal unit capacity.
    ///
    /// # Returns
    ///
    /// Returns a buffered asynchronous output adapter.
    ///
    /// # Panics
    ///
    /// Panics if initializing the buffer requires `O::Item::default()` or
    /// `O::Item::clone()` and either operation panics, or if the requested
    /// backing length exceeds `Vec`'s supported capacity.
    pub fn with_capacity(inner: O, capacity: usize) -> Self {
        Self {
            output: AsyncBufferedOutput::with_capacity(inner, capacity),
            required_spare: 1,
        }
    }

    /// Tries to create an adapter with at least `capacity` buffered units.
    ///
    /// # Parameters
    ///
    /// * `inner` - Output moved into the adapter, including on allocation
    ///   failure.
    /// * `capacity` - Minimum number of units to reserve.
    ///
    /// # Returns
    ///
    /// Returns an empty buffered adapter on successful allocation.
    ///
    /// # Errors
    ///
    /// Returns an allocation error when the internal unit buffer cannot be
    /// allocated.
    ///
    /// # Panics
    ///
    /// Panics if initializing the buffer requires `O::Item::default()` or
    /// `O::Item::clone()` and either operation panics.
    pub fn try_with_capacity(inner: O, capacity: usize) -> std::result::Result<Self, TryReserveError> {
        Ok(Self {
            output: AsyncBufferedOutput::try_with_capacity(inner, capacity)?,
            required_spare: 1,
        })
    }

    /// Returns the total internal unit-buffer capacity.
    ///
    /// # Returns
    ///
    /// The allocated unit capacity, including occupied and spare slots.
    #[must_use]
    #[inline]
    pub fn capacity(&self) -> usize {
        self.output.capacity()
    }

    /// Returns a shared reference to the wrapped asynchronous output.
    ///
    /// # Returns
    ///
    /// Returns the wrapped output. Units can still be pending in this adapter.
    #[must_use]
    #[inline]
    pub const fn inner(&self) -> &O {
        self.output.inner()
    }

    /// Returns a mutable reference to the wrapped asynchronous output.
    ///
    /// Pending adapter units are not automatically delivered before direct
    /// output operations. Call [`Self::flush_async`] first when ordering with
    /// previously transcoded data matters.
    ///
    /// # Returns
    ///
    /// Returns the wrapped output.
    #[must_use]
    #[inline]
    pub fn inner_mut(&mut self) -> &mut O {
        self.output.inner_mut()
    }

    /// Returns the number of units retained pending delivery.
    ///
    /// # Returns
    ///
    /// Returns the length of the pending unit window.
    #[must_use]
    #[inline]
    pub const fn pending_len(&self) -> usize {
        self.output.pending_len()
    }

    /// Consumes this adapter without flushing the wrapped output.
    ///
    /// This method performs no asynchronous I/O. The returned buffer holds
    /// units that must be delivered before the logical stream can continue.
    ///
    /// Recovery assumes that wrapped I/O calls obey their normal return
    /// contracts. If an inner I/O panic is caught, the pending units may
    /// overlap a prefix already delivered. The buffer does not track this
    /// indeterminate delivery; reconcile external side effects before
    /// retrying or replaying those units.
    ///
    /// # Returns
    ///
    /// Returns the wrapped output and its pending unit buffer, assuming
    /// wrapped I/O calls obey their normal return contracts.
    #[must_use = "the returned output and pending buffer must be handled"]
    #[inline]
    pub fn into_parts(self) -> (O, Buffer<O::Item>) {
        self.output.into_parts()
    }
}

impl<O> AsyncTranscodeEncodeOutput<O>
where
    O: AsyncOutput + Unpin,
    O::Item: Clone + Default + Unpin,
{
    /// Flushes pending transcoded units and the wrapped asynchronous output.
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` after pending units are delivered and the sink is
    /// flushed.
    ///
    /// # Errors
    ///
    /// Returns output-delivery or flush errors from the wrapped output. A
    /// failed delivery retains every unit that was not accepted.
    pub async fn flush_async(&mut self) -> Result<()> {
        self.output.flush_async().await
    }

    /// Delivers pending transcoded units without flushing the wrapped output.
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` when the pending buffer is empty.
    ///
    /// # Errors
    ///
    /// Returns output-delivery errors from the wrapped output. A failed
    /// delivery retains every unit that was not accepted.
    pub async fn drain_async(&mut self) -> Result<()> {
        let capacity = self.output.capacity();
        self.output.ensure_spare_capacity_async(capacity).await
    }

    /// Runs encoder reset and buffers its stream-prefix units.
    ///
    /// # Type Parameters
    ///
    /// * `E` - Encoder producing the wrapped output item type.
    /// * `M` - Error mapper invoked for encoder failures.
    /// * `Value` - Logical input value type accepted by the encoder.
    ///
    /// # Parameters
    ///
    /// * `encoder` - Streaming encoder whose lifecycle is being started.
    /// * `map_error` - Maps transcoder errors into I/O errors.
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` with the reset prefix retained in the buffer.
    /// A successful reset discards the previous stream's spare-capacity hint.
    /// Previously committed pending units remain buffered in delivery order.
    ///
    /// # Errors
    ///
    /// Returns capacity planning, output delivery, allocation, or mapped reset
    /// errors. Pending units remain retained if asynchronous delivery fails.
    ///
    /// # Panics
    ///
    /// Panics when `encoder` writes more units than its declared reset bound.
    pub async fn reset_async<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
    where
        E: Transcoder<Input = Value, Output = O::Item>,
        M: FnMut(E::Error) -> Error,
    {
        let required = encoder.max_reset_output_len().map_err(capacity_error_to_invalid_data)?;
        self.ensure_spare_capacity_async(required).await?;
        let (units, output_index, available) = self.output.spare_raw_parts_mut();
        debug_assert!(available >= required);
        let written = encoder.reset(units, output_index).map_err(&mut *map_error)?;
        assert!(written <= required, "reset wrote beyond its bound");
        // SAFETY: `written` is bounded by the reserved spare range above.
        unsafe {
            self.output.advance(written);
        }
        self.required_spare = 1;
        Ok(())
    }

    /// Polls one cancellation-safe encoder operation.
    ///
    /// The adapter may first deliver previously pending output to make a
    /// conservative spare range available. Once it invokes `encoder`, it
    /// advances the produced units and returns immediately without another
    /// poll. The returned [`TranscodeProgress`] is therefore the exact source
    /// range committed by this call.
    ///
    /// # Type Parameters
    ///
    /// * `E` - Encoder producing the wrapped output item type.
    /// * `M` - Error mapper invoked for encoder failures.
    /// * `Value` - Logical input value type accepted by the encoder.
    ///
    /// # Parameters
    ///
    /// * `cx` - Task context used by the sink to register readiness.
    /// * `encoder` - Streaming encoder used for the conversion.
    /// * `map_error` - Maps transcoder errors into I/O errors.
    /// * `input` - Source values.
    /// * `input_index` - First source value to transcode.
    /// * `count` - Number of source values available from `input_index`.
    ///
    /// # Returns
    ///
    /// Returns `Pending` while making output space, without consuming input,
    /// or `Ready` with one validated encoder progress report or an error.
    ///
    /// # Errors
    ///
    /// Returns invalid input ranges, output delivery, allocation, contract, or
    /// mapped transcoder errors. Pending units remain retained after a failed
    /// asynchronous delivery. An encoder error does not provide general
    /// transactional rollback or complete progress: the encoder may already
    /// have advanced, but units written by its failing call are not committed
    /// to this adapter's pending output. Do not replay the original input
    /// range unless the concrete encoder provides an explicit recovery
    /// protocol. Previously committed pending units follow the wrapped output's
    /// recovery contract; reset starts a new logical stream and does not undo
    /// output already delivered.
    pub fn poll_transcode<E, M, Value>(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        encoder: &mut E,
        map_error: &mut M,
        input: &[Value],
        input_index: usize,
        count: usize,
    ) -> Poll<Result<TranscodeProgress>>
    where
        E: TranscodeEncoder<Input = Value, Output = O::Item>,
        M: FnMut(E::Error) -> Error,
    {
        let input_end = SliceRange::checked_range_end(
            input.len(),
            input_index,
            count,
            "encode input range exceeds source buffer",
        )?;
        if count == 0 {
            return Poll::Ready(Ok(TranscodeProgress::complete(0, 0)));
        }
        let input = &input[..input_end];
        let this = self.as_mut().get_mut();
        let per_value = match encoder.max_transcode_output_len(1) {
            Ok(required) => required.max(1),
            Err(error) => {
                return Poll::Ready(Err(capacity_error_to_invalid_data(error)));
            }
        };
        let required_spare = this.required_spare.max(per_value);
        if let Err(error) = this.output.try_reserve_capacity(required_spare) {
            return Poll::Ready(Err(allocation_error(error)));
        }
        match Pin::new(&mut this.output).poll_ensure_spare_capacity(cx, required_spare) {
            Poll::Ready(Ok(())) => {}
            Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
            Poll::Pending => return Poll::Pending,
        }
        let (units, output_index, available_output) = this.output.spare_raw_parts_mut();
        let progress = encoder
            .transcode(input, input_index, units, output_index)
            .map_err(&mut *map_error)
            .and_then(|progress| {
                validate_encode_progress(progress, input_index, count, output_index, available_output)
            });
        let progress = match progress {
            Ok(progress) => progress,
            Err(error) => return Poll::Ready(Err(error)),
        };
        // SAFETY: Progress validation proves the initialized output count fits
        // the spare window provided to the encoder.
        unsafe {
            this.output.advance(progress.written());
        }
        this.required_spare = if let TranscodeStatus::NeedOutput { required, .. } = progress.status() {
            required.get()
        } else {
            // `validate_encode_progress` rejects `NeedInput` before this
            // point.
            debug_assert!(progress.is_complete());
            1
        };
        Poll::Ready(Ok(progress))
    }

    /// Transcodes one cancellation-safe progress step into the retained buffer.
    ///
    /// This is the async wrapper for [`Self::poll_transcode`]. It returns after
    /// one encoder invocation; callers that need to consume a whole source
    /// range must advance the source index by [`TranscodeProgress::read`] and
    /// call it again.
    ///
    /// # Type Parameters
    ///
    /// * `E` - Encoder producing the wrapped output item type.
    /// * `M` - Error mapper invoked for encoder failures.
    /// * `Value` - Logical input value type accepted by the encoder.
    ///
    /// # Parameters
    ///
    /// * `encoder` - Encoder invoked once after output space is available.
    /// * `map_error` - Maps encoder errors to I/O errors.
    /// * `input` - Source values borrowed for this operation.
    /// * `input_index` - Starting source offset.
    /// * `count` - Number of source values available at that offset.
    ///
    /// # Returns
    ///
    /// One validated progress report describing committed source consumption.
    ///
    /// # Errors
    ///
    /// Returns range, capacity, allocation, delivery, contract or mapped
    /// encoder errors, with pending units retained as described by
    /// [`Self::poll_transcode`].
    pub async fn transcode_async<E, M, Value>(
        &mut self,
        encoder: &mut E,
        map_error: &mut M,
        input: &[Value],
        input_index: usize,
        count: usize,
    ) -> Result<TranscodeProgress>
    where
        E: TranscodeEncoder<Input = Value, Output = O::Item>,
        M: FnMut(E::Error) -> Error,
    {
        poll_fn(|cx| Pin::new(&mut *self).poll_transcode(cx, encoder, map_error, input, input_index, count)).await
    }

    /// Finishes the encoder and retains its final output for delivery.
    ///
    /// # Type Parameters
    ///
    /// * `E` - Encoder producing the wrapped output item type.
    /// * `M` - Error mapper invoked for encoder failures.
    /// * `Value` - Logical input value type accepted by the encoder.
    ///
    /// # Parameters
    ///
    /// * `encoder` - Streaming encoder whose lifecycle is being finished.
    /// * `map_error` - Maps transcoder errors into I/O errors.
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` after final units are committed to the buffer.
    ///
    /// # Errors
    ///
    /// Returns capacity planning, output delivery, allocation, or mapped finish
    /// errors. Call [`Self::flush_async`] after this method to deliver
    /// final units. Keeping finalization separate from delivery lets a
    /// caller record the completed lifecycle before any later suspension
    /// point.
    ///
    /// # Panics
    ///
    /// Panics when `encoder` writes more units than its declared finish bound.
    pub async fn finish_async<E, M, Value>(&mut self, encoder: &mut E, map_error: &mut M) -> Result<()>
    where
        E: Transcoder<Input = Value, Output = O::Item>,
        M: FnMut(E::Error) -> Error,
    {
        let required = encoder
            .max_finish_output_len()
            .map_err(capacity_error_to_invalid_data)?;
        self.ensure_spare_capacity_async(required).await?;
        let (units, output_index, available) = self.output.spare_raw_parts_mut();
        debug_assert!(available >= required);
        let written = encoder.finish(units, output_index).map_err(&mut *map_error)?;
        assert!(written <= required, "finish wrote beyond its bound");
        // SAFETY: `written` is bounded by the reserved spare range above.
        unsafe {
            self.output.advance(written);
        }
        Ok(())
    }

    /// Reserves enough total buffer capacity and makes `count` spare slots
    /// available for one transcoder operation. Pending units may be delivered
    /// before the codec operation, and an async poll can return `Pending` until
    /// the wrapped output accepts them.
    ///
    /// # Parameters
    ///
    /// * `count` - Minimum spare units needed after any pending delivery.
    ///
    /// # Returns
    ///
    /// `Ok(())` when the requested spare capacity is available.
    ///
    /// # Errors
    ///
    /// Returns allocation or wrapped output errors; undelivered units remain
    /// buffered.
    async fn ensure_spare_capacity_async(&mut self, count: usize) -> Result<()> {
        self.output.try_reserve_capacity(count).map_err(allocation_error)?;
        self.output.ensure_spare_capacity_async(count).await
    }
}

impl<O> fmt::Debug for AsyncTranscodeEncodeOutput<O>
where
    O: AsyncOutput,
    O::Item: Clone + Default,
    AsyncBufferedOutput<O>: fmt::Debug,
{
    /// Formats this asynchronous output adapter for debugging.
    ///
    /// # Parameters
    ///
    /// * `formatter` - Destination for the adapter and wrapped buffer
    ///   description.
    ///
    /// # Returns
    ///
    /// `Ok(())` after the description is written.
    ///
    /// # Errors
    ///
    /// Propagates errors reported by the formatter.
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("AsyncTranscodeEncodeOutput")
            .field("output", &self.output)
            .finish()
    }
}