//! Blob management for a contiguous journal.

use crate::{
    Context, SyncCompletion,
    journal::{
        Error,
        frame::{FrameReader, Limited, decode_item, decode_length_prefix},
    },
};
use bytes::Bytes;
use commonware_codec::{Buf, Codec, EncodeSize, Error as CodecError, OwnedBytes, ReadExt, Write};
use commonware_formatting::hex;
use commonware_runtime::{
    Blob as RBlob, Buf as _, Error as RError, Handle, IoBuf, IoBufMut, IoBufs, ReadOptions,
    buffer::paged::{CacheRef, Recovery as PagedRecovery, Replay as PagedReplay, Sealed, Writer},
    telemetry::metrics::{Counter, Gauge, GaugeExt as _, MetricsExt as _},
};
use commonware_utils::futures::try_join_all;
use futures::{FutureExt as _, future};
use std::{collections::BTreeMap, num::NonZeroUsize, sync::Arc};
use tracing::debug;

/// Metrics for a journal's blobs.
struct Metrics {
    tracked: Gauge,
    synced: Counter,
    pruned: Counter,
}

impl Metrics {
    fn new<E: Context>(context: &E) -> Self {
        Self {
            tracked: context.gauge("tracked", "Number of blobs held"),
            synced: context.counter("synced", "Number of blob sync requests"),
            pruned: context.counter("pruned", "Number of blobs pruned"),
        }
    }
}

/// A storage partition holding blobs.
pub(super) struct Partition<E: Context> {
    context: E,
    name: String,
    page_cache: CacheRef,
    write_buffer: NonZeroUsize,
}

impl<E: Context> Partition<E> {
    pub(super) const fn new(
        context: E,
        name: String,
        page_cache: CacheRef,
        write_buffer: NonZeroUsize,
    ) -> Self {
        Self {
            context,
            name,
            page_cache,
            write_buffer,
        }
    }

    /// Open the given blob as a [`Writer`], creating it if it does not exist.
    pub(super) async fn open(&self, blob: u64) -> Result<Writer<E::Blob>, Error> {
        let name = blob.to_be_bytes();
        let (blob, size) = self.context.open(&self.name, &name).await?;
        Ok(Writer::new(blob, size, self.write_buffer.get(), self.page_cache.clone()).await?)
    }

    /// Open a blob under exclusive initialization ownership.
    pub(super) async fn open_recovery(&self, index: u64) -> Result<PagedRecovery<E::Blob>, Error> {
        let (blob, size) = self.context.open(&self.name, &index.to_be_bytes()).await?;
        Ok(
            PagedRecovery::open(blob, size, self.write_buffer.get(), self.page_cache.clone())
                .await?,
        )
    }

    /// Scan a partition's blob names, treating a missing partition as empty.
    pub(super) async fn scan_names(context: &E, name: &str) -> Result<Vec<Vec<u8>>, Error> {
        match context.scan(name).await {
            Ok(names) => Ok(names),
            Err(RError::PartitionMissing(_)) => Ok(Vec::new()),
            Err(err) => Err(Error::Runtime(err)),
        }
    }

    /// Parse blob names into sorted indices.
    pub(super) fn indices(names: Vec<Vec<u8>>) -> Result<Vec<u64>, Error> {
        let mut indices = Vec::with_capacity(names.len());
        for name in names {
            let bytes: [u8; 8] = name
                .as_slice()
                .try_into()
                .map_err(|_| Error::InvalidBlobName(hex(&name)))?;
            indices.push(u64::from_be_bytes(bytes));
        }
        indices.sort_unstable();
        Ok(indices)
    }

    /// Open every blob in `names` as a [`PagedRecovery`], keyed by blob index.
    pub(super) async fn open_many(
        &self,
        names: Vec<Vec<u8>>,
    ) -> Result<BTreeMap<u64, PagedRecovery<E::Blob>>, Error> {
        let mut blobs = BTreeMap::new();
        for name in names {
            let hex_name = hex(&name);
            let bytes: [u8; 8] = name
                .try_into()
                .map_err(|_| Error::InvalidBlobName(hex_name.clone()))?;
            let index = u64::from_be_bytes(bytes);
            let writer = self.open_recovery(index).await?;
            debug!(index, blob = hex_name, "loaded blob");
            blobs.insert(index, writer);
        }
        Ok(blobs)
    }

    /// Scan the partition and open every existing blob as a [`PagedRecovery`], keyed by blob index.
    pub(super) async fn open_all(&self) -> Result<BTreeMap<u64, PagedRecovery<E::Blob>>, Error> {
        let names = Self::scan_names(&self.context, &self.name).await?;
        self.open_many(names).await
    }

    /// Open only blobs that may contain items below a cap, returning discarded blob indices.
    pub(super) async fn open_bounded(
        &self,
        max_size: u64,
        items_per_blob: u64,
    ) -> Result<(BTreeMap<u64, PagedRecovery<E::Blob>>, Vec<u64>), Error> {
        let names = Self::scan_names(&self.context, &self.name).await?;
        let indices = Self::indices(names)?;
        let mut pending = BTreeMap::new();
        let mut discarded = Vec::new();
        for index in indices {
            let first = super::blob_first_position(index, items_per_blob)?;
            if first >= max_size {
                discarded.push(index);
            } else {
                pending.insert(index, self.open_recovery(index).await?);
            }
        }
        Ok((pending, discarded))
    }

    /// Remove the given blob.
    pub(super) async fn remove(&self, blob: u64) -> Result<(), Error> {
        self.context
            .remove(&self.name, Some(&blob.to_be_bytes()))
            .await?;
        Ok(())
    }

