//! Blob management for a contiguous journal.

use crate::{
    Context, SyncCompletion,
    journal::{Error, frame::FrameReader},
};
use commonware_formatting::hex;
use commonware_runtime::{
    Blob as RBlob, Buf, Error as RError, Handle, IoBufMut, IoBufs, ReadOptions,
    buffer::paged::{CacheRef, Replay as PagedReplay, Sealed, Writer},
    telemetry::metrics::{Counter, Gauge, GaugeExt as _, MetricsExt as _},
};
use futures::{
    FutureExt as _,
    future::{self, try_join_all},
};
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?)
    }

    /// Scan a partition's blob names, treating a missing partition as empty.
    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)),
        }
    }

    /// Open every blob in `names` as a [`Writer`], keyed by blob index.
    pub(super) async fn open_many(
        &self,
        names: Vec<Vec<u8>>,
    ) -> Result<BTreeMap<u64, Writer<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(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 [`Writer`], keyed by blob index.
    pub(super) async fn open_all(&self) -> Result<BTreeMap<u64, Writer<E::Blob>>, Error> {
        let names = Self::scan_names(&self.context, &self.name).await?;
        self.open_many(names).await
    }

    /// 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>,

    /// Sync of the live tail. Kept on failure so later operations keep failing.
    tail_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, Writer<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 {
            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,
            tail_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
    }

    /// A write handle for the tail.
    pub(super) const fn tail_writer(&mut self) -> &mut Writer<E::Blob> {
        &mut self.tail
    }

    /// 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<Blobs<'static, E::Blob>, Error> {
        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
            }
        };
        Ok(Blobs {
            oldest_blob_index: self.oldest_blob_index,
            sealed: SealedBlobs::Owned(sealed),
            tail: Blob::Sealed(self.tail.snapshot().await?),
        })
    }

    /// Seal the tail, start syncing it, and open the next blob as the new tail.
    pub(super) async fn seal_tail(&mut self) -> Result<(), Error> {
        self.drain_tail_predecessor_sync().await?;
        // seal() waits only for syncs the writer started: a commit whose flush failed before its
        // sync began is retained solely in the tail sync slot, so it must be drained here too.
        self.drain_tail_sync().await?;

        // Open the next tail first so a failure leaves the current tail untouched.
        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;
        debug_assert!(self.tail_predecessor_sync.is_none());
        debug_assert!(self.tail_sync.is_none());
        self.tail_predecessor_sync = Some(handle.boxed().shared());
        Ok(())
    }

    /// 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<(), Error> {
        assert!(self.oldest_blob_index < min_blob && min_blob <= self.tail_blob_index());
        self.drain_tail_predecessor_sync().await?;
        self.drain_tail_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(())
    }

    /// Rewind the tail to `byte_offset`, shrinking it in place.
    ///
    /// # Invariants
    ///
    /// - `byte_offset <= tail size`
    ///
    pub(super) async fn rewind_tail(&mut self, byte_offset: u64) -> Result<(), Error> {
        let current_bytes = self.tail.size();
        assert!(byte_offset <= current_bytes);
        if byte_offset < current_bytes {
            self.drain_tail_predecessor_sync().await?;
            self.drain_tail_sync().await?;
            self.tail.resize(byte_offset).await?;
        }
        Ok(())
    }

    /// Rewind into a sealed blob: demote it to the writable tail, rewinding to `byte_offset`,
    /// and discarding every newer blob.
    ///
    /// # Invariants
    ///
    /// - `blob < tail_blob_index`
    pub(super) async fn rewind_into_sealed(
        &mut self,
        blob: u64,
        byte_offset: u64,
    ) -> Result<(), Error> {
        self.drain_tail_predecessor_sync().await?;
        self.drain_tail_sync().await?;

        let idx = blob
            .checked_sub(self.oldest_blob_index)
            .map(|idx| idx as usize)
            .filter(|&idx| idx < self.sealed.len())
            .ok_or_else(|| Error::Corruption(format!("rewind target blob {blob} not retained")))?;

        // Reopen the target as the writable tail and truncate in place. The fresh Writer
        // gets a fresh page-cache id, so pages cached under the sealed handle's id are
        // unreachable.
        let mut new_writer = self.partition.open(blob).await?;
        let current_bytes = new_writer.size();
        if byte_offset < current_bytes {
            new_writer.resize(byte_offset).await?;
        }

        // Remove blobs newest-first so a crash leaves a contiguous prefix: the old tail, then
        // sealed blobs down to the target. Capture the old tail before truncating `sealed`
        // (which redefines `tail_blob`).
        let old_tail_blob = self.tail_blob_index();
        self.tail = new_writer;
        self.partition.remove(old_tail_blob).await?;
        self.metrics.tracked.dec();
        for newer in ((blob + 1)..old_tail_blob).rev() {
            self.partition.remove(newer).await?;
            self.metrics.tracked.dec();
        }

        // Sealed history now ends below the target, which is the tail.
        self.sealed.truncate(idx);
        self.sealed_snapshot = None;
        Ok(())
    }

    /// 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.
    pub(super) async fn clear(&mut self, tail_blob: u64) -> Result<(), Error> {
        self.drain_tail_predecessor_sync().await?;
        self.drain_tail_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(())
    }

    /// Drain predecessor sync. Clear only on success.
    async fn drain_tail_predecessor_sync(&mut self) -> Result<(), Error> {
        let Some(predecessor) = self.tail_predecessor_sync.clone() else {
            return Ok(());
        };
        predecessor.await?;
        self.tail_predecessor_sync = None;
        Ok(())
    }

    /// Drain tail sync. Clear only on success.
    async fn drain_tail_sync(&mut self) -> Result<(), Error> {
        let Some(tail) = self.tail_sync.clone() else {
            return Ok(());
        };
        tail.await?;
        self.tail_sync = None;
        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) -> Handle<()> {
        // 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.
        if let Some(prior) = self.tail_sync.clone()
            && let Err(err) = prior.await
        {
            return Handle::ready(Err(err));
        }
        let tail = self.tail.start_sync().await.boxed().shared();
        self.metrics.synced.inc();
        self.tail_sync = Some(tail.clone());
        let predecessor = self.tail_predecessor_sync.clone();
        Handle::from_future(async move {
            if let Some(predecessor) = predecessor {
                let (predecessor, tail) = future::join(predecessor, tail).await;
                predecessor?;
                tail
            } else {
                tail.await
            }
        })
    }

    /// Remove every blob and the partition itself.
    pub(super) async fn destroy(mut self) -> Result<(), Error> {
        self.drain_tail_predecessor_sync().await?;
        self.drain_tail_sync().await?;

        let tail_blob = self.tail_blob_index();
        drop(self.tail);
        for blob in self.oldest_blob_index..=tail_blob {
            self.partition.remove(blob).await?;
        }
        Partition::remove_all(&self.partition.context, &self.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, read through the writer's cache-aware logical view.
    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`.
    ///
    /// Sealed blobs can use paged replay directly because their bytes are already fixed. The
    /// writable tail is replayed through a live view so replay observes logical bytes without
    /// mutating or flushing the writer.
    pub(super) fn replay_from(
        self,
        offset: u64,
        buffer_size: NonZeroUsize,
        read_options: ReadOptions,
    ) -> Result<Replay<'a, B>, Error> {
        match self {
            Self::Writer(writer) => Replay::view(Self::Writer(writer), offset, buffer_size),
            Self::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 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.
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> {
    /// 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> Buf for Replay<'_, B> {
    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: Vec<u8>,
    /// Offset of the next unread byte in `buf`.
    cursor: usize,
    /// 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: Vec::new(),
            cursor: 0,
            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 [`Buf`] implementation.
    async fn ensure(&mut self, n: usize) -> Result<bool, Error> {
        while self.remaining() < n && !self.exhausted {
            self.compact();

            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.extend_from_slice(&buf.chunk()[..read]);
            if self.offset == blob_size {
                self.exhausted = true;
            }
        }

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

    /// Discard bytes already consumed through [`Buf::advance`].
    fn compact(&mut self) {
        match self.cursor {
            0 => {}
            cursor if cursor == self.buf.len() => {
                self.buf.clear();
                self.cursor = 0;
            }
            cursor => {
                self.buf.drain(..cursor);
                self.cursor = 0;
            }
        }
    }
}

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

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

    fn advance(&mut self, cnt: usize) {
        self.cursor = self
            .cursor
            .checked_add(cnt)
            .expect("advance overflowed replay cursor");
        assert!(self.cursor <= self.buf.len(), "advanced past replay buffer");
    }
}

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)]
mod tests {
    use super::*;
    use commonware_runtime::{IoBufMut, Runner as _, Storage as _, deterministic};
    use commonware_utils::{NZU16, NZUsize};

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

    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<(), Error> {
            if blob == self.tail_blob_index() {
                self.tail.sync().await?;
                return Ok(());
            }
            if blob < self.oldest_blob_index || blob >= self.tail_blob_index() {
                return Ok(());
            }
            self.partition.open(blob).await?.sync().await?;
            Ok(())
        }
    }

    #[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::from_pooler(&context, PAGE_SIZE, NZUsize!(3));
            let (blob, size) = context
                .open("read_up_to_eof_parity", b"blob")
                .await
                .unwrap();
            let mut writer = Writer::new(blob, size, 128, cache_ref).await.unwrap();
            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 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
            );
        });
    }
}