    /// Remove an entire partition by name, treating "already missing" as success.
    pub(super) async fn remove_all(context: &E, name: &str) -> Result<(), Error> {
        match context.remove(name, None).await {
            Ok(()) | Err(RError::PartitionMissing(_)) => Ok(()),
            Err(err) => Err(Error::Runtime(err)),
        }
    }

    /// Select the blob partition for `prefix` using legacy-first compatibility rules: the
    /// legacy partition (`prefix` itself) if it contains data, otherwise `{prefix}-blobs`.
    /// Both containing data is corruption.
    ///
    /// Returns the chosen partition's name together with the blob names found in it while
    /// selecting, so the caller can open those blobs without listing the partition a second time.
    // TODO(#2941): Remove legacy partition support
    pub(super) async fn select(context: &E, prefix: &str) -> Result<(String, Vec<Vec<u8>>), Error> {
        let new_partition = format!("{prefix}-blobs");
        let legacy_blobs = Self::scan_names(context, prefix).await?;
        let new_blobs = Self::scan_names(context, &new_partition).await?;

        if !legacy_blobs.is_empty() && !new_blobs.is_empty() {
            return Err(Error::Corruption(format!(
                "both legacy and blobs partitions contain data: legacy={prefix} blobs={new_partition}"
            )));
        }

        if !legacy_blobs.is_empty() {
            Ok((prefix.into(), legacy_blobs))
        } else {
            Ok((new_partition, new_blobs))
        }
    }
}

/// A journal's blobs: contiguous sealed blobs ending in one writable tail.
pub(super) struct Writable<E: Context> {
    partition: Partition<E>,
    metrics: Metrics,

    /// The writable tail.
    tail: Writer<E::Blob>,

    /// Index of the first blob in [Self::sealed].
    oldest_blob_index: u64,

    /// Sealed historical blobs.
    sealed: Vec<Sealed<E::Blob>>,

    /// Cached owned snapshot of [Self::sealed].
    sealed_snapshot: Option<Arc<[Sealed<E::Blob>]>>,

    /// Sync of the tail's predecessor.
    tail_predecessor_sync: Option<SyncCompletion>,
}

impl<E: Context> Writable<E> {
    /// Build from recovered writers: seal every blob below `tail_blob` and install the tail,
    /// opening an empty one if absent.
    ///
    /// # Invariants
    ///
    /// - Retained blobs must be contiguous.
    /// - No blob may exceed `tail_blob`.
    /// - Any blobs present must end at `tail_blob`.
    pub(super) async fn recover(
        partition: Partition<E>,
        pending: BTreeMap<u64, PagedRecovery<E::Blob>>,
        tail_blob: u64,
    ) -> Result<Self, Error> {
        if let Some(&newest) = pending.keys().next_back()
            && newest > tail_blob
        {
            return Err(Error::Corruption(format!(
                "blobs > tail blob {tail_blob} exist (newest={newest})"
            )));
        }
        let oldest = pending.keys().next().copied();
        let mut sealed = Vec::with_capacity(pending.len());
        let mut syncs = Vec::with_capacity(pending.len());
        let mut tail: Option<Writer<E::Blob>> = None;
        let mut expected = oldest;
        for (blob, writer) in pending {
            let writer: Writer<_> = writer.into();
            if expected != Some(blob) {
                return Err(Error::Corruption(format!(
                    "retained blobs must be contiguous (expected {expected:?}, got {blob})"
                )));
            }
            expected = blob.checked_add(1);
            if blob == tail_blob {
                tail = Some(writer);
            } else {
                let (sealed_blob, sync) = writer.seal().await?;
                syncs.push(sync);
                sealed.push(sealed_blob);
            }
        }
        try_join_all(syncs).await?;
        let tail = match tail {
            Some(writer) => writer,
            None => partition.open(tail_blob).await?,
        };

        // The retained blobs must reach the tail: `oldest_blob_index = tail_blob - sealed.len()`.
        // A gap (scanned blobs that do not end at `tail_blob`) is corruption.
        let oldest_blob_index = tail_blob
            .checked_sub(sealed.len() as u64)
            .filter(|&idx| sealed.is_empty() || Some(idx) == oldest)
            .ok_or_else(|| {
                Error::Corruption(format!("retained blobs must end at tail blob {tail_blob}"))
            })?;

        let metrics = Metrics::new(&partition.context);
        let _ = metrics.tracked.try_set(sealed.len() + 1);
        Ok(Self {
            partition,
            metrics,
            oldest_blob_index,
            tail,
            sealed,
            sealed_snapshot: None,
            tail_predecessor_sync: None,
        })
    }

    /// Index of the oldest retained blob.
    pub(super) const fn oldest_blob_index(&self) -> u64 {
        self.oldest_blob_index
    }

    /// Index of the newest blob.
    pub(super) const fn tail_blob_index(&self) -> u64 {
        self.oldest_blob_index + self.sealed.len() as u64
    }

    /// Encode a value directly into the tail's write buffer if it fits, returning its logical
    /// offset. See [Writer::try_append_value].
    pub(super) fn try_append_value<T: EncodeSize + Write>(&mut self, value: &T) -> Option<u64> {
        self.tail.try_append_value(value)
    }

    /// Append `buf` to the tail's write buffer if it fits, returning its logical offset. See
    /// [Writer::try_append].
    pub(super) fn try_append(&mut self, buf: &[u8]) -> Option<u64> {
        self.tail.try_append(buf)
    }

    /// Append owned bytes to the tail, returning the logical offset of the first byte.
    pub(super) async fn append_owned(mut self, buf: IoBuf) -> Result<(Self, u64), Error> {
        let (tail, offset) = self.tail.append_owned(buf).await?;
        self.tail = tail;
        Ok((self, offset))
    }

    /// Borrow the current blobs for a live reader.
    pub(super) fn reader(&self) -> Blobs<'_, E::Blob> {
        Blobs {
            oldest_blob_index: self.oldest_blob_index,
            sealed: SealedBlobs::Borrowed(&self.sealed),
            tail: Blob::Writer(&self.tail),
        }
    }

    /// Capture owned blob handles for a snapshot reader.
    pub(super) async fn snapshot(mut self) -> Result<(Self, Blobs<'static, E::Blob>), Error> {
        // Reuse the immutable history shared by snapshots until its membership changes.
        let sealed = match &self.sealed_snapshot {
            Some(sealed) => sealed.clone(),
            None => {
                let sealed: Arc<[Sealed<E::Blob>]> = self.sealed.clone().into();
                self.sealed_snapshot = Some(sealed.clone());
                sealed
            }
        };

        // Freeze the live tail before combining it with the immutable history.
        let (tail, snapshot) = self.tail.snapshot().await?;
        self.tail = tail;
        let blobs = Blobs {
            oldest_blob_index: self.oldest_blob_index,
            sealed: SealedBlobs::Owned(sealed),
            tail: Blob::Sealed(snapshot),
        };
        Ok((self, blobs))
    }

    /// Seal the tail, start syncing it, and open the next blob as the new tail.
    pub(super) async fn seal_tail(mut self) -> Result<Self, Error> {
        // Observe both outstanding syncs before creating a successor blob.
        self.drain_tail_predecessor_sync().await?;
        self.tail = self.tail.wait_for_sync().await?;

        // Install a fresh tail and retain the sealed predecessor's durability completion.
        let next_blob = self
            .tail_blob_index()
            .checked_add(1)
            .ok_or(Error::OffsetOverflow)?;
        let new_writer = self.partition.open(next_blob).await?;
        let old_writer = std::mem::replace(&mut self.tail, new_writer);
        let (sealed, handle) = old_writer.seal().await?;
        self.metrics.tracked.inc();
        self.metrics.synced.inc();
        self.sealed.push(sealed);
        self.sealed_snapshot = None;
        assert!(self.tail_predecessor_sync.is_none());
        self.tail_predecessor_sync = Some(handle.boxed().shared());
        Ok(self)
    }

    /// Drop every blob below `min_blob` and remove its file, oldest-first. Safe with live readers:
    /// snapshot readers keep their own handles, which the runtime's read-after-remove contract keeps
    /// valid.
    ///
    /// # Invariants
    ///
    /// - `oldest_blob_index < min_blob <= tail_blob_index`
    pub(super) async fn prune(mut self, min_blob: u64) -> Result<Self, Error> {
        assert!(self.oldest_blob_index < min_blob && min_blob <= self.tail_blob_index());

        // Observe outstanding sync failures before removing any retained data.
        self.drain_tail_predecessor_sync().await?;
        self.tail = self.tail.wait_for_sync().await?;

        let drop_count = (min_blob - self.oldest_blob_index) as usize;
        let prev_oldest_blob_index = self.oldest_blob_index;
        self.sealed.drain(..drop_count);
        self.sealed_snapshot = None;
        self.oldest_blob_index = min_blob;

        for blob in prev_oldest_blob_index..min_blob {
            self.partition.remove(blob).await?;
            self.metrics.tracked.dec();
            self.metrics.pruned.inc();
        }
        Ok(self)
    }

    /// Remove every blob and start an empty journal with its tail at `tail_blob`.
    ///
    /// Safe with live readers, like [Self::prune]: snapshot readers keep their own handles, which
    /// the runtime's read-after-remove contract keeps valid.
    #[commonware_macros::stability(ALPHA)]
    pub(super) async fn clear(mut self, tail_blob: u64) -> Result<Self, Error> {
        // Observe outstanding sync failures before removing any retained data.
        self.drain_tail_predecessor_sync().await?;
        self.tail = self.tail.wait_for_sync().await?;

        for blob in self.oldest_blob_index..=self.tail_blob_index() {
            self.partition.remove(blob).await?;
        }
        let _ = self.metrics.tracked.try_set(0);
        self.tail = self.partition.open(tail_blob).await?;
        self.metrics.tracked.inc();
        self.oldest_blob_index = tail_blob;
        self.sealed.clear();
        self.sealed_snapshot = None;
        Ok(self)
    }

    /// Wait for the predecessor's durability completion.
    async fn drain_tail_predecessor_sync(&mut self) -> Result<(), Error> {
        if let Some(predecessor) = self.tail_predecessor_sync.take() {
            predecessor.await?;
        }
        Ok(())
    }

    /// Start syncing the tail, returning a handle that completes once both the tail and its
    /// predecessor are durable.
    pub(super) async fn start_sync(mut self) -> Result<(Self, Handle<()>), Error> {
        // Keep at most one tail sync in flight. A pending predecessor sync is not awaited here:
        // the returned handle joins it, so handles from consecutive calls can be pending at once.
        self.tail = self.tail.wait_for_sync().await?;

        // The tail writer retains its completion. The returned handle also joins the predecessor.
        let (tail, tail_sync) = self.tail.start_sync().await?;
        self.tail = tail;
        self.metrics.synced.inc();
        let predecessor = self.tail_predecessor_sync.clone();
        let handle = Handle::from_future(async move {
            if let Some(predecessor) = predecessor {
                let (predecessor, tail) = future::join(predecessor, tail_sync).await;
                predecessor?;
                tail
            } else {
                tail_sync.await
            }
        });
        Ok((self, handle))
    }

    /// Wait for every started sync to complete, surfacing a failure without changing any blob.
    pub(super) async fn wait_for_sync(mut self) -> Result<Self, Error> {
        self.drain_tail_predecessor_sync().await?;
        self.tail = self.tail.wait_for_sync().await?;
        Ok(self)
    }

    /// Remove every blob and the partition itself, after observing any started sync so a failed
    /// sync removes nothing.
    pub(super) async fn destroy(self) -> Result<(), Error> {
        let blobs = self.wait_for_sync().await?;
        let tail_blob = blobs.tail_blob_index();
        drop(blobs.tail);
        for blob in blobs.oldest_blob_index..=tail_blob {
            blobs.partition.remove(blob).await?;
        }
        Partition::remove_all(&blobs.partition.context, &blobs.partition.name).await
    }
}

/// Blob handles for a contiguous journal view.
///
/// Stores sealed history separately from the tail because the tail can use a different read path.
pub(super) struct Blobs<'a, B: RBlob> {
    /// Index of the first sealed blob.
    oldest_blob_index: u64,
    /// Sealed historical blobs.
    sealed: SealedBlobs<'a, B>,
    /// Tail blob.
    tail: Blob<'a, B>,
}

/// Storage for the sealed history slice.
enum SealedBlobs<'a, B: RBlob> {
    /// Borrowed sealed slice.
    Borrowed(&'a [Sealed<B>]),
    /// Owned sealed slice.
    Owned(Arc<[Sealed<B>]>),
}

impl<B: RBlob> SealedBlobs<'_, B> {
    fn as_slice(&self) -> &[Sealed<B>] {
        match self {
            Self::Borrowed(sealed) => sealed,
            Self::Owned(sealed) => sealed,
        }
    }
}

/// A read handle for one journal blob.
pub(super) enum Blob<'a, B: RBlob> {
    /// Writable tail. Point reads go through the writer's cache-aware logical view, and
    /// [`Self::replay_from`] chooses the replay path.
    Writer(&'a Writer<B>),
    /// Immutable historical blob.
    Sealed(Sealed<B>),
}

impl<B: RBlob> Clone for Blob<'_, B> {
    fn clone(&self) -> Self {
        match self {
            Self::Writer(writer) => Self::Writer(writer),
            Self::Sealed(sealed) => Self::Sealed(sealed.clone()),
        }
    }
}

impl<'a, B: RBlob> Blob<'a, B> {
    /// Return the blob's logical size.
    pub(super) fn size(&self) -> u64 {
        match self {
            Self::Writer(writer) => writer.size(),
            Self::Sealed(sealed) => sealed.size(),
        }
    }

    /// Read into `buf` if the data is already cached.
    pub(super) fn try_read_sync_into(&self, buf: &mut [u8], offset: u64) -> bool {
        match self {
            Self::Writer(writer) => writer.try_read_sync_into(buf, offset),
            Self::Sealed(sealed) => sealed.try_read_sync_into(buf, offset),
        }
    }

    /// Read exactly `len` bytes at `offset`.
    pub(super) async fn read_at(&self, offset: u64, len: usize) -> Result<IoBufs, Error> {
        match self {
            Self::Writer(writer) => Ok(writer.read_at(offset, len).await?),
            Self::Sealed(sealed) => Ok(sealed.read_at(offset, len).await?),
        }
    }

    /// Read up to `len` bytes at `offset`.
    pub(super) async fn read_up_to(
        &self,
        offset: u64,
        len: usize,
        bufs: impl Into<IoBufMut> + Send,
    ) -> Result<(IoBufMut, usize), Error> {
        match self {
            Self::Writer(writer) => Ok(writer.read_up_to(offset, len, bufs).await?),
            Self::Sealed(sealed) => Ok(sealed.read_up_to(offset, len, bufs).await?),
        }
    }

    /// Return a sequential replay handle starting at `offset`.
    ///
    /// Constructing the handle is cheap. Paged replay stores prefetch settings, and writer-view
    /// replay starts with an empty buffer. Read buffers are allocated later by `Replay::ensure`.
    ///
    /// A sealed blob, or a writable tail whose write buffer holds no unwritten full page (as
    /// after a sync), is replayed from the blob in batches with `read_options`. A tail still
    /// holding one is replayed through a live view of the writer, which reads through its page
    /// cache without flushing it.
    pub(super) fn replay_from(
        self,
        offset: u64,
        buffer_size: NonZeroUsize,
        read_options: ReadOptions,
    ) -> Result<Replay<'a, B>, Error> {
        let sealed = match self {
            Self::Writer(writer) => match writer.try_snapshot() {
                Some(snapshot) => snapshot,
                None => return Replay::view(Self::Writer(writer), offset, buffer_size),
            },
            Self::Sealed(sealed) => sealed,
        };
        let mut replay = sealed.replay(buffer_size, read_options)?;
        replay.seek_to(offset)?;
        Ok(Replay::paged(replay))
    }

    /// Read fixed-size items at sorted byte offsets into `buf`.
    pub(super) async fn read_many_into(
        &self,
        buf: &mut [u8],
        offsets: &[u64],
        item_size: NonZeroUsize,
    ) -> Result<usize, Error> {
        match self {
            Self::Writer(writer) => Ok(writer.read_many_into(buf, offsets, item_size).await?),
            Self::Sealed(sealed) => Ok(sealed.read_many_into(buf, offsets, item_size).await?),
        }
    }

    /// Like [`Self::read_many_into`], but each `(offset, len)` range has its own length.
    pub(super) async fn read_ranges_into(
        &self,
        buf: &mut [u8],
        ranges: &[(u64, usize)],
    ) -> Result<usize, Error> {
        match self {
            Self::Writer(writer) => Ok(writer.read_ranges_into(buf, ranges).await?),
            Self::Sealed(sealed) => Ok(sealed.read_ranges_into(buf, ranges).await?),
        }
    }

    /// Like [`Self::read_many_into`], but synchronous and cache-only. Returns the indices of
    /// items that require a blob read. Their slots in `buf` hold unspecified bytes.
    pub(super) fn try_read_many_sync_into(
        &self,
        buf: &mut [u8],
        offsets: &[u64],
        item_size: NonZeroUsize,
    ) -> Vec<usize> {
        match self {
            Self::Writer(writer) => writer.try_read_many_sync_into(buf, offsets, item_size),
            Self::Sealed(sealed) => sealed.try_read_many_sync_into(buf, offsets, item_size),
        }
    }

    /// Like [`Self::try_read_many_sync_into`], but for variable-length `(offset, len)` ranges.
    pub(super) fn try_read_ranges_sync_into(
        &self,
        buf: &mut [u8],
        ranges: &[(u64, usize)],
    ) -> Vec<usize> {
        match self {
            Self::Writer(writer) => writer.try_read_ranges_sync_into(buf, ranges),
            Self::Sealed(sealed) => sealed.try_read_ranges_sync_into(buf, ranges),
        }
    }
}

impl<B: RBlob> FrameReader for Blob<'_, B> {
    async fn read_up_to(
        &self,
        offset: u64,
        len: usize,
        bufs: impl Into<IoBufMut> + Send,
    ) -> Result<(IoBufMut, usize), Error> {
        Self::read_up_to(self, offset, len, bufs).await
    }

    async fn read_at(&self, offset: u64, len: usize) -> Result<IoBufs, Error> {
        Self::read_at(self, offset, len).await
    }
}

/// Sequential replay over either a sealed paged blob or a live writer view.
///
/// Decoding methods select a concrete backing buffer before reading fields so the
/// compiler can specialize length checks and field-copy loops. Only the concrete
/// buffers implement [`Buf`], keeping codec readers behind this dispatch.
pub(super) struct Replay<'a, B: RBlob> {
    inner: ReplayInner<'a, B>,
}

/// Backing strategy for sequential blob replay.
enum ReplayInner<'a, B: RBlob> {
    /// Paged replay over sealed data. CRC bytes are validated and skipped by the runtime buffer.
    Paged(PagedReplay<B>),
    /// Logical replay over a live writer or any other blob view.
    View(ViewReplay<'a, B>),
}

impl<'a, B: RBlob> Replay<'a, B> {
    /// Decode an item through its concrete backing buffer.
    pub(super) fn read<A: ReadExt>(&mut self) -> Result<A, CodecError> {
        match &mut self.inner {
            ReplayInner::Paged(replay) => A::read(replay),
            ReplayInner::View(replay) => A::read(replay),
        }
    }

    /// Decode a frame length through its concrete backing buffer.
    #[inline]
    pub(super) fn read_length(&mut self) -> Result<(usize, usize), Error> {
        match &mut self.inner {
            ReplayInner::Paged(replay) => decode_length_prefix(replay),
            ReplayInner::View(replay) => decode_length_prefix(replay),
        }
    }

    /// Decode a frame payload bounded to `len` bytes through its concrete backing buffer.
    pub(super) fn decode<V: Codec>(
        &mut self,
        len: usize,
        cfg: &V::Cfg,
        compressed: bool,
    ) -> Result<V, Error> {
        match &mut self.inner {
            ReplayInner::Paged(replay) => {
                decode_item::<V>(Limited::new(replay, len), cfg, compressed)
            }
            ReplayInner::View(replay) => decode_item::<V>(replay.take(len), cfg, compressed),
        }
    }

    /// Wrap a paged replay handle.
    const fn paged(replay: PagedReplay<B>) -> Self {
        Self {
            inner: ReplayInner::Paged(replay),
        }
    }

    /// Build a replay handle over a logical blob view.
    fn view(blob: Blob<'a, B>, offset: u64, buffer_size: NonZeroUsize) -> Result<Self, Error> {
        Ok(Self {
            inner: ReplayInner::View(ViewReplay::new(blob, offset, buffer_size)?),
        })
    }

    /// Return whether the underlying blob has reached EOF.
    ///
    /// The replay buffer may still hold bytes after this becomes true.
    pub(super) const fn is_exhausted(&self) -> bool {
        match &self.inner {
            ReplayInner::Paged(replay) => replay.is_exhausted(),
            ReplayInner::View(replay) => replay.is_exhausted(),
        }
    }

    /// Ensure at least `n` logical bytes are buffered unless EOF is reached first.
    pub(super) async fn ensure(&mut self, n: usize) -> Result<bool, Error> {
        match &mut self.inner {
            ReplayInner::Paged(replay) => Ok(replay.ensure(n).await?),
            ReplayInner::View(replay) => replay.ensure(n).await,
        }
    }
}

impl<B: RBlob> bytes::Buf for Replay<'_, B> {
    fn copy_to_bytes(&mut self, len: usize) -> Bytes {
        match &mut self.inner {
            ReplayInner::Paged(replay) => replay.copy_to_bytes(len),
            ReplayInner::View(replay) => replay.copy_to_bytes(len),
        }
    }

    fn remaining(&self) -> usize {
        match &self.inner {
            ReplayInner::Paged(replay) => replay.remaining(),
            ReplayInner::View(replay) => replay.remaining(),
        }
    }

    fn chunk(&self) -> &[u8] {
        match &self.inner {
            ReplayInner::Paged(replay) => replay.chunk(),
            ReplayInner::View(replay) => replay.chunk(),
        }
    }

    fn advance(&mut self, cnt: usize) {
        match &mut self.inner {
            ReplayInner::Paged(replay) => replay.advance(cnt),
            ReplayInner::View(replay) => replay.advance(cnt),
        }
    }
}

/// Sequential read buffer over a live journal blob view.
struct ViewReplay<'a, B: RBlob> {
    /// The source blob view.
    blob: Blob<'a, B>,
    /// Next logical offset to read from `blob`.
    offset: u64,
    /// Minimum read size when more bytes are needed.
    buffer_size: NonZeroUsize,
    /// Buffered logical bytes.
    buf: IoBufs,
    /// Whether `offset` has reached the source blob's logical size.
    exhausted: bool,
}

impl<'a, B: RBlob> ViewReplay<'a, B> {
    /// Create a view replay positioned at `offset`.
    fn new(blob: Blob<'a, B>, offset: u64, buffer_size: NonZeroUsize) -> Result<Self, Error> {
        if offset > blob.size() {
            return Err(Error::Runtime(RError::BlobInsufficientLength));
        }

        Ok(Self {
            blob,
            offset,
            buffer_size,
            buf: IoBufs::default(),
            exhausted: false,
        })
    }

    /// Return whether the source view has no more bytes to read.
    const fn is_exhausted(&self) -> bool {
        self.exhausted
    }

    /// Ensure at least `n` bytes are available through the [`bytes::Buf`] implementation.
    async fn ensure(&mut self, n: usize) -> Result<bool, Error> {
        while self.remaining() < n && !self.exhausted {
            let blob_size = self.blob.size();
            let remaining = blob_size.saturating_sub(self.offset);
            if remaining == 0 {
                self.exhausted = true;
                break;
            }

            // Read enough to satisfy the request, but keep ordinary prefetch bounded by the replay
            // budget and never ask past the current logical EOF.
            let needed = n.saturating_sub(self.remaining());
            let read_len = self
                .buffer_size
                .get()
                .max(needed)
                .min(usize::try_from(remaining).unwrap_or(usize::MAX));
            let (buf, read) = self
                .blob
                .read_up_to(self.offset, read_len, IoBufMut::with_capacity(read_len))
                .await?;
            self.offset = self
                .offset
                .checked_add(read as u64)
                .ok_or(Error::OffsetOverflow)?;
            self.buf.append(buf.freeze());
            if self.offset == blob_size {
                self.exhausted = true;
            }
        }

        Ok(self.remaining() >= n)
    }
}

impl<B: RBlob> Buf for ViewReplay<'_, B> {
    fn take_owned<T: OwnedBytes>(&mut self, len: usize) -> T {
        self.buf.take_owned(len)
    }
}

impl<B: RBlob> bytes::Buf for ViewReplay<'_, B> {
    fn copy_to_bytes(&mut self, len: usize) -> Bytes {
        self.buf.copy_to_bytes(len)
    }

    fn remaining(&self) -> usize {
        self.buf.remaining()
    }

    fn chunk(&self) -> &[u8] {
        self.buf.chunk()
    }

    fn advance(&mut self, cnt: usize) {
        self.buf.advance(cnt);
    }
}

impl<'a, B: RBlob> Blobs<'a, B> {
    /// Index of the newest blob (the tail).
    pub(super) fn tail_blob_index(&self) -> u64 {
        self.oldest_blob_index + self.sealed.as_slice().len() as u64
    }

    /// Resolve a blob, if retained.
    pub(super) fn get(&self, blob: u64) -> Option<Blob<'a, B>> {
        if blob == self.tail_blob_index() {
            return Some(self.tail.clone());
        }
        self.sealed
            .as_slice()
            .get(blob.checked_sub(self.oldest_blob_index)? as usize)
            .cloned()
            .map(Blob::Sealed)
    }
}

#[cfg(test)]
impl<E: Context> Writable<E> {
    pub(in crate::journal::contiguous) const fn has_tail_predecessor_sync(&self) -> bool {
        self.tail_predecessor_sync.is_some()
    }

    /// Open `blob` as an independent writer, outside this journal's tracking
    /// (simulates a crash-artifact blob).
    pub(crate) async fn open_blob(&self, blob: u64) -> Result<Writer<E::Blob>, Error> {
        self.partition.open(blob).await
    }

    /// Make one blob durable.
    pub(crate) async fn sync_blob(mut self, blob: u64) -> Result<Self, Error> {
        if blob == self.tail_blob_index() {
            self.tail = self.tail.sync().await?;
            return Ok(self);
        }
        if blob < self.oldest_blob_index || blob >= self.tail_blob_index() {
            return Ok(self);
        }
        self.drain_tail_predecessor_sync().await?;
        Ok(self)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{journal::frame::encode_frame_into, utils::codec::View};
    use commonware_runtime::{IoBufMut, Runner as _, Storage as _, deterministic};
    use commonware_utils::{NZU16, NZUsize};

    impl<E: crate::Context> Writable<E> {
        pub(in super::super) fn test_configuration(&self) -> (E, String, CacheRef, NonZeroUsize) {
            (
                self.partition.context.child("recovery_fixture"),
                self.partition.name.clone(),
                self.partition.page_cache.clone(),
                self.partition.write_buffer,
            )
        }
    }

    fn assert_insufficient_length(result: Result<(IoBufMut, usize), Error>) {
        assert!(matches!(
            result,
            Err(Error::Runtime(RError::BlobInsufficientLength))
        ));
    }

    #[test]
    fn test_paged_replay_decodes_bulk_fields_at_page_boundaries() {
        const PAGE_SIZE: std::num::NonZeroU16 = NZU16!(64);

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            // The first frame fills the first page, and the second frame's payload ends one
            // byte into the third page.
            let exact_page = [0x11; 63];
            let cross_page = [0x22; 64];
            let mut encoded = Vec::new();
            encode_frame_into(None, &exact_page, &mut encoded).unwrap();
            encode_frame_into(None, &cross_page, &mut encoded).unwrap();

            // Replay a sealed snapshot so payloads are copied out of the paged replay buffer.
            let cache = CacheRef::new(PAGE_SIZE, NZUsize!(3));
            let (blob, size) = context
                .open("paged-replay-bulk-fields", b"blob")
                .await
                .unwrap();
            let writer = Writer::new(blob, size, 128, cache).await.unwrap();
            let (writer, _) = writer.append(&encoded).await.unwrap();
            let (_, snapshot) = writer.snapshot().await.unwrap();
            let source = Blob::Sealed(snapshot);
            let mut replay = source
                .replay_from(0, NZUsize!(128), ReadOptions::default())
                .unwrap();

            // A payload ending exactly at a page boundary is copied from one page.
            assert!(replay.ensure(1).await.unwrap());
            let (len, _) = replay.read_length().unwrap();
            assert_eq!(len, exact_page.len());
            assert!(replay.ensure(len).await.unwrap());
            assert_eq!(
                replay.decode::<[u8; 63]>(len, &(), false).unwrap(),
                exact_page
            );

            // The next frame starts on the second page, and its payload crosses into the third.
            assert!(replay.ensure(1).await.unwrap());
            let (len, _) = replay.read_length().unwrap();
            assert_eq!(len, cross_page.len());
            assert!(replay.ensure(len).await.unwrap());
            assert_eq!(
                replay.decode::<[u8; 64]>(len, &(), false).unwrap(),
                cross_page
            );
            assert_eq!(replay.remaining(), 0);
        });
    }

    #[test]
    fn test_replay_preserves_byte_views_across_fills() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cache = CacheRef::new(NZU16!(64), NZUsize!(3));
            let (blob, size) = context.open("replay-views", b"blob").await.unwrap();
            let writer = Writer::new(blob, size, 128, cache.clone()).await.unwrap();
            let (writer, _) = writer.append(b"abcdefghijklmnopqrstuvwx").await.unwrap();
            let (writer, snapshot) = writer.snapshot().await.unwrap();

            // A writer holding an unwritten full page is replayed through a live view. Its
            // page of filler precedes the same bytes, which the view reads from the tip.
            let (blob, size) = context.open("replay-views", b"pending").await.unwrap();
            let pending = Writer::new(blob, size, 128, cache).await.unwrap();
            let (pending, _) = pending.append(&[0; 40]).await.unwrap();
            let (pending, _) = pending.append(b"abcdefghijklmnopqrstuvwx").await.unwrap();
            assert!(pending.try_snapshot().is_none());

            for (source, offset, paged) in [
                (Blob::Writer(&pending), 40, false),
                (Blob::Writer(&writer), 0, true),
                (Blob::Sealed(snapshot), 0, true),
            ] {
                let mut replay = source
                    .replay_from(offset, NZUsize!(12), ReadOptions::default())
                    .unwrap();
                assert!(replay.ensure(12).await.unwrap());
                let first_range = replay.chunk().as_ptr_range();
                let first = replay.read::<View>().unwrap().bytes;
                assert!(first_range.contains(&first.as_ptr()));

                assert!(replay.ensure(16).await.unwrap());
                let middle = replay.read::<View>().unwrap().bytes;
                assert_eq!(first_range.contains(&middle.as_ptr()), paged);

                let last_range = replay.chunk().as_ptr_range();
                let last = replay.copy_to_bytes(8);
                assert!(last_range.contains(&last.as_ptr()));
                assert_eq!(replay.remaining(), 0);
                assert!(!replay.ensure(1).await.unwrap());
                assert!(replay.is_exhausted());
                drop(replay);
                assert_eq!(first.as_ref(), b"abcdefgh");
                assert_eq!(middle.as_ref(), b"ijklmnop");
                assert_eq!(last.as_ref(), b"qrstuvwx");
            }
        });
    }

    /// A writer-view replay decodes an `IoBuf` item as a view of its read buffer when the frame
    /// lies within one read, and as one native copy when it spans reads; either way the item ends
    /// up as the sole owner of native storage.
    #[test]
    fn test_view_replay_decodes_iobuf_views() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let mut encoded = Vec::new();
            encode_frame_into(None, &IoBuf::from(b"abcdefgh"), &mut encoded).unwrap();
            let cache = CacheRef::new(NZU16!(64), NZUsize!(3));
            let (blob, size) = context.open("replay-views", &[0]).await.unwrap();
            let pending = Writer::new(blob, size, 256, cache).await.unwrap();
            let (pending, _) = pending.append(&[0; 64]).await.unwrap();
            let (pending, _) = pending.append(&encoded).await.unwrap();
            assert!(pending.try_snapshot().is_none());

            for (buffer_size, contiguous) in [(64, true), (4, false)] {
                let mut replay = Blob::Writer(&pending)
                    .replay_from(64, NZUsize!(buffer_size), ReadOptions::default())
                    .unwrap();
                assert!(replay.ensure(1).await.unwrap());
                let (len, _) = replay.read_length().unwrap();
                assert!(replay.ensure(len).await.unwrap());
                assert_eq!(replay.chunk().len() >= len, contiguous);
                let range = replay.chunk().as_ptr_range();
                let item = replay.decode::<IoBuf>(len, &(..).into(), false).unwrap();
                assert_eq!(item, b"abcdefgh");
                assert_eq!(range.contains(&item.as_ptr()), contiguous);
                drop(replay);
                assert!(item.try_into_mut().is_ok());
            }
        });
    }

    #[test]
    fn test_replay_decodes_bounded_frames() {
        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let items = [
                Bytes::from_static(b"abcdefgh"),
                Bytes::from_static(b"ijklmnop"),
                Bytes::from_static(b"qrstuvwx"),
            ];
            for compression in [None, Some(1)] {
                let mut encoded = Vec::new();
                for item in &items {
                    encode_frame_into(compression, item, &mut encoded).unwrap();
                }
                let cache = CacheRef::new(NZU16!(64), NZUsize!(3));
                let (blob, size) = context
                    .open("replay-frames", &[compression.unwrap_or(0)])
                    .await
                    .unwrap();
                let writer = Writer::new(blob, size, 128, cache.clone()).await.unwrap();
                let (writer, _) = writer.append(&encoded).await.unwrap();
                let (writer, snapshot) = writer.snapshot().await.unwrap();

                // A writer holding an unwritten full page is replayed through a live view. Its
                // page of filler precedes the same frames.
                let (blob, size) = context
                    .open("replay-frames-pending", &[compression.unwrap_or(0)])
                    .await
                    .unwrap();
                let pending = Writer::new(blob, size, 256, cache).await.unwrap();
                let (pending, _) = pending.append(&[0; 64]).await.unwrap();
                let (pending, _) = pending.append(&encoded).await.unwrap();
                assert!(pending.try_snapshot().is_none());

                for (source, offset) in [
                    (Blob::Writer(&pending), 64),
                    (Blob::Writer(&writer), 0),
                    (Blob::Sealed(snapshot), 0),
                ] {
                    let mut replay = source
                        .clone()
                        .replay_from(offset, NZUsize!(12), ReadOptions::default())
                        .unwrap();
                    let mut retained = Vec::new();
                    for _ in &items {
                        assert!(replay.ensure(1).await.unwrap());
                        let (len, prefix_len) = replay.read_length().unwrap();
                        assert_eq!(prefix_len, 1);
                        assert!(replay.ensure(len).await.unwrap());
                        let contiguous = replay.chunk().len() >= len;
                        let range = replay.chunk().as_ptr_range();
                        let item = replay
                            .decode::<Bytes>(len, &(8..=8).into(), compression.is_some())
                            .unwrap();
                        if compression.is_none() && contiguous {
                            assert!(range.contains(&item.as_ptr()));
                        }
                        retained.push(item);
                    }
                    assert!(!replay.ensure(1).await.unwrap());
                    assert_eq!(replay.remaining(), 0);
                    drop(replay);
                    assert_eq!(retained, items);

                    if compression.is_none() {
                        for case in 0..3 {
                            let mut replay = source
                                .clone()
                                .replay_from(offset, NZUsize!(12), ReadOptions::default())
                                .unwrap();
                            assert!(replay.ensure(12).await.unwrap());
                            let (len, _) = replay.read_length().unwrap();
                            match case {
                                0 => assert!(matches!(
                                    replay.decode::<Bytes>(len, &(..=7).into(), false),
                                    Err(Error::Codec(CodecError::InvalidLength(8)))
                                )),
                                1 => {
                                    let remaining = replay.remaining();
                                    assert!(matches!(
                                        replay.decode::<u64>(1, &(), false),
                                        Err(Error::Codec(CodecError::EndOfBuffer))
                                    ));
                                    assert_eq!(replay.remaining(), remaining);
                                }
                                _ => assert!(matches!(
                                    replay.decode::<u8>(len, &(), false),
                                    Err(Error::Codec(CodecError::ExtraData(8)))
                                )),
                            }
                        }
                    }
                }
            }
        });
    }

    #[test]
    fn test_read_up_to_eof_parity() {
        const PAGE_SIZE: std::num::NonZeroU16 = NZU16!(64);

        let executor = deterministic::Runner::default();
        executor.start(|context| async move {
            let cache_ref = CacheRef::new(PAGE_SIZE, NZUsize!(3));
            let (blob, size) = context
                .open("read_up_to_eof_parity", b"blob")
                .await
                .unwrap();
            let writer = Writer::new(blob, size, 128, cache_ref).await.unwrap();
            let (writer, _) = writer.append(b"abc").await.unwrap();

            let size = writer.size();
            let tail = Blob::Writer(&writer);
            assert_insufficient_length(tail.read_up_to(size, 1, IoBufMut::with_capacity(1)).await);
            assert_eq!(
                tail.read_up_to(size, 0, IoBufMut::with_capacity(0))
                    .await
                    .unwrap()
                    .1,
                0
            );

            let (writer, snapshot) = writer.snapshot().await.unwrap();
            let snapshot_blob = Blob::Sealed(snapshot);
            assert_insufficient_length(
                snapshot_blob
                    .read_up_to(size, 1, IoBufMut::with_capacity(1))
                    .await,
            );
            assert_eq!(
                snapshot_blob
                    .read_up_to(size, 0, IoBufMut::with_capacity(0))
                    .await
                    .unwrap()
                    .1,
                0
            );

            let (sealed, sync) = writer.seal().await.unwrap();
            sync.await.unwrap();
            let sealed_blob = Blob::Sealed(sealed);
            assert_insufficient_length(
                sealed_blob
                    .read_up_to(size, 1, IoBufMut::with_capacity(1))
                    .await,
            );
            assert_eq!(
                sealed_blob
                    .read_up_to(size, 0, IoBufMut::with_capacity(0))
                    .await
                    .unwrap()
                    .1,
                0
            );
        });
    }
}