lanspread/crates/lanspread-peer/src/stream_install.rs

use std::{
    collections::HashSet,
    fs::File,
    future::Future,
    path::{Path, PathBuf},
    pin::Pin,
    process::Stdio,
    sync::Arc,
    time::{Duration, Instant},
};

use bytes::Bytes;
use crc32fast::Hasher;
use futures::{SinkExt, StreamExt};
use lanspread_db::content_manifest::{
    Blake3Digest,
    CanonicalCatalogPath,
    CatalogContentManifest,
    CatalogEntryKind,
    ContentId,
    MAX_CATALOG_ENTRIES,
    MAX_CATALOG_TOTAL_BYTES,
};
use lanspread_proto::{
    ControlMessage,
    MAX_STREAM_INSTALL_FRAME_BYTES,
    Message,
    PeerEndpoint,
    Request,
    StreamInstallFrame,
};
use s2n_quic::{
    application,
    stream::{ReceiveStream, SendStream},
};
use tokio::{
    io::{AsyncRead, AsyncReadExt},
    process::Command,
    sync::{mpsc, mpsc::UnboundedSender},
    time::{self, Instant as TokioInstant, MissedTickBehavior},
};
use tokio_util::{
    codec::{FramedRead, FramedWrite, LengthDelimitedCodec},
    sync::CancellationToken,
};

use crate::{
    DownloadProgress,
    PeerEvent,
    install::root_eti_archives,
    network::connect_to_peer,
    path_validation::validate_game_file_path,
    quic_runtime::QuicConnector,
    scoped_blocking::scoped_blocking,
    scoped_process::{ReapedTokioChild, ScopedProcess},
    transfer_status::DownloadAttemptReporter,
};

const FRAME_CHANNEL_DEPTH: usize = 16;
const STREAM_INSTALL_PROGRESS_UPDATE_INTERVAL: Duration = Duration::from_millis(500);
const STREAM_CHUNK_SIZE: usize = 256 * 1024;
const UNRAR_LISTING_CAPTURE_LIMIT: usize = 64 * 1024 * 1024;
const STREAM_INSTALL_INACTIVITY_TIMEOUT: Duration = Duration::from_mins(10);

#[derive(Clone, Copy, Debug)]
struct StreamInstallInactivityDeadline {
    expires_at: TokioInstant,
    timeout: Duration,
}

impl StreamInstallInactivityDeadline {
    fn ordinary() -> Self {
        Self::after(STREAM_INSTALL_INACTIVITY_TIMEOUT)
    }

    fn after(timeout: Duration) -> Self {
        Self {
            expires_at: TokioInstant::now() + timeout,
            timeout,
        }
    }

    fn reset_after_frame(&mut self) {
        self.expires_at = TokioInstant::now() + self.timeout;
    }

    fn timeout_error(self, game_id: &str, context: &str) -> StreamInstallReceiveError {
        let timeout = self.timeout;
        StreamInstallReceiveError::transport(eyre::eyre!(
            "streamed install for {game_id} made no frame progress for {timeout:?} {context}"
        ))
    }

    async fn run<T>(
        self,
        operation: impl Future<Output = T>,
        game_id: &str,
        cancel_token: &CancellationToken,
        context: &str,
    ) -> StreamInstallReceiveResult<T> {
        tokio::select! {
            biased;
            () = cancel_token.cancelled() => {
                Err(StreamInstallReceiveError::cancelled(game_id, context))
            }
            () = time::sleep_until(self.expires_at) => {
                Err(self.timeout_error(game_id, context))
            }
            result = operation => Ok(result),
        }
    }
}

/// Integrity metadata advertised by the sender's RAR archive.
///
/// This remains an early corruption check only. The catalog-owned BLAKE3 digest
/// in [`IncomingFile`] is the trusted-content boundary.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct SenderArchiveIntegrity {
    expected_size: u64,
    expected_crc32: u32,
}

impl SenderArchiveIntegrity {
    fn new(expected_size: u64, expected_crc32: u32) -> Self {
        Self {
            expected_size,
            expected_crc32,
        }
    }

    fn verify(
        self,
        relative_path: &CanonicalCatalogPath,
        received: u64,
        actual_crc32: u32,
    ) -> eyre::Result<()> {
        if received != self.expected_size {
            eyre::bail!(
                "streamed file {relative_path} size mismatch: got {received}, expected {}",
                self.expected_size
            );
        }

        if actual_crc32 != self.expected_crc32 {
            eyre::bail!(
                "streamed file {relative_path} sender RAR CRC32 mismatch: got {actual_crc32:08X}, expected {:08X}",
                self.expected_crc32
            );
        }

        Ok(())
    }
}

pub type StreamInstallFuture<'a> = Pin<Box<dyn Future<Output = eyre::Result<()>> + Send + 'a>>;

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub(crate) enum StreamInstallReceiveErrorKind {
    Integrity,
    Transport,
    Cancelled,
    Setup,
}

#[derive(Debug)]
pub(crate) struct StreamInstallReceiveError {
    kind: StreamInstallReceiveErrorKind,
    report: eyre::Report,
}

impl StreamInstallReceiveError {
    #[must_use]
    pub(crate) const fn kind(&self) -> StreamInstallReceiveErrorKind {
        self.kind
    }

    fn integrity(error: impl Into<eyre::Report>) -> Self {
        Self::new(StreamInstallReceiveErrorKind::Integrity, error)
    }

    fn transport(error: impl Into<eyre::Report>) -> Self {
        Self::new(StreamInstallReceiveErrorKind::Transport, error)
    }

    fn setup(error: impl Into<eyre::Report>) -> Self {
        Self::new(StreamInstallReceiveErrorKind::Setup, error)
    }

    fn cancelled(game_id: &str, context: &str) -> Self {
        Self::new(
            StreamInstallReceiveErrorKind::Cancelled,
            eyre::eyre!("streamed install for {game_id} was cancelled {context}"),
        )
    }

    fn transport_or_cancelled(
        error: impl Into<eyre::Report>,
        game_id: &str,
        cancel_token: &CancellationToken,
        context: &str,
    ) -> Self {
        if cancel_token.is_cancelled() {
            Self::cancelled(game_id, context)
        } else {
            Self::transport(error)
        }
    }

    fn new(kind: StreamInstallReceiveErrorKind, error: impl Into<eyre::Report>) -> Self {
        Self {
            kind,
            report: error.into(),
        }
    }
}

impl std::fmt::Display for StreamInstallReceiveError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        std::fmt::Display::fmt(&self.report, formatter)
    }
}

impl std::error::Error for StreamInstallReceiveError {}

type StreamInstallReceiveResult<T> = Result<T, StreamInstallReceiveError>;

#[derive(Clone)]
pub struct StreamInstallFrameSink {
    frames: mpsc::Sender<StreamInstallFrame>,
    cancel_token: CancellationToken,
}

impl StreamInstallFrameSink {
    fn new(frames: mpsc::Sender<StreamInstallFrame>, cancel_token: CancellationToken) -> Self {
        Self {
            frames,
            cancel_token,
        }
    }

    pub async fn send(&self, frame: StreamInstallFrame) -> eyre::Result<()> {
        tokio::select! {
            biased;
            () = self.cancel_token.cancelled() => {
                eyre::bail!("streamed install frame send was cancelled");
            }
            result = self.frames.send(frame) => {
                result.map_err(|_| eyre::eyre!("streamed install frame receiver closed"))
            }
        }
    }
}

pub trait StreamInstallProvider: Send + Sync {
    /// Streams one archive and stops all archive work before returning.
    ///
    /// Implementations must observe `cancel_token` promptly. In particular,
    /// cancellation must terminate and reap any child process before this
    /// future resolves so directory-switch draining has a strict quiescence
    /// boundary.
    fn stream_archive<'a>(
        &'a self,
        archive: &'a Path,
        frames: StreamInstallFrameSink,
        cancel_token: CancellationToken,
    ) -> StreamInstallFuture<'a>;
}

#[derive(Debug, Default)]
pub struct NoopStreamInstallProvider;

impl StreamInstallProvider for NoopStreamInstallProvider {
    fn stream_archive<'a>(
        &'a self,
        archive: &'a Path,
        _frames: StreamInstallFrameSink,
        _cancel_token: CancellationToken,
    ) -> StreamInstallFuture<'a> {
        Box::pin(async move {
            eyre::bail!(
                "streamed install provider is not configured for {}",
                archive.display()
            )
        })
    }
}

#[derive(Debug)]
pub struct ExternalUnrarStreamProvider {
    program: PathBuf,
}

impl ExternalUnrarStreamProvider {
    #[must_use]
    pub fn new(program: PathBuf) -> Self {
        Self { program }
    }
}

impl StreamInstallProvider for ExternalUnrarStreamProvider {
    fn stream_archive<'a>(
        &'a self,
        archive: &'a Path,
        frames: StreamInstallFrameSink,
        cancel_token: CancellationToken,
    ) -> StreamInstallFuture<'a> {
        Box::pin(async move {
            let archive_name = archive_catalog_name(archive)?;
            let listing = unrar_listing(&self.program, archive, &cancel_token).await?;

            frames
                .send(StreamInstallFrame::ArchiveBegin {
                    archive_name: archive_name.clone(),
                    solid: listing.solid,
                    unpacked_size: listing.unpacked_size(),
                })
                .await?;

            stream_unrar_entries(
                &self.program,
                archive,
                &listing.entries,
                &frames,
                cancel_token.clone(),
            )
            .await?;

            frames
                .send(StreamInstallFrame::ArchiveEnd { archive_name })
                .await
        })
    }
}

fn archive_catalog_name(archive: &Path) -> eyre::Result<CanonicalCatalogPath> {
    let name = archive
        .file_name()
        .ok_or_else(|| eyre::eyre!("archive path has no file name: {}", archive.display()))?;
    let name = name.to_str().ok_or_else(|| {
        eyre::eyre!(
            "archive file name is not valid UTF-8: {}",
            archive.display()
        )
    })?;
    CanonicalCatalogPath::new(name)
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct RarListing {
    solid: bool,
    entries: Vec<RarEntry>,
}

impl RarListing {
    fn unpacked_size(&self) -> u64 {
        self.entries
            .iter()
            .filter(|entry| entry.kind == RarEntryKind::File)
            .map(|entry| entry.size)
            .sum()
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct RarEntry {
    relative_path: CanonicalCatalogPath,
    kind: RarEntryKind,
    size: u64,
    crc32: Option<u32>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RarEntryKind {
    File,
    Directory,
}

#[derive(Default)]
struct RarEntryDraft {
    relative_path: Option<String>,
    kind: Option<RarEntryKind>,
    size: Option<u64>,
    crc32: Option<u32>,
}

async fn unrar_listing(
    program: &Path,
    archive: &Path,
    cancel_token: &CancellationToken,
) -> eyre::Result<RarListing> {
    let process = ScopedProcess::spawn(
        program,
        [
            std::ffi::OsString::from("lt"),
            std::ffi::OsString::from("-cfg-"),
            std::ffi::OsString::from("-p-"),
            archive.as_os_str().to_owned(),
        ],
        cancel_token,
        UNRAR_LISTING_CAPTURE_LIMIT,
    )?;
    let output = process.wait().await?;
    reject_truncated_unrar_listing(output.stdout_truncated, output.stderr_truncated, archive)?;
    if !output.status.success() {
        eyre::bail!(
            "unrar lt failed for {} with status {}: {}",
            archive.display(),
            output.status,
            String::from_utf8_lossy(&output.stderr)
        );
    }

    parse_unrar_listing(&String::from_utf8_lossy(&output.stdout))
}

fn reject_truncated_unrar_listing(
    stdout_truncated: bool,
    stderr_truncated: bool,
    archive: &Path,
) -> eyre::Result<()> {
    if stdout_truncated || stderr_truncated {
        eyre::bail!(
            "unrar lt output for {} exceeded the {} byte per-pipe metadata limit",
            archive.display(),
            UNRAR_LISTING_CAPTURE_LIMIT
        );
    }
    Ok(())
}

fn parse_unrar_listing(output: &str) -> eyre::Result<RarListing> {
    let mut solid = false;
    let mut entries = Vec::new();
    let mut current = RarEntryDraft::default();

    for line in output.lines() {
        let trimmed = line.trim();
        if let Some(details) = trimmed.strip_prefix("Details:") {
            solid = details.to_ascii_lowercase().contains("solid");
            continue;
        }

        if let Some(name) = trimmed.strip_prefix("Name:") {
            push_rar_entry(&mut entries, std::mem::take(&mut current))?;
            current.relative_path = Some(name.trim().to_string());
            continue;
        }

        if let Some(kind) = trimmed.strip_prefix("Type:") {
            current.kind = match kind.trim() {
                "File" => Some(RarEntryKind::File),
                "Directory" => Some(RarEntryKind::Directory),
                _ => None,
            };
            continue;
        }

        if let Some(size) = trimmed.strip_prefix("Size:") {
            current.size = Some(size.trim().parse()?);
            continue;
        }

        if let Some(crc) = trimmed.strip_prefix("CRC32:") {
            current.crc32 = Some(u32::from_str_radix(crc.trim(), 16)?);
        }
    }

    push_rar_entry(&mut entries, current)?;
    Ok(RarListing { solid, entries })
}

fn push_rar_entry(entries: &mut Vec<RarEntry>, draft: RarEntryDraft) -> eyre::Result<()> {
    let Some(relative_path) = draft.relative_path else {
        return Ok(());
    };

    let Some(kind) = draft.kind else {
        return Ok(());
    };

    let (size, crc32) = match kind {
        RarEntryKind::File => {
            let size = draft
                .size
                .ok_or_else(|| eyre::eyre!("RAR file entry {relative_path} has no Size"))?;
            let crc32 = match (size, draft.crc32) {
                (_, Some(crc32)) => crc32,
                (0, None) => 0,
                (_, None) => {
                    eyre::bail!("RAR file entry {relative_path} has no CRC32");
                }
            };
            (size, Some(crc32))
        }
        RarEntryKind::Directory => (0, None),
    };

    entries.push(RarEntry {
        relative_path: CanonicalCatalogPath::new(relative_path)?,
        kind,
        size,
        crc32,
    });
    Ok(())
}

async fn stream_unrar_entries(
    program: &Path,
    archive: &Path,
    entries: &[RarEntry],
    frames: &StreamInstallFrameSink,
    cancel_token: CancellationToken,
) -> eyre::Result<()> {
    let mut command = Command::new(program);
    command
        .arg("p")
        .arg("-inul")
        .arg("-cfg-")
        .arg("-p-")
        .arg(archive)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::null());
    #[cfg(target_os = "windows")]
    command.creation_flags(crate::scoped_process::CREATE_NO_WINDOW);
    let child = command.spawn()?;
    let mut child = ReapedTokioChild::new(child);

    let result = async {
        let mut stdout = child
            .child_mut()
            .stdout
            .take()
            .ok_or_else(|| eyre::eyre!("unrar stdout was not captured"))?;
        let mut buffer = vec![0_u8; STREAM_CHUNK_SIZE];

        for entry in entries {
            if cancel_token.is_cancelled() {
                eyre::bail!("streamed archive {} was cancelled", archive.display());
            }

            match entry.kind {
                RarEntryKind::Directory => {
                    frames
                        .send(StreamInstallFrame::Directory {
                            relative_path: entry.relative_path.clone(),
                        })
                        .await?;
                }
                RarEntryKind::File => {
                    let Some(crc32) = entry.crc32 else {
                        eyre::bail!("RAR file entry {} has no CRC32", entry.relative_path);
                    };
                    frames
                        .send(StreamInstallFrame::FileBegin {
                            relative_path: entry.relative_path.clone(),
                            size: entry.size,
                            crc32,
                        })
                        .await?;
                    stream_unrar_file_from_stdout(
                        &mut stdout,
                        archive,
                        entry,
                        frames,
                        &mut buffer,
                        &cancel_token,
                    )
                    .await?;
                    frames
                        .send(StreamInstallFrame::FileEnd {
                            relative_path: entry.relative_path.clone(),
                        })
                        .await?;
                }
            }
        }

        let extra =
            read_unrar_stdout(&mut stdout, &mut buffer[..1], &cancel_token, archive).await?;
        if extra != 0 {
            eyre::bail!(
                "unrar produced bytes after listed entries for {}",
                archive.display()
            );
        }

        let status = wait_unrar_child(&mut child, &cancel_token, archive).await?;
        if !status.success() {
            eyre::bail!(
                "unrar p failed for {} with status {status}",
                archive.display()
            );
        }

        Ok(())
    }
    .await;

    if let Err(error) = result {
        if let Err(cleanup_error) = child.terminate_and_wait().await {
            return Err(eyre::eyre!(
                "{error}; failed to settle unrar for {}: {cleanup_error}",
                archive.display()
            ));
        }
        return Err(error);
    }

    Ok(())
}

async fn stream_unrar_file_from_stdout(
    stdout: &mut (impl AsyncRead + Unpin),
    archive: &Path,
    entry: &RarEntry,
    frames: &StreamInstallFrameSink,
    buffer: &mut [u8],
    cancel_token: &CancellationToken,
) -> eyre::Result<()> {
    let mut remaining = entry.size;
    while remaining > 0 {
        let read_len = usize::try_from(remaining.min(u64::try_from(buffer.len())?))?;
        let read =
            read_unrar_stdout(stdout, &mut buffer[..read_len], cancel_token, archive).await?;
        if read == 0 {
            eyre::bail!(
                "unrar ended while streaming {} from {}; {remaining} bytes missing",
                entry.relative_path,
                archive.display()
            );
        }

        frames
            .send(StreamInstallFrame::FileChunk {
                bytes: Bytes::copy_from_slice(&buffer[..read]),
            })
            .await?;
        remaining = remaining.saturating_sub(u64::try_from(read)?);
    }

    Ok(())
}

async fn read_unrar_stdout(
    stdout: &mut (impl AsyncRead + Unpin),
    buffer: &mut [u8],
    cancel_token: &CancellationToken,
    archive: &Path,
) -> eyre::Result<usize> {
    tokio::select! {
        () = cancel_token.cancelled() => {
            eyre::bail!("streamed archive {} was cancelled", archive.display());
        }
        read = stdout.read(buffer) => Ok(read?),
    }
}

async fn wait_unrar_child(
    child: &mut ReapedTokioChild,
    cancel_token: &CancellationToken,
    archive: &Path,
) -> eyre::Result<std::process::ExitStatus> {
    tokio::select! {
        () = cancel_token.cancelled() => {
            child.terminate_and_wait().await?;
            eyre::bail!("streamed archive {} was cancelled", archive.display());
        }
        status = child.wait() => Ok(status?),
    }
}

pub(crate) async fn send_stream_install_error(
    tx: SendStream,
    message: impl Into<String>,
    game_id: &str,
    cancel_token: &CancellationToken,
) -> SendStream {
    send_stream_install_error_cancellable(tx, message.into(), game_id, cancel_token).await
}

pub(crate) async fn send_game_install_stream(
    provider: Arc<dyn StreamInstallProvider>,
    tx: SendStream,
    game_root: &Path,
    game_id: &str,
    manifest: Arc<CatalogContentManifest>,
    cancel_token: CancellationToken,
) -> (SendStream, eyre::Result<()>) {
    let archives = match catalog_stream_archives(game_root, game_id, &manifest) {
        Ok(archives) => archives,
        Err(err) => {
            let message = err.to_string();
            let tx =
                send_stream_install_error_cancellable(tx, message.clone(), game_id, &cancel_token)
                    .await;
            return (tx, Err(eyre::eyre!(message)));
        }
    };
    let (frame_tx, mut frame_rx) = mpsc::channel(FRAME_CHANNEL_DEPTH);
    let producer_cancel = cancel_token.child_token();
    let frame_sink = StreamInstallFrameSink::new(frame_tx, producer_cancel.clone());
    let game_id_for_producer = game_id.to_string();
    let producer = {
        let provider = provider.clone();
        let producer_cancel = producer_cancel.clone();
        async move {
            for archive in archives {
                if producer_cancel.is_cancelled() {
                    eyre::bail!("streamed install for {game_id_for_producer} was cancelled");
                }

                if let Err(err) = provider
                    .stream_archive(&archive, frame_sink.clone(), producer_cancel.clone())
                    .await
                {
                    let message = err.to_string();
                    let _ = frame_sink.send(StreamInstallFrame::Error { message }).await;
                    return Err(err);
                }
            }

            let _ = frame_sink.send(StreamInstallFrame::Complete).await;
            Ok(())
        }
    };
    tokio::pin!(producer);

    let mut framed_tx = FramedWrite::new(tx, LengthDelimitedCodec::new());
    let (egress_outcome, producer_result) = {
        let egress = forward_stream_install_frames(&mut framed_tx, &mut frame_rx, &producer_cancel);
        tokio::pin!(egress);

        tokio::select! {
            biased;
            outcome = &mut egress => (outcome, None),
            result = &mut producer => {
                let outcome = egress.await;
                (outcome, Some(result))
            }
        }
    };
    let egress_result = egress_outcome.into_result(game_id);

    // Once egress stops exceptionally, make the stream unusable before waiting
    // for producer cleanup. This prevents queued or partially buffered frames
    // from looking like a valid prefix while archive work unwinds.
    let mut tx = framed_tx.into_inner();
    if egress_result.is_err() {
        producer_cancel.cancel();
        reset_stream_install(&mut tx, game_id);
    }
    drop(frame_rx);

    let producer_result = match producer_result {
        Some(result) => result,
        None => producer.await,
    };
    let result = egress_result.and(producer_result);

    (tx, result)
}

fn catalog_stream_archives(
    game_root: &Path,
    game_id: &str,
    manifest: &CatalogContentManifest,
) -> eyre::Result<Vec<PathBuf>> {
    if manifest.game_id() != game_id {
        eyre::bail!(
            "streamed-install catalog game mismatch: requested {game_id}, manifest is for {}",
            manifest.game_id()
        );
    }
    if !manifest.supports_streamed_install() {
        eyre::bail!("catalog game {game_id} does not support streamed install");
    }

    let expected = manifest
        .files()
        .iter()
        .filter(|entry| entry.kind() == CatalogEntryKind::File)
        .map(|entry| entry.canonical_path().as_str())
        .filter(|path| {
            !path.contains('/')
                && Path::new(path)
                    .extension()
                    .is_some_and(|extension| extension == "eti")
        })
        .collect::<Vec<_>>();
    if expected.is_empty() {
        eyre::bail!("catalog game {game_id} has streamed output but no root .eti archive");
    }

    let archives = root_eti_archives(game_root)?;
    let mut actual = archives
        .into_iter()
        .map(|archive| {
            let name = archive
                .file_name()
                .and_then(std::ffi::OsStr::to_str)
                .map(str::to_owned)
                .ok_or_else(|| {
                    eyre::eyre!(
                        "streamed-install archive name is not valid UTF-8: {}",
                        archive.display()
                    )
                })?;
            Ok((name, archive))
        })
        .collect::<eyre::Result<Vec<_>>>()?;
    actual.sort_by(|(left, _), (right, _)| left.cmp(right));
    let actual_names = actual
        .iter()
        .map(|(name, _)| name.as_str())
        .collect::<Vec<_>>();
    if actual_names != expected {
        eyre::bail!(
            "streamed-install archive set does not match catalog for {game_id}: expected {expected:?}, found {actual_names:?}"
        );
    }
    Ok(actual.into_iter().map(|(_, archive)| archive).collect())
}

async fn send_stream_install_error_cancellable(
    tx: SendStream,
    message: String,
    game_id: &str,
    cancel_token: &CancellationToken,
) -> SendStream {
    let (frame_tx, mut frame_rx) = mpsc::channel(1);
    frame_tx
        .try_send(StreamInstallFrame::Error { message })
        .expect("new one-slot StreamInstall error channel should accept one frame");
    drop(frame_tx);

    let mut framed_tx = FramedWrite::new(tx, LengthDelimitedCodec::new());
    let outcome = forward_stream_install_frames(&mut framed_tx, &mut frame_rx, cancel_token).await;
    let mut tx = framed_tx.into_inner();
    if let Err(err) = outcome.into_result(game_id) {
        reset_stream_install(&mut tx, game_id);
        log::debug!("Failed to send StreamInstall error for {game_id}: {err}");
    }
    tx
}

enum StreamInstallEgressOutcome {
    Complete,
    Cancelled,
    Failed(eyre::Report),
}

impl StreamInstallEgressOutcome {
    fn into_result(self, game_id: &str) -> eyre::Result<()> {
        match self {
            Self::Complete => Ok(()),
            Self::Cancelled => Err(eyre::eyre!("streamed install for {game_id} was cancelled")),
            Self::Failed(err) => Err(err),
        }
    }
}

async fn forward_stream_install_frames<W>(
    framed_tx: &mut FramedWrite<W, LengthDelimitedCodec>,
    frame_rx: &mut mpsc::Receiver<StreamInstallFrame>,
    cancel_token: &CancellationToken,
) -> StreamInstallEgressOutcome
where
    W: tokio::io::AsyncWrite + Unpin,
{
    forward_stream_install_frames_with_timeout(
        framed_tx,
        frame_rx,
        cancel_token,
        STREAM_INSTALL_INACTIVITY_TIMEOUT,
    )
    .await
}

async fn forward_stream_install_frames_with_timeout<W>(
    framed_tx: &mut FramedWrite<W, LengthDelimitedCodec>,
    frame_rx: &mut mpsc::Receiver<StreamInstallFrame>,
    cancel_token: &CancellationToken,
    inactivity_timeout: Duration,
) -> StreamInstallEgressOutcome
where
    W: tokio::io::AsyncWrite + Unpin,
{
    loop {
        let frame = tokio::select! {
            biased;
            () = cancel_token.cancelled() => {
                return StreamInstallEgressOutcome::Cancelled;
            }
            () = tokio::time::sleep(inactivity_timeout) => {
                return StreamInstallEgressOutcome::Failed(eyre::eyre!(
                    "streamed install producer timed out after {inactivity_timeout:?} without a frame"
                ));
            }
            frame = frame_rx.recv() => frame,
        };
        let Some(frame) = frame else {
            break;
        };

        let send_result = tokio::select! {
            biased;
            () = cancel_token.cancelled() => {
                return StreamInstallEgressOutcome::Cancelled;
            }
            () = tokio::time::sleep(inactivity_timeout) => {
                return StreamInstallEgressOutcome::Failed(eyre::eyre!(
                    "streamed install frame send timed out after {inactivity_timeout:?}"
                ));
            }
            result = framed_tx.send(frame.encode()) => result,
        };
        if let Err(err) = send_result {
            return StreamInstallEgressOutcome::Failed(eyre::eyre!(
                "failed to send streamed install frame: {err}"
            ));
        }
    }

    tokio::select! {
        biased;
        () = cancel_token.cancelled() => StreamInstallEgressOutcome::Cancelled,
        () = tokio::time::sleep(inactivity_timeout) => {
            StreamInstallEgressOutcome::Failed(eyre::eyre!(
                "streamed install close timed out after {inactivity_timeout:?}"
            ))
        }
        result = framed_tx.close() => match result {
            Ok(()) => StreamInstallEgressOutcome::Complete,
            Err(err) => StreamInstallEgressOutcome::Failed(eyre::eyre!(
                "failed to close streamed install stream: {err}"
            )),
        },
    }
}

fn reset_stream_install(tx: &mut SendStream, game_id: &str) {
    if let Err(err) = tx.reset(application::Error::UNKNOWN) {
        log::debug!("Failed to reset cancelled StreamInstall for {game_id}: {err}");
    }
}

/// Catalog-owned acceptance state for one streamed-install response.
///
/// The sender's archive framing is useful for bounded extraction progress, but
/// it is not authority. Every materialized path, shape, size, and file digest
/// ultimately comes from `manifest`.
#[derive(Debug)]
struct CatalogStreamVerifier {
    manifest: Arc<CatalogContentManifest>,
    expected_archives: HashSet<CanonicalCatalogPath>,
    expected_file_bytes: u64,
    seen_archives: HashSet<CanonicalCatalogPath>,
    reported_unpacked_bytes: u64,
    active_archive: Option<ActiveArchive>,
    open_file: Option<usize>,
    seen_entries: Vec<bool>,
    entry_frames: usize,
}

#[derive(Debug)]
struct ActiveArchive {
    name: CanonicalCatalogPath,
}

impl CatalogStreamVerifier {
    fn new(game_id: &str, manifest: Arc<CatalogContentManifest>) -> eyre::Result<Self> {
        if manifest.game_id() != game_id {
            eyre::bail!(
                "streamed install catalog game mismatch: requested {game_id}, manifest is for {}",
                manifest.game_id()
            );
        }
        if !manifest.supports_streamed_install() {
            eyre::bail!("catalog game {game_id} does not support streamed install");
        }

        let expected_archives = manifest
            .files()
            .iter()
            .filter(|entry| entry.kind() == CatalogEntryKind::File)
            .map(lanspread_db::content_manifest::CatalogFileEntry::canonical_path)
            .filter(|path| {
                !path.as_str().contains('/')
                    && Path::new(path.as_str())
                        .extension()
                        .is_some_and(|extension| extension == "eti")
            })
            .cloned()
            .collect::<HashSet<_>>();
        if expected_archives.is_empty() {
            eyre::bail!("catalog game {game_id} has streamed output but no root .eti archive");
        }

        let expected_file_bytes = manifest
            .streamed_install_files()
            .iter()
            .filter(|entry| entry.kind() == CatalogEntryKind::File)
            .try_fold(0_u64, |total, entry| total.checked_add(entry.size()))
            .ok_or_else(|| eyre::eyre!("catalog streamed file-size total overflow"))?;
        if expected_file_bytes > MAX_CATALOG_TOTAL_BYTES {
            eyre::bail!(
                "catalog streamed files exceed the {MAX_CATALOG_TOTAL_BYTES}-byte total limit"
            );
        }

        let seen_entries = vec![false; manifest.streamed_install_files().len()];
        Ok(Self {
            manifest,
            expected_archives,
            expected_file_bytes,
            seen_archives: HashSet::new(),
            reported_unpacked_bytes: 0,
            active_archive: None,
            open_file: None,
            seen_entries,
            entry_frames: 0,
        })
    }

    const fn expected_file_bytes(&self) -> u64 {
        self.expected_file_bytes
    }

    fn begin_archive(
        &mut self,
        archive_name: &CanonicalCatalogPath,
        unpacked_size: u64,
    ) -> eyre::Result<()> {
        self.ensure_no_open_file("ArchiveBegin")?;
        if let Some(active) = &self.active_archive {
            eyre::bail!(
                "received ArchiveBegin for {archive_name} before ArchiveEnd for {}",
                active.name
            );
        }
        let reported_unpacked_bytes = self
            .reported_unpacked_bytes
            .checked_add(unpacked_size)
            .ok_or_else(|| eyre::eyre!("streamed archive unpacked-size total overflow"))?;
        if reported_unpacked_bytes > MAX_CATALOG_TOTAL_BYTES {
            eyre::bail!(
                "streamed archives exceed the {MAX_CATALOG_TOTAL_BYTES}-byte reported unpacked limit"
            );
        }

        if !self.expected_archives.contains(archive_name) {
            eyre::bail!("received unknown streamed archive {archive_name}");
        }
        if !self.seen_archives.insert(archive_name.clone()) {
            eyre::bail!("received duplicate streamed archive {archive_name}");
        }

        self.reported_unpacked_bytes = reported_unpacked_bytes;
        self.active_archive = Some(ActiveArchive {
            name: archive_name.clone(),
        });
        Ok(())
    }

    fn record_directory(&mut self, relative_path: &CanonicalCatalogPath) -> eyre::Result<()> {
        self.ensure_no_open_file("Directory")?;
        self.record_entry_frame()?;
        let entry_index = self.expected_entry_index(relative_path, CatalogEntryKind::Directory)?;
        debug_assert_eq!(
            self.manifest.streamed_install_files()[entry_index].size(),
            0
        );

        self.active_archive_mut("Directory")?;
        self.seen_entries[entry_index] = true;
        Ok(())
    }

    fn begin_file(
        &mut self,
        relative_path: &CanonicalCatalogPath,
        size: u64,
    ) -> eyre::Result<Blake3Digest> {
        self.ensure_no_open_file("FileBegin")?;
        self.record_entry_frame()?;
        self.active_archive_mut("FileBegin")?;
        let entry_index = self.expected_entry_index(relative_path, CatalogEntryKind::File)?;
        let expected = &self.manifest.streamed_install_files()[entry_index];
        let expected_size = expected.size();
        let expected_blake3 = expected.file_blake3().ok_or_else(|| {
            eyre::eyre!("catalog streamed file {relative_path} has no BLAKE3 digest")
        })?;
        if size != expected_size {
            eyre::bail!(
                "streamed file {relative_path} size mismatch: sender declared {size}, catalog expects {}",
                expected_size
            );
        }
        if self.seen_entries[entry_index] {
            eyre::bail!("streamed install repeated file {relative_path}");
        }
        self.seen_entries[entry_index] = true;
        self.open_file = Some(entry_index);
        Ok(expected_blake3)
    }

    fn record_file_chunk(&self, length: usize) -> eyre::Result<()> {
        if self.open_file.is_none() {
            eyre::bail!("received FileChunk without FileBegin");
        }
        if length == 0 {
            eyre::bail!("received an empty FileChunk");
        }
        Ok(())
    }

    fn end_file(&mut self, relative_path: &CanonicalCatalogPath) -> eyre::Result<()> {
        let Some(open_file) = self.open_file else {
            eyre::bail!("received FileEnd for {relative_path} without FileBegin");
        };
        let open_path = self.manifest.streamed_install_files()[open_file].canonical_path();
        if open_path != relative_path {
            eyre::bail!("streamed file end mismatch: began {open_path}, ended {relative_path}");
        }
        self.open_file = None;
        Ok(())
    }

    fn end_archive(&mut self, archive_name: &CanonicalCatalogPath) -> eyre::Result<()> {
        self.ensure_no_open_file("ArchiveEnd")?;
        let Some(active) = self.active_archive.take() else {
            eyre::bail!("received ArchiveEnd for {archive_name} without ArchiveBegin");
        };
        if &active.name != archive_name {
            eyre::bail!(
                "streamed archive end mismatch: began {}, ended {archive_name}",
                active.name
            );
        }
        Ok(())
    }

    fn verify_complete(&self) -> eyre::Result<()> {
        self.ensure_no_open_file("Complete")?;
        if let Some(active) = &self.active_archive {
            eyre::bail!(
                "streamed install completed before ArchiveEnd for {}",
                active.name
            );
        }
        if let Some(missing) = self
            .expected_archives
            .difference(&self.seen_archives)
            .next()
        {
            eyre::bail!("streamed install completed before expected archive {missing}");
        }

        for (entry, seen) in self
            .manifest
            .streamed_install_files()
            .iter()
            .zip(&self.seen_entries)
        {
            if !*seen {
                let path = entry.canonical_path();
                eyre::bail!("streamed install is missing catalog entry {path}");
            }
        }
        Ok(())
    }

    fn expected_entry_index(
        &self,
        relative_path: &CanonicalCatalogPath,
        kind: CatalogEntryKind,
    ) -> eyre::Result<usize> {
        let entry_index = self
            .manifest
            .streamed_install_files()
            .binary_search_by(|entry| entry.canonical_path().cmp(relative_path))
            .map_err(|_| eyre::eyre!("streamed install sent unknown path {relative_path}"))?;
        let entry = &self.manifest.streamed_install_files()[entry_index];
        if entry.kind() != kind {
            eyre::bail!(
                "streamed install path {relative_path} has kind {:?}, catalog expects {:?}",
                kind,
                entry.kind()
            );
        }
        Ok(entry_index)
    }

    fn active_archive_mut(&mut self, frame: &str) -> eyre::Result<&mut ActiveArchive> {
        self.active_archive
            .as_mut()
            .ok_or_else(|| eyre::eyre!("received {frame} outside an archive"))
    }

    fn ensure_no_open_file(&self, frame: &str) -> eyre::Result<()> {
        if let Some(open_file) = self.open_file {
            let open_path = self.manifest.streamed_install_files()[open_file]
                .canonical_path()
                .as_str();
            eyre::bail!("received {frame} while streamed file {open_path} is open");
        }
        Ok(())
    }

    fn record_entry_frame(&mut self) -> eyre::Result<()> {
        // The publisher applies the same bound to the aggregate raw archive
        // listings before it folds repeated directories into final outputs.
        self.entry_frames = self
            .entry_frames
            .checked_add(1)
            .ok_or_else(|| eyre::eyre!("streamed install entry-frame count overflow"))?;
        if self.entry_frames > MAX_CATALOG_ENTRIES {
            eyre::bail!("streamed install exceeds the {MAX_CATALOG_ENTRIES}-entry frame limit");
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ReceiveFrameOutcome {
    Continue,
    Complete,
}

struct StreamInstallReceiveState {
    verifier: CatalogStreamVerifier,
    staging_dir: PathBuf,
    current_file: Option<IncomingFile>,
    progress: StreamInstallProgress,
}

impl StreamInstallReceiveState {
    fn new(
        game_id: &str,
        manifest: Arc<CatalogContentManifest>,
        staging_dir: &Path,
        attempt: DownloadAttemptReporter,
    ) -> StreamInstallReceiveResult<Self> {
        let verifier = CatalogStreamVerifier::new(game_id, manifest)
            .map_err(StreamInstallReceiveError::setup)?;
        let staging_dir =
            scoped_blocking(|| std::fs::canonicalize(staging_dir)).map_err(|error| {
                StreamInstallReceiveError::setup(eyre::eyre!(
                    "failed to resolve streamed install staging directory {}: {error}",
                    staging_dir.display()
                ))
            })?;
        let progress = StreamInstallProgress::new(attempt, verifier.expected_file_bytes());
        Ok(Self {
            verifier,
            staging_dir,
            current_file: None,
            progress,
        })
    }

    fn emit_current_progress(&mut self) {
        self.progress.emit_current();
    }

    fn handle_frame(
        &mut self,
        frame: StreamInstallFrame,
        game_id: &str,
        peer_endpoint: PeerEndpoint,
        content_id: ContentId,
        tx_notify_ui: &UnboundedSender<PeerEvent>,
    ) -> StreamInstallReceiveResult<ReceiveFrameOutcome> {
        match frame {
            StreamInstallFrame::ArchiveBegin {
                archive_name,
                solid,
                unpacked_size,
            } => {
                self.verifier
                    .begin_archive(&archive_name, unpacked_size)
                    .map_err(StreamInstallReceiveError::integrity)?;
                self.progress.emit_snapshot(0);
                log::info!(
                    "Receiving streamed install archive {archive_name} for {game_id} \
                     (solid={solid}, unpacked_size={unpacked_size})"
                );
            }
            StreamInstallFrame::Directory { relative_path } => {
                self.verifier
                    .record_directory(&relative_path)
                    .map_err(StreamInstallReceiveError::integrity)?;
                let path = resolve_stream_path(&self.staging_dir, &relative_path)
                    .map_err(StreamInstallReceiveError::setup)?;
                scoped_blocking(|| std::fs::create_dir_all(path))
                    .map_err(StreamInstallReceiveError::setup)?;
            }
            StreamInstallFrame::FileBegin {
                relative_path,
                size,
                crc32,
            } => {
                let expected_blake3 = self
                    .verifier
                    .begin_file(&relative_path, size)
                    .map_err(StreamInstallReceiveError::integrity)?;
                let path = resolve_stream_path(&self.staging_dir, &relative_path)
                    .map_err(StreamInstallReceiveError::setup)?;
                if let Some(parent) = path.parent() {
                    scoped_blocking(|| std::fs::create_dir_all(parent))
                        .map_err(StreamInstallReceiveError::setup)?;
                }
                let file = scoped_blocking(|| File::create(&path))
                    .map_err(StreamInstallReceiveError::setup)?;
                self.current_file = Some(IncomingFile::new(
                    relative_path,
                    path,
                    size,
                    crc32,
                    expected_blake3,
                    file,
                ));
            }
            StreamInstallFrame::FileChunk { bytes } => {
                self.verifier
                    .record_file_chunk(bytes.len())
                    .map_err(StreamInstallReceiveError::integrity)?;
                let Some(file) = self.current_file.as_mut() else {
                    unreachable!("verifier and incoming file state must stay in lockstep");
                };
                let length = file.write_chunk(&bytes)?;
                self.progress.record_bytes(length);
            }
            StreamInstallFrame::FileEnd { relative_path } => {
                self.verifier
                    .end_file(&relative_path)
                    .map_err(StreamInstallReceiveError::integrity)?;
                let Some(file) = self.current_file.take() else {
                    unreachable!("verifier and incoming file state must stay in lockstep");
                };
                file.finish(
                    &relative_path,
                    game_id,
                    peer_endpoint,
                    content_id,
                    tx_notify_ui,
                )?;
            }
            StreamInstallFrame::ArchiveEnd { archive_name } => {
                self.verifier
                    .end_archive(&archive_name)
                    .map_err(StreamInstallReceiveError::integrity)?;
                log::info!("Finished streamed install archive {archive_name} for {game_id}");
            }
            StreamInstallFrame::Complete => {
                self.verifier
                    .verify_complete()
                    .map_err(StreamInstallReceiveError::integrity)?;
                debug_assert!(self.current_file.is_none());
                self.progress.emit_snapshot(0);
                return Ok(ReceiveFrameOutcome::Complete);
            }
            StreamInstallFrame::Error { message } => {
                return Err(StreamInstallReceiveError::transport(eyre::eyre!(
                    "streamed install sender failed: {message}"
                )));
            }
        }
        Ok(ReceiveFrameOutcome::Continue)
    }
}

pub(crate) struct ReceiveStreamedInstallRequest<'a> {
    pub(crate) endpoint: PeerEndpoint,
    pub(crate) game_id: &'a str,
    pub(crate) manifest: Arc<CatalogContentManifest>,
    pub(crate) staging_dir: &'a Path,
    pub(crate) attempt: DownloadAttemptReporter,
    pub(crate) tx_notify_ui: UnboundedSender<PeerEvent>,
    pub(crate) quic: &'a QuicConnector,
    pub(crate) cancel_token: CancellationToken,
}

pub(crate) async fn receive_streamed_install(
    request: ReceiveStreamedInstallRequest<'_>,
) -> StreamInstallReceiveResult<()> {
    let ReceiveStreamedInstallRequest {
        endpoint,
        game_id,
        manifest,
        staging_dir,
        attempt,
        tx_notify_ui,
        quic,
        cancel_token,
    } = request;
    let content_id = manifest.content_id();
    let mut state = StreamInstallReceiveState::new(game_id, manifest, staging_dir, attempt)?;
    let mut conn = connect_to_peer(quic, &endpoint, &cancel_token)
        .await
        .map_err(|error| {
            StreamInstallReceiveError::transport_or_cancelled(
                error,
                game_id,
                &cancel_token,
                "while connecting",
            )
        })?;
    let mut inactivity = StreamInstallInactivityDeadline::ordinary();
    let stream = await_stream_install_open(
        async { Ok(conn.open_bidirectional_stream().await?) },
        game_id,
        &cancel_token,
        inactivity,
    )
    .await?;
    let (rx, tx) = stream.split();
    let mut request_scope = StreamInstallRequestScope::new(rx, tx, game_id, content_id);
    request_scope
        .send_request(&cancel_token, inactivity)
        .await?;
    let rx = request_scope.into_receive();
    let mut framed_rx = StreamInstallReceiveScope::new(rx, game_id);
    let mut progress_interval = time::interval(STREAM_INSTALL_PROGRESS_UPDATE_INTERVAL);
    progress_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
    progress_interval.tick().await;

    loop {
        let next = match await_stream_install_input(
            framed_rx.next(),
            progress_interval.tick(),
            game_id,
            &cancel_token,
            inactivity,
        )
        .await?
        {
            StreamInstallInput::ProgressTick => {
                state.emit_current_progress();
                continue;
            }
            StreamInstallInput::Frame(next) => next,
        };

        let Some(frame) = next else {
            return Err(stream_ended_before_complete(game_id));
        };
        let frame = frame
            .map_err(|error| {
                classify_stream_install_read_error(
                    error,
                    game_id,
                    &cancel_token,
                    "while reading its response",
                )
            })?
            .freeze();
        inactivity.reset_after_frame();
        let frame = decode_received_stream_install_frame(frame)?;
        if state.handle_frame(frame, game_id, endpoint, content_id, &tx_notify_ui)?
            == ReceiveFrameOutcome::Complete
        {
            return framed_rx.drain_fin(&cancel_token, inactivity).await;
        }
    }
}

#[derive(Debug)]
enum StreamInstallInput<T> {
    ProgressTick,
    Frame(T),
}

async fn await_stream_install_input<T, P>(
    frame: impl Future<Output = T>,
    progress_tick: impl Future<Output = P>,
    game_id: &str,
    cancel_token: &CancellationToken,
    inactivity: StreamInstallInactivityDeadline,
) -> StreamInstallReceiveResult<StreamInstallInput<T>> {
    tokio::select! {
        biased;
        () = cancel_token.cancelled() => {
            Err(StreamInstallReceiveError::cancelled(game_id, "while receiving"))
        }
        () = time::sleep_until(inactivity.expires_at) => {
            Err(inactivity.timeout_error(game_id, "while waiting for a response frame"))
        }
        _ = progress_tick => Ok(StreamInstallInput::ProgressTick),
        frame = frame => Ok(StreamInstallInput::Frame(frame)),
    }
}

fn stream_ended_before_complete(game_id: &str) -> StreamInstallReceiveError {
    StreamInstallReceiveError::transport(eyre::eyre!(
        "streamed install for {game_id} ended before Complete"
    ))
}

fn data_after_complete(game_id: &str) -> StreamInstallReceiveError {
    StreamInstallReceiveError::integrity(eyre::eyre!(
        "streamed install for {game_id} sent data after Complete"
    ))
}

fn decode_received_stream_install_frame(
    frame: Bytes,
) -> StreamInstallReceiveResult<StreamInstallFrame> {
    StreamInstallFrame::decode_checked(frame).map_err(|error| {
        StreamInstallReceiveError::integrity(eyre::eyre!(
            "invalid streamed install frame from peer: {error}"
        ))
    })
}

fn classify_stream_install_read_error(
    error: std::io::Error,
    game_id: &str,
    cancel_token: &CancellationToken,
    context: &str,
) -> StreamInstallReceiveError {
    if error.kind() == std::io::ErrorKind::InvalidData {
        return StreamInstallReceiveError::integrity(eyre::eyre!(
            "invalid or oversized streamed install frame from peer: {error}"
        ));
    }
    StreamInstallReceiveError::transport_or_cancelled(error, game_id, cancel_token, context)
}

async fn await_stream_install_open<T>(
    open: impl Future<Output = eyre::Result<T>>,
    game_id: &str,
    cancel_token: &CancellationToken,
    inactivity: StreamInstallInactivityDeadline,
) -> StreamInstallReceiveResult<T> {
    inactivity
        .run(open, game_id, cancel_token, "while opening its stream")
        .await?
        .map_err(|error| {
            StreamInstallReceiveError::transport_or_cancelled(
                error,
                game_id,
                cancel_token,
                "while opening its stream",
            )
        })
}

async fn send_stream_install_request_frame<W>(
    framed_tx: &mut FramedWrite<W, LengthDelimitedCodec>,
    game_id: &str,
    content_id: ContentId,
    cancel_token: &CancellationToken,
    inactivity: StreamInstallInactivityDeadline,
) -> StreamInstallReceiveResult<()>
where
    W: tokio::io::AsyncWrite + Unpin,
{
    let request = Request::StreamInstall {
        game_id: game_id.to_string(),
        content_id,
    }
    .encode()
    .map_err(StreamInstallReceiveError::setup)?;
    inactivity
        .run(
            framed_tx.send(request),
            game_id,
            cancel_token,
            "while sending its request",
        )
        .await?
        .map_err(|error| {
            StreamInstallReceiveError::transport_or_cancelled(
                error,
                game_id,
                cancel_token,
                "while sending its request",
            )
        })?;

    inactivity
        .run(
            framed_tx.close(),
            game_id,
            cancel_token,
            "while closing its request",
        )
        .await?
        .map_err(|error| {
            StreamInstallReceiveError::transport_or_cancelled(
                error,
                game_id,
                cancel_token,
                "while closing its request",
            )
        })?;
    Ok(())
}

struct StreamInstallRequestScope<'a> {
    rx: Option<ReceiveStream>,
    framed_tx: Option<FramedWrite<SendStream, LengthDelimitedCodec>>,
    game_id: &'a str,
    content_id: ContentId,
}

impl<'a> StreamInstallRequestScope<'a> {
    fn new(rx: ReceiveStream, tx: SendStream, game_id: &'a str, content_id: ContentId) -> Self {
        Self {
            rx: Some(rx),
            framed_tx: Some(FramedWrite::new(tx, LengthDelimitedCodec::new())),
            game_id,
            content_id,
        }
    }

    async fn send_request(
        &mut self,
        cancel_token: &CancellationToken,
        inactivity: StreamInstallInactivityDeadline,
    ) -> StreamInstallReceiveResult<()> {
        let framed_tx = self
            .framed_tx
            .as_mut()
            .expect("request scope should retain its send half until completion");
        send_stream_install_request_frame(
            framed_tx,
            self.game_id,
            self.content_id,
            cancel_token,
            inactivity,
        )
        .await
    }

    fn into_receive(mut self) -> ReceiveStream {
        let framed_tx = self
            .framed_tx
            .take()
            .expect("completed request should retain its send half");
        drop(framed_tx.into_inner());
        self.rx
            .take()
            .expect("completed request should retain its receive half")
    }
}

impl Drop for StreamInstallRequestScope<'_> {
    fn drop(&mut self) {
        if let Some(framed_tx) = self.framed_tx.take() {
            let mut tx = framed_tx.into_inner();
            reset_stream_install(&mut tx, self.game_id);
        }
        if let Some(mut rx) = self.rx.take() {
            stop_stream_install_receive(&mut rx, self.game_id);
        }
    }
}

fn stop_stream_install_receive(rx: &mut ReceiveStream, game_id: &str) {
    if let Err(err) = rx.stop_sending(application::Error::UNKNOWN) {
        log::debug!("Failed to stop streamed install receive for {game_id}: {err}");
    }
}

struct StreamInstallReceiveScope<'a> {
    framed_rx: FramedRead<ReceiveStream, LengthDelimitedCodec>,
    game_id: &'a str,
    finished: bool,
}

impl<'a> StreamInstallReceiveScope<'a> {
    fn new(rx: ReceiveStream, game_id: &'a str) -> Self {
        Self {
            framed_rx: FramedRead::new(
                rx,
                LengthDelimitedCodec::builder()
                    .max_frame_length(MAX_STREAM_INSTALL_FRAME_BYTES)
                    .new_codec(),
            ),
            game_id,
            finished: false,
        }
    }

    async fn next(&mut self) -> Option<Result<bytes::BytesMut, std::io::Error>> {
        self.framed_rx.next().await
    }

    async fn drain_fin(
        mut self,
        cancel_token: &CancellationToken,
        inactivity: StreamInstallInactivityDeadline,
    ) -> StreamInstallReceiveResult<()> {
        let trailing = await_stream_install_fin(
            self.framed_rx.next(),
            self.game_id,
            cancel_token,
            inactivity,
        )
        .await?;
        match trailing {
            None => {
                self.finished = true;
                Ok(())
            }
            Some(Ok(_)) => Err(data_after_complete(self.game_id)),
            Some(Err(error)) => Err(classify_stream_install_read_error(
                error,
                self.game_id,
                cancel_token,
                "while draining its response",
            )),
        }
    }
}

async fn await_stream_install_fin(
    next: impl Future<Output = Option<Result<bytes::BytesMut, std::io::Error>>>,
    game_id: &str,
    cancel_token: &CancellationToken,
    inactivity: StreamInstallInactivityDeadline,
) -> StreamInstallReceiveResult<Option<Result<bytes::BytesMut, std::io::Error>>> {
    inactivity
        .run(
            next,
            game_id,
            cancel_token,
            "while waiting for response FIN",
        )
        .await
}

impl Drop for StreamInstallReceiveScope<'_> {
    fn drop(&mut self) {
        if !self.finished {
            stop_stream_install_receive(self.framed_rx.get_mut(), self.game_id);
        }
    }
}

struct StreamInstallProgress {
    attempt: DownloadAttemptReporter,
    total_bytes: u64,
    downloaded_bytes: u64,
    last_downloaded_bytes: u64,
    last_at: Instant,
}

impl StreamInstallProgress {
    fn new(attempt: DownloadAttemptReporter, total_bytes: u64) -> Self {
        Self {
            attempt,
            total_bytes,
            downloaded_bytes: 0,
            last_downloaded_bytes: 0,
            last_at: Instant::now(),
        }
    }

    fn record_bytes(&mut self, bytes: u64) {
        self.downloaded_bytes = self.downloaded_bytes.saturating_add(bytes);
    }

    fn emit_current(&mut self) {
        let now = Instant::now();
        let speed = bytes_per_second(
            self.downloaded_bytes
                .saturating_sub(self.last_downloaded_bytes),
            now.duration_since(self.last_at),
        );

        self.last_downloaded_bytes = self.downloaded_bytes;
        self.last_at = now;
        self.emit_snapshot(speed);
    }

    fn emit_snapshot(&self, bytes_per_second: u64) {
        self.attempt.emit_progress(DownloadProgress {
            attempt: self.attempt.key().clone(),
            downloaded_bytes: self.downloaded_bytes,
            total_bytes: self.total_bytes,
            bytes_per_second,
            active_peer_count: 1,
        });
    }
}

fn bytes_per_second(bytes: u64, elapsed: Duration) -> u64 {
    let millis = elapsed.as_millis().max(1);
    let rate = u128::from(bytes).saturating_mul(1_000) / millis;
    u64::try_from(rate).unwrap_or(u64::MAX)
}

struct IncomingFile<W = File> {
    relative_path: CanonicalCatalogPath,
    path: PathBuf,
    integrity: SenderArchiveIntegrity,
    expected_blake3: Blake3Digest,
    received: u64,
    crc32: Hasher,
    blake3: blake3::Hasher,
    file: W,
}

impl<W: std::io::Write> IncomingFile<W> {
    fn new(
        relative_path: CanonicalCatalogPath,
        path: PathBuf,
        expected_size: u64,
        expected_crc32: u32,
        expected_blake3: Blake3Digest,
        file: W,
    ) -> Self {
        Self {
            relative_path,
            path,
            integrity: SenderArchiveIntegrity::new(expected_size, expected_crc32),
            expected_blake3,
            received: 0,
            crc32: Hasher::new(),
            blake3: blake3::Hasher::new(),
            file,
        }
    }

    fn write_chunk(&mut self, bytes: &[u8]) -> StreamInstallReceiveResult<u64> {
        let offset = self.received;
        let length = u64::try_from(bytes.len()).map_err(StreamInstallReceiveError::setup)?;
        if offset.saturating_add(length) > self.integrity.expected_size {
            return Err(StreamInstallReceiveError::integrity(eyre::eyre!(
                "streamed file {} exceeded expected size {}",
                self.relative_path,
                self.integrity.expected_size
            )));
        }
        scoped_blocking(|| self.file.write_all(bytes)).map_err(StreamInstallReceiveError::setup)?;
        self.crc32.update(bytes);
        self.blake3.update(bytes);
        self.received = self.received.saturating_add(length);

        Ok(length)
    }

    fn finish(
        mut self,
        relative_path: &CanonicalCatalogPath,
        game_id: &str,
        peer_endpoint: PeerEndpoint,
        content_id: ContentId,
        tx_notify_ui: &UnboundedSender<PeerEvent>,
    ) -> StreamInstallReceiveResult<()> {
        if &self.relative_path != relative_path {
            return Err(StreamInstallReceiveError::integrity(eyre::eyre!(
                "streamed file end mismatch: began {}, ended {relative_path}",
                self.relative_path
            )));
        }
        let actual_crc32 = self.crc32.finalize();
        self.integrity
            .verify(&self.relative_path, self.received, actual_crc32)
            .map_err(StreamInstallReceiveError::integrity)?;
        let actual_blake3 = Blake3Digest::from_bytes(*self.blake3.finalize().as_bytes());
        if actual_blake3 != self.expected_blake3 {
            return Err(StreamInstallReceiveError::integrity(eyre::eyre!(
                "streamed file {} catalog BLAKE3 mismatch: got {actual_blake3}, expected {}",
                self.relative_path,
                self.expected_blake3
            )));
        }
        scoped_blocking(|| self.file.flush()).map_err(StreamInstallReceiveError::setup)?;

        let _ = tx_notify_ui.send(PeerEvent::DownloadGameFileChunkFinished {
            id: game_id.to_string(),
            peer_id: peer_endpoint.peer_id,
            peer_addr: peer_endpoint.addr,
            content_id,
            relative_path: self.relative_path.clone(),
            offset: 0,
            length: self.received,
        });

        log::debug!(
            "Received streamed file {} -> {}",
            self.relative_path,
            self.path.display()
        );
        Ok(())
    }
}

fn resolve_stream_path(
    staging_dir: &Path,
    relative_path: &CanonicalCatalogPath,
) -> eyre::Result<PathBuf> {
    validate_game_file_path(staging_dir, relative_path.as_str())
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{
            Arc,
            Condvar,
            Mutex,
            atomic::{AtomicBool, AtomicUsize, Ordering},
            mpsc as std_mpsc,
        },
        task::{Context, Poll},
    };

    use lanspread_db::content_manifest::{
        CatalogContentManifestBody,
        CatalogExtractedEntry,
        CatalogFileEntry,
    };
    use tokio::sync::Notify;

    use super::*;
    use crate::test_support::TempDir;

    fn canonical_path(value: &str) -> CanonicalCatalogPath {
        CanonicalCatalogPath::new(value).expect("test path should be canonical")
    }

    fn content_id() -> ContentId {
        ContentId::from_bytes([7; 32])
    }

    fn peer_endpoint() -> PeerEndpoint {
        PeerEndpoint::new(
            lanspread_proto::PeerId::from_bytes([9; 32]),
            "127.0.0.1:1".parse().expect("address should parse"),
        )
    }

    struct PendingWriter {
        write_polls: Arc<AtomicUsize>,
        first_write: Arc<Notify>,
    }

    struct CancelOnShutdownWriter {
        cancel_token: CancellationToken,
    }

    struct PendingShutdownWriter {
        first_shutdown: Arc<Notify>,
    }

    struct DelayedFileWriter {
        gate: Arc<(Mutex<bool>, Condvar)>,
        entered: Option<tokio::sync::oneshot::Sender<()>>,
        quiesced: Arc<std::sync::atomic::AtomicBool>,
    }

    struct RollbackCanary {
        quiesced: Arc<std::sync::atomic::AtomicBool>,
        rollback_started: Arc<std::sync::atomic::AtomicBool>,
        rollback_before_quiescence: Arc<std::sync::atomic::AtomicBool>,
    }

    struct DropProbe(Arc<AtomicBool>);

    impl Drop for DropProbe {
        fn drop(&mut self) {
            self.0.store(true, Ordering::SeqCst);
        }
    }

    impl std::io::Write for DelayedFileWriter {
        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
            if let Some(entered) = self.entered.take() {
                let _ = entered.send(());
            }
            let (gate_open, wake) = &*self.gate;
            let gate_open = gate_open.lock().expect("write gate must not be poisoned");
            let _gate_open = wake
                .wait_while(gate_open, |gate_open| !*gate_open)
                .expect("write gate must not be poisoned");
            self.quiesced
                .store(true, std::sync::atomic::Ordering::SeqCst);
            Ok(bytes.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    impl Drop for RollbackCanary {
        fn drop(&mut self) {
            if !self.quiesced.load(std::sync::atomic::Ordering::SeqCst) {
                self.rollback_before_quiescence
                    .store(true, std::sync::atomic::Ordering::SeqCst);
            }
            self.rollback_started
                .store(true, std::sync::atomic::Ordering::SeqCst);
        }
    }

    impl tokio::io::AsyncWrite for PendingWriter {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            _buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            self.write_polls.fetch_add(1, Ordering::SeqCst);
            self.first_write.notify_one();
            Poll::Pending
        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    impl tokio::io::AsyncWrite for CancelOnShutdownWriter {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            Poll::Ready(Ok(buf.len()))
        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            self.cancel_token.cancel();
            Poll::Ready(Ok(()))
        }
    }

    impl tokio::io::AsyncWrite for PendingShutdownWriter {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            Poll::Ready(Ok(buf.len()))
        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            self.first_shutdown.notify_one();
            Poll::Pending
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn aborted_incoming_write_quiesces_before_rollback_can_start() {
        let gate = Arc::new((Mutex::new(false), Condvar::new()));
        let gate_for_release = gate.clone();
        let (request_release, release_requested) = std_mpsc::channel();
        let release_thread = std::thread::spawn(move || {
            let _ = release_requested.recv_timeout(Duration::from_secs(2));
            let (gate_open, wake) = &*gate_for_release;
            let mut gate_open = gate_open.lock().expect("write gate must not be poisoned");
            *gate_open = true;
            wake.notify_one();
        });
        let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
        let quiesced = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let rollback_started = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let rollback_before_quiescence = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let mut task = tokio::spawn({
            let quiesced = quiesced.clone();
            let rollback_started = rollback_started.clone();
            let rollback_before_quiescence = rollback_before_quiescence.clone();
            async move {
                let _rollback = RollbackCanary {
                    quiesced: quiesced.clone(),
                    rollback_started,
                    rollback_before_quiescence,
                };
                let bytes = Bytes::from_static(b"x");
                let mut incoming = IncomingFile::new(
                    canonical_path("payload.bin"),
                    PathBuf::from("payload.bin"),
                    1,
                    crc32_of(&bytes),
                    Blake3Digest::hash(&bytes),
                    DelayedFileWriter {
                        gate,
                        entered: Some(entered_tx),
                        quiesced,
                    },
                );
                incoming
                    .write_chunk(&bytes)
                    .expect("controlled write should succeed");
                std::future::pending::<()>().await;
            }
        });

        tokio::time::timeout(Duration::from_secs(2), entered_rx)
            .await
            .expect("controlled write should start")
            .expect("controlled writer should retain entry sender");
        task.abort();
        assert!(
            tokio::time::timeout(Duration::from_millis(50), &mut task)
                .await
                .is_err(),
            "aborted receiver must remain alive while its write is in progress"
        );
        assert!(!quiesced.load(std::sync::atomic::Ordering::SeqCst));
        assert!(!rollback_started.load(std::sync::atomic::Ordering::SeqCst));

        request_release
            .send(())
            .expect("release thread should remain available");
        release_thread
            .join()
            .expect("release thread should not panic");
        task.await
            .expect_err("aborted receiver should stop after write quiescence");

        assert!(quiesced.load(std::sync::atomic::Ordering::SeqCst));
        assert!(rollback_started.load(std::sync::atomic::Ordering::SeqCst));
        assert!(!rollback_before_quiescence.load(std::sync::atomic::Ordering::SeqCst));
    }

    #[tokio::test]
    async fn producer_sink_cancellation_wins_over_channel_capacity() {
        let (frame_tx, mut frame_rx) = mpsc::channel(1);
        let cancel_token = CancellationToken::new();
        cancel_token.cancel();
        let sink = StreamInstallFrameSink::new(frame_tx, cancel_token);

        let err = sink
            .send(StreamInstallFrame::Complete)
            .await
            .expect_err("cancelled producer sink should reject frames");

        assert!(err.to_string().contains("cancelled"));
        assert!(matches!(
            frame_rx.try_recv(),
            Err(mpsc::error::TryRecvError::Empty)
        ));
    }

    #[tokio::test]
    async fn incoming_open_cancellation_settles_its_inline_scope() {
        let cancelled = CancellationToken::new();
        let dropped = Arc::new(AtomicBool::new(false));
        let probe = DropProbe(dropped.clone());
        let open = async move {
            let _probe = probe;
            std::future::pending::<eyre::Result<()>>().await
        };
        cancelled.cancel();

        let error = await_stream_install_open(
            open,
            "game",
            &cancelled,
            StreamInstallInactivityDeadline::ordinary(),
        )
        .await
        .expect_err("cancelled stream open should fail");

        assert!(error.to_string().contains("opening its stream"));
        assert!(dropped.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn incoming_open_has_an_application_inactivity_deadline() {
        let cancellation = CancellationToken::new();
        let error = await_stream_install_open(
            std::future::pending::<eyre::Result<()>>(),
            "game",
            &cancellation,
            StreamInstallInactivityDeadline::after(Duration::from_millis(25)),
        )
        .await
        .expect_err("pending stream open must time out");

        assert_eq!(error.kind(), StreamInstallReceiveErrorKind::Transport);
        assert!(error.to_string().contains("no frame progress"));
    }

    #[tokio::test]
    async fn progress_ticks_do_not_reset_a_pending_frame_deadline() {
        let cancellation = CancellationToken::new();
        let inactivity = StreamInstallInactivityDeadline::after(Duration::from_millis(35));
        let mut progress_ticks = 0;

        let error = loop {
            match await_stream_install_input(
                std::future::pending::<()>(),
                time::sleep(Duration::from_millis(5)),
                "game",
                &cancellation,
                inactivity,
            )
            .await
            {
                Ok(StreamInstallInput::ProgressTick) => progress_ticks += 1,
                Ok(StreamInstallInput::Frame(())) => {
                    panic!("pending frame future must not complete")
                }
                Err(error) => break error,
            }
        };

        assert!(
            progress_ticks > 1,
            "test must observe periodic progress ticks"
        );
        assert_eq!(error.kind(), StreamInstallReceiveErrorKind::Transport);
        assert!(error.to_string().contains("no frame progress"));
    }

    #[tokio::test]
    async fn post_complete_fin_has_the_same_inactivity_deadline() {
        let cancellation = CancellationToken::new();
        let error = await_stream_install_fin(
            std::future::pending::<Option<Result<bytes::BytesMut, std::io::Error>>>(),
            "game",
            &cancellation,
            StreamInstallInactivityDeadline::after(Duration::from_millis(25)),
        )
        .await
        .expect_err("pending response FIN must time out");

        assert_eq!(error.kind(), StreamInstallReceiveErrorKind::Transport);
        assert!(error.to_string().contains("response FIN"));
    }

    #[tokio::test]
    async fn incoming_request_cancellation_interrupts_a_blocked_send() {
        let write_polls = Arc::new(AtomicUsize::new(0));
        let first_write = Arc::new(Notify::new());
        let mut framed_tx = FramedWrite::new(
            PendingWriter {
                write_polls: write_polls.clone(),
                first_write: first_write.clone(),
            },
            LengthDelimitedCodec::new(),
        );
        let cancel_token = CancellationToken::new();
        let cancellation = tokio::spawn({
            let cancel_token = cancel_token.clone();
            async move {
                first_write.notified().await;
                cancel_token.cancel();
            }
        });

        let error = send_stream_install_request_frame(
            &mut framed_tx,
            "game",
            content_id(),
            &cancel_token,
            StreamInstallInactivityDeadline::ordinary(),
        )
        .await
        .expect_err("blocked request send should observe cancellation");
        cancellation
            .await
            .expect("cancellation helper should finish");

        assert!(error.to_string().contains("sending its request"));
        let polls_after_return = write_polls.load(Ordering::SeqCst);
        assert!(polls_after_return > 0, "test must reach the blocked send");
        tokio::task::yield_now().await;
        assert_eq!(write_polls.load(Ordering::SeqCst), polls_after_return);
    }

    #[tokio::test]
    async fn incoming_request_send_has_an_application_inactivity_deadline() {
        let write_polls = Arc::new(AtomicUsize::new(0));
        let mut framed_tx = FramedWrite::new(
            PendingWriter {
                write_polls: write_polls.clone(),
                first_write: Arc::new(Notify::new()),
            },
            LengthDelimitedCodec::new(),
        );

        let error = send_stream_install_request_frame(
            &mut framed_tx,
            "game",
            content_id(),
            &CancellationToken::new(),
            StreamInstallInactivityDeadline::after(Duration::from_millis(25)),
        )
        .await
        .expect_err("pending request send must time out");

        assert_eq!(error.kind(), StreamInstallReceiveErrorKind::Transport);
        assert!(error.to_string().contains("sending its request"));
        assert!(error.to_string().contains("no frame progress"));
        assert!(write_polls.load(Ordering::SeqCst) > 0);
    }

    #[tokio::test]
    async fn incoming_request_cancellation_interrupts_a_blocked_close() {
        let first_shutdown = Arc::new(Notify::new());
        let mut framed_tx = FramedWrite::new(
            PendingShutdownWriter {
                first_shutdown: first_shutdown.clone(),
            },
            LengthDelimitedCodec::new(),
        );
        let cancel_token = CancellationToken::new();
        let cancellation = tokio::spawn({
            let cancel_token = cancel_token.clone();
            async move {
                first_shutdown.notified().await;
                cancel_token.cancel();
            }
        });

        let error = send_stream_install_request_frame(
            &mut framed_tx,
            "game",
            content_id(),
            &cancel_token,
            StreamInstallInactivityDeadline::ordinary(),
        )
        .await
        .expect_err("blocked request close should observe cancellation");
        cancellation
            .await
            .expect("cancellation helper should finish");

        assert!(error.to_string().contains("closing its request"));
    }

    #[tokio::test]
    async fn incoming_request_fin_has_an_application_inactivity_deadline() {
        let mut framed_tx = FramedWrite::new(
            PendingShutdownWriter {
                first_shutdown: Arc::new(Notify::new()),
            },
            LengthDelimitedCodec::new(),
        );

        let error = send_stream_install_request_frame(
            &mut framed_tx,
            "game",
            content_id(),
            &CancellationToken::new(),
            StreamInstallInactivityDeadline::after(Duration::from_millis(25)),
        )
        .await
        .expect_err("pending request FIN must time out");

        assert_eq!(error.kind(), StreamInstallReceiveErrorKind::Transport);
        assert!(error.to_string().contains("closing its request"));
        assert!(error.to_string().contains("no frame progress"));
    }

    #[tokio::test]
    async fn incoming_request_success_waits_for_its_close_completion() {
        let cancel_token = CancellationToken::new();
        let mut framed_tx = FramedWrite::new(
            CancelOnShutdownWriter {
                cancel_token: cancel_token.clone(),
            },
            LengthDelimitedCodec::new(),
        );

        send_stream_install_request_frame(
            &mut framed_tx,
            "game",
            content_id(),
            &cancel_token,
            StreamInstallInactivityDeadline::ordinary(),
        )
        .await
        .expect("request close that completes in the current poll should succeed");

        assert!(
            cancel_token.is_cancelled(),
            "test writer must have observed the request close"
        );
    }

    #[tokio::test]
    async fn incoming_request_carries_the_exact_catalog_content_id() {
        let expected_content_id = content_id();
        let cancel_token = CancellationToken::new();
        let (writer, reader) = tokio::io::duplex(4_096);
        let mut framed_tx = FramedWrite::new(writer, LengthDelimitedCodec::new());

        send_stream_install_request_frame(
            &mut framed_tx,
            "game",
            expected_content_id,
            &cancel_token,
            StreamInstallInactivityDeadline::ordinary(),
        )
        .await
        .expect("request should send and close");

        let mut framed_rx = FramedRead::new(reader, LengthDelimitedCodec::new());
        let encoded = framed_rx
            .next()
            .await
            .expect("request frame should exist")
            .expect("request frame should decode")
            .freeze();
        assert_eq!(
            <Request as ControlMessage>::decode(encoded)
                .expect("request should pass strict control decoding"),
            Request::StreamInstall {
                game_id: "game".to_owned(),
                content_id: expected_content_id,
            }
        );
        assert!(framed_rx.next().await.is_none(), "request must end at FIN");
    }

    #[tokio::test]
    async fn outbound_cancellation_wins_over_already_queued_frames() {
        let (frame_tx, mut frame_rx) = mpsc::channel(2);
        frame_tx
            .try_send(StreamInstallFrame::Directory {
                relative_path: canonical_path("bin"),
            })
            .expect("first frame should fit");
        frame_tx
            .try_send(StreamInstallFrame::Complete)
            .expect("second frame should fit");

        let cancel_token = CancellationToken::new();
        cancel_token.cancel();
        let mut framed_tx = FramedWrite::new(tokio::io::sink(), LengthDelimitedCodec::new());

        let outcome =
            forward_stream_install_frames(&mut framed_tx, &mut frame_rx, &cancel_token).await;

        assert!(matches!(outcome, StreamInstallEgressOutcome::Cancelled));
        assert_eq!(frame_rx.len(), 2, "cancelled egress must not drain frames");
    }

    #[tokio::test]
    async fn stalled_provider_cannot_hold_stream_install_authority_forever() {
        let (_frame_tx, mut frame_rx) = mpsc::channel(1);
        let mut framed_tx = FramedWrite::new(tokio::io::sink(), LengthDelimitedCodec::new());

        let outcome = forward_stream_install_frames_with_timeout(
            &mut framed_tx,
            &mut frame_rx,
            &CancellationToken::new(),
            Duration::from_millis(25),
        )
        .await;

        let StreamInstallEgressOutcome::Failed(error) = outcome else {
            panic!("stalled producer should hit the application inactivity deadline");
        };
        assert!(
            error
                .to_string()
                .contains("producer timed out after 25ms without a frame")
        );
    }

    #[tokio::test]
    async fn outbound_cancellation_interrupts_a_blocked_frame_write() {
        let (frame_tx, mut frame_rx) = mpsc::channel(1);
        frame_tx
            .try_send(StreamInstallFrame::Complete)
            .expect("frame should fit");

        let write_polls = Arc::new(AtomicUsize::new(0));
        let first_write = Arc::new(Notify::new());
        let mut framed_tx = FramedWrite::new(
            PendingWriter {
                write_polls: write_polls.clone(),
                first_write: first_write.clone(),
            },
            LengthDelimitedCodec::new(),
        );
        let cancel_token = CancellationToken::new();
        let cancellation = tokio::spawn({
            let cancel_token = cancel_token.clone();
            async move {
                first_write.notified().await;
                cancel_token.cancel();
            }
        });

        let outcome =
            forward_stream_install_frames(&mut framed_tx, &mut frame_rx, &cancel_token).await;
        cancellation
            .await
            .expect("cancellation helper should finish");

        assert!(matches!(outcome, StreamInstallEgressOutcome::Cancelled));
        let polls_after_return = write_polls.load(Ordering::SeqCst);
        assert!(polls_after_return > 0, "test must reach the blocked write");
        tokio::task::yield_now().await;
        assert_eq!(
            write_polls.load(Ordering::SeqCst),
            polls_after_return,
            "no write work may remain after egress returns"
        );
    }

    #[tokio::test]
    async fn slow_reader_cannot_hold_stream_install_frame_send_forever() {
        let (frame_tx, mut frame_rx) = mpsc::channel(1);
        frame_tx
            .try_send(StreamInstallFrame::Complete)
            .expect("frame should fit");
        let mut framed_tx = FramedWrite::new(
            PendingWriter {
                write_polls: Arc::new(AtomicUsize::new(0)),
                first_write: Arc::new(Notify::new()),
            },
            LengthDelimitedCodec::new(),
        );

        let outcome = forward_stream_install_frames_with_timeout(
            &mut framed_tx,
            &mut frame_rx,
            &CancellationToken::new(),
            Duration::from_millis(25),
        )
        .await;

        let StreamInstallEgressOutcome::Failed(error) = outcome else {
            panic!("blocked send should hit the application inactivity deadline");
        };
        assert!(
            error
                .to_string()
                .contains("frame send timed out after 25ms")
        );
    }

    #[tokio::test]
    async fn successful_close_is_the_egress_completion_point() {
        let (frame_tx, mut frame_rx) = mpsc::channel(1);
        drop(frame_tx);
        let cancel_token = CancellationToken::new();
        let mut framed_tx = FramedWrite::new(
            CancelOnShutdownWriter {
                cancel_token: cancel_token.clone(),
            },
            LengthDelimitedCodec::new(),
        );

        let outcome =
            forward_stream_install_frames(&mut framed_tx, &mut frame_rx, &cancel_token).await;

        assert!(cancel_token.is_cancelled());
        assert!(matches!(outcome, StreamInstallEgressOutcome::Complete));
    }

    #[tokio::test]
    async fn outbound_cancellation_interrupts_a_blocked_close() {
        let (frame_tx, mut frame_rx) = mpsc::channel(1);
        drop(frame_tx);
        let first_shutdown = Arc::new(Notify::new());
        let mut framed_tx = FramedWrite::new(
            PendingShutdownWriter {
                first_shutdown: first_shutdown.clone(),
            },
            LengthDelimitedCodec::new(),
        );
        let cancel_token = CancellationToken::new();
        let cancellation = tokio::spawn({
            let cancel_token = cancel_token.clone();
            async move {
                first_shutdown.notified().await;
                cancel_token.cancel();
            }
        });

        let outcome =
            forward_stream_install_frames(&mut framed_tx, &mut frame_rx, &cancel_token).await;
        cancellation
            .await
            .expect("cancellation helper should finish");

        assert!(matches!(outcome, StreamInstallEgressOutcome::Cancelled));
    }

    #[tokio::test]
    async fn slow_reader_cannot_hold_stream_install_close_forever() {
        let (frame_tx, mut frame_rx) = mpsc::channel(1);
        drop(frame_tx);
        let mut framed_tx = FramedWrite::new(
            PendingShutdownWriter {
                first_shutdown: Arc::new(Notify::new()),
            },
            LengthDelimitedCodec::new(),
        );

        let outcome = forward_stream_install_frames_with_timeout(
            &mut framed_tx,
            &mut frame_rx,
            &CancellationToken::new(),
            Duration::from_millis(25),
        )
        .await;

        let StreamInstallEgressOutcome::Failed(error) = outcome else {
            panic!("blocked close should hit the application inactivity deadline");
        };
        assert!(error.to_string().contains("close timed out after 25ms"));
    }

    #[test]
    fn stream_paths_stay_inside_staging_dir() {
        let temp = TempDir::new("lanspread-stream-install-path");
        let staging = temp.path().join("staging");
        std::fs::create_dir_all(&staging).expect("staging should be created");
        let staging = std::fs::canonicalize(staging).expect("staging should canonicalize");

        assert!(resolve_stream_path(&staging, &canonical_path("bin/game.exe")).is_ok());
        for invalid in ["../outside", "/absolute", "C:/windows"] {
            assert!(CanonicalCatalogPath::new(invalid).is_err());
        }
    }

    #[test]
    fn catalog_verifier_accepts_exact_output_and_cross_archive_directories() {
        let payload = b"catalog payload";
        let manifest = test_catalog_manifest(
            &["a.eti", "b.eti"],
            vec![
                CatalogExtractedEntry::directory("bin").expect("directory entry should validate"),
                CatalogExtractedEntry::file(
                    "bin/payload.bin",
                    u64::try_from(payload.len()).expect("payload size should fit"),
                    Blake3Digest::hash(payload),
                )
                .expect("file entry should validate"),
            ],
        );
        let mut verifier =
            CatalogStreamVerifier::new("game", manifest).expect("streamed catalog should verify");

        verifier
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("first archive should begin");
        verifier
            .record_directory(&canonical_path("bin"))
            .expect("catalog directory should be accepted");
        verifier
            .end_archive(&canonical_path("a.eti"))
            .expect("first archive should end");
        verifier
            .begin_archive(
                &canonical_path("b.eti"),
                u64::try_from(payload.len()).expect("payload size should fit"),
            )
            .expect("second archive should begin");
        verifier
            .record_directory(&canonical_path("bin"))
            .expect("a directory may recur in another bounded archive");
        assert_eq!(
            verifier
                .begin_file(
                    &canonical_path("bin/payload.bin"),
                    u64::try_from(payload.len()).expect("payload size should fit")
                )
                .expect("catalog file should begin"),
            Blake3Digest::hash(payload)
        );
        verifier
            .record_file_chunk(payload.len())
            .expect("nonempty file chunk should be accepted");
        verifier
            .end_file(&canonical_path("bin/payload.bin"))
            .expect("catalog file should end");
        verifier
            .end_archive(&canonical_path("b.eti"))
            .expect("second archive should end");
        verifier
            .verify_complete()
            .expect("the exact catalog output should complete");
    }

    #[test]
    fn catalog_verifier_rejects_unknown_shape_size_and_duplicate_files() {
        let payload = b"payload";
        let extracted = vec![
            CatalogExtractedEntry::directory("bin").expect("directory should validate"),
            CatalogExtractedEntry::file(
                "bin/payload.bin",
                u64::try_from(payload.len()).expect("payload size should fit"),
                Blake3Digest::hash(payload),
            )
            .expect("file should validate"),
        ];

        let mut unknown = CatalogStreamVerifier::new(
            "game",
            test_catalog_manifest(&["a.eti"], extracted.clone()),
        )
        .expect("manifest should authorize streaming");
        unknown
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        assert!(
            unknown
                .record_directory(&canonical_path("unknown"))
                .expect_err("unknown path must fail")
                .to_string()
                .contains("unknown path")
        );

        let mut wrong_shape = CatalogStreamVerifier::new(
            "game",
            test_catalog_manifest(&["a.eti"], extracted.clone()),
        )
        .expect("manifest should authorize streaming");
        wrong_shape
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        assert!(
            wrong_shape
                .record_directory(&canonical_path("bin/payload.bin"))
                .expect_err("file-as-directory must fail")
                .to_string()
                .contains("catalog expects File")
        );

        let mut wrong_size = CatalogStreamVerifier::new(
            "game",
            test_catalog_manifest(&["a.eti"], extracted.clone()),
        )
        .expect("manifest should authorize streaming");
        wrong_size
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        assert!(
            wrong_size
                .begin_file(&canonical_path("bin/payload.bin"), 1)
                .expect_err("catalog size mismatch must fail")
                .to_string()
                .contains("catalog expects")
        );

        let mut duplicate = CatalogStreamVerifier::new(
            "game",
            test_catalog_manifest(&["a.eti", "b.eti"], extracted),
        )
        .expect("manifest should authorize streaming");
        duplicate
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("first archive should begin");
        duplicate
            .begin_file(
                &canonical_path("bin/payload.bin"),
                u64::try_from(payload.len()).expect("payload size should fit"),
            )
            .expect("first file occurrence should begin");
        duplicate
            .end_file(&canonical_path("bin/payload.bin"))
            .expect("first file occurrence should end");
        duplicate
            .end_archive(&canonical_path("a.eti"))
            .expect("first archive should end");
        duplicate
            .begin_archive(&canonical_path("b.eti"), 0)
            .expect("second archive should begin");
        assert!(
            duplicate
                .begin_file(
                    &canonical_path("bin/payload.bin"),
                    u64::try_from(payload.len()).expect("payload size should fit")
                )
                .expect_err("duplicate file across archives must fail")
                .to_string()
                .contains("repeated file")
        );
    }

    #[test]
    fn catalog_verifier_rejects_missing_and_unbalanced_completion() {
        let payload = b"payload";
        let extracted = vec![
            CatalogExtractedEntry::file(
                "payload.bin",
                u64::try_from(payload.len()).expect("payload size should fit"),
                Blake3Digest::hash(payload),
            )
            .expect("file should validate"),
        ];

        let mut missing = CatalogStreamVerifier::new(
            "game",
            test_catalog_manifest(&["a.eti"], extracted.clone()),
        )
        .expect("manifest should authorize streaming");
        missing
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        missing
            .end_archive(&canonical_path("a.eti"))
            .expect("archive should end");
        assert!(
            missing
                .verify_complete()
                .expect_err("missing catalog file must fail")
                .to_string()
                .contains("missing catalog entry")
        );

        let mut open_archive = CatalogStreamVerifier::new(
            "game",
            test_catalog_manifest(&["a.eti"], extracted.clone()),
        )
        .expect("manifest should authorize streaming");
        open_archive
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        assert!(
            open_archive
                .verify_complete()
                .expect_err("Complete before ArchiveEnd must fail")
                .to_string()
                .contains("before ArchiveEnd")
        );

        let mut open_file =
            CatalogStreamVerifier::new("game", test_catalog_manifest(&["a.eti"], extracted))
                .expect("manifest should authorize streaming");
        assert!(
            open_file
                .begin_archive(&canonical_path("unknown.eti"), 0)
                .expect_err("archive outside the catalog set must fail")
                .to_string()
                .contains("unknown streamed archive")
        );
        open_file
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        assert!(
            open_file
                .record_file_chunk(1)
                .expect_err("chunk before FileBegin must fail")
                .to_string()
                .contains("without FileBegin")
        );
        open_file
            .begin_file(
                &canonical_path("payload.bin"),
                u64::try_from(payload.len()).expect("payload size should fit"),
            )
            .expect("file should begin");
        assert!(
            open_file
                .record_file_chunk(0)
                .expect_err("empty chunks must not make an open file unbounded")
                .to_string()
                .contains("empty FileChunk")
        );
        assert!(
            open_file
                .end_archive(&canonical_path("a.eti"))
                .expect_err("ArchiveEnd with an open file must fail")
                .to_string()
                .contains("while streamed file payload.bin is open")
        );
        assert!(
            open_file
                .verify_complete()
                .expect_err("Complete with an open file must fail")
                .to_string()
                .contains("while streamed file payload.bin is open")
        );
    }

    #[test]
    fn catalog_verifier_requires_stream_support_and_bounds_entry_frames() {
        let unsupported = test_catalog_manifest(&["a.eti"], Vec::new());
        assert!(
            CatalogStreamVerifier::new("game", unsupported)
                .expect_err("missing extracted manifest must disable streaming")
                .to_string()
                .contains("does not support")
        );

        let directory_manifest = || {
            test_catalog_manifest(
                &["a.eti"],
                vec![CatalogExtractedEntry::directory("bin").expect("directory should validate")],
            )
        };
        let mut repeated = CatalogStreamVerifier::new("game", directory_manifest())
            .expect("manifest should authorize streaming");
        repeated
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        repeated
            .record_directory(&canonical_path("bin"))
            .expect("first directory frame should be accepted");
        repeated
            .record_directory(&canonical_path("bin"))
            .expect("a repeated expected directory should be accepted within the frame bound");
        repeated
            .end_archive(&canonical_path("a.eti"))
            .expect("archive should end after repeated directories");
        repeated
            .verify_complete()
            .expect("repeated expected directories should still verify exactly");

        let mut bounded = CatalogStreamVerifier::new("game", directory_manifest())
            .expect("manifest should authorize streaming");
        bounded
            .begin_archive(&canonical_path("a.eti"), 0)
            .expect("archive should begin");
        bounded.entry_frames = MAX_CATALOG_ENTRIES;
        assert!(
            bounded
                .record_directory(&canonical_path("bin"))
                .expect_err("entry frames beyond the catalog bound must fail")
                .to_string()
                .contains("entry frame limit")
        );

        let mut telemetry = CatalogStreamVerifier::new(
            "game",
            test_catalog_manifest(
                &["a.eti", "b.eti"],
                vec![CatalogExtractedEntry::directory("bin").expect("directory should validate")],
            ),
        )
        .expect("manifest should authorize streaming");
        telemetry
            .begin_archive(&canonical_path("a.eti"), MAX_CATALOG_TOTAL_BYTES)
            .expect("reported total at the bound should be accepted");
        telemetry
            .end_archive(&canonical_path("a.eti"))
            .expect("first archive should end");
        assert!(
            telemetry
                .begin_archive(&canonical_path("b.eti"), 1)
                .expect_err("aggregate sender telemetry must remain bounded")
                .to_string()
                .contains("reported unpacked limit")
        );
    }

    #[test]
    fn streamed_install_progress_total_is_catalog_owned() {
        let payload = b"catalog-sized payload";
        let manifest = test_catalog_manifest(
            &["a.eti"],
            vec![
                CatalogExtractedEntry::file(
                    "payload.bin",
                    u64::try_from(payload.len()).expect("payload size should fit"),
                    Blake3Digest::hash(payload),
                )
                .expect("file entry should validate"),
            ],
        );
        let mut verifier = CatalogStreamVerifier::new("game", manifest)
            .expect("manifest should authorize streaming");
        verifier
            .begin_archive(&canonical_path("a.eti"), 1)
            .expect("sender telemetry need not equal the catalog-owned total");

        let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
        let status = crate::transfer_status::DownloadAttemptStatus::new(
            crate::DownloadAttemptKey::next("game".to_owned()),
            CancellationToken::new(),
            tx,
        );
        let progress =
            StreamInstallProgress::new(status.reporter(), verifier.expected_file_bytes());
        assert_eq!(
            progress.total_bytes,
            u64::try_from(payload.len()).expect("payload size should fit")
        );
    }

    #[test]
    fn receive_boundary_classifies_disconnects_without_quarantining_them() {
        assert_eq!(
            stream_ended_before_complete("game").kind(),
            StreamInstallReceiveErrorKind::Transport
        );

        let cancel_token = CancellationToken::new();
        let read_error = StreamInstallReceiveError::transport_or_cancelled(
            std::io::Error::new(std::io::ErrorKind::ConnectionReset, "peer disconnected"),
            "game",
            &cancel_token,
            "while reading its response",
        );
        assert_eq!(read_error.kind(), StreamInstallReceiveErrorKind::Transport);

        assert_eq!(
            data_after_complete("game").kind(),
            StreamInstallReceiveErrorKind::Integrity
        );

        let malformed_path =
            Bytes::from_static(b"\0{\"Directory\":{\"relative_path\":\"../escape\"}}");
        assert_eq!(
            decode_received_stream_install_frame(malformed_path)
                .expect_err("malformed typed path must fail strict decoding")
                .kind(),
            StreamInstallReceiveErrorKind::Integrity
        );

        let explicit_error = StreamInstallFrame::Error {
            message: "sender failed".to_owned(),
        };
        assert_eq!(
            decode_received_stream_install_frame(explicit_error.encode())
                .expect("explicit sender Error must remain a valid frame"),
            explicit_error
        );
    }

    #[tokio::test]
    async fn oversized_length_delimited_frame_is_an_integrity_failure() {
        use tokio::io::AsyncWriteExt as _;

        let (mut writer, reader) = tokio::io::duplex(16);
        let oversized = u32::try_from(MAX_STREAM_INSTALL_FRAME_BYTES + 1)
            .expect("stream frame bound should fit u32");
        writer
            .write_all(&oversized.to_be_bytes())
            .await
            .expect("oversized length prefix should write");
        drop(writer);

        let mut framed_rx = FramedRead::new(
            reader,
            LengthDelimitedCodec::builder()
                .max_frame_length(MAX_STREAM_INSTALL_FRAME_BYTES)
                .new_codec(),
        );
        let codec_error = framed_rx
            .next()
            .await
            .expect("oversized prefix should produce a codec result")
            .expect_err("oversized frame must fail before allocation");
        assert_eq!(codec_error.kind(), std::io::ErrorKind::InvalidData);

        let cancellation = CancellationToken::new();
        let classified = classify_stream_install_read_error(
            codec_error,
            "game",
            &cancellation,
            "while reading its response",
        );
        assert_eq!(classified.kind(), StreamInstallReceiveErrorKind::Integrity);
    }

    #[test]
    fn incoming_file_requires_catalog_blake3_after_sender_crc_passes() {
        let payload = b"payload";
        let endpoint = peer_endpoint();
        let exact_content_id = content_id();
        let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
        let mut accepted = IncomingFile::new(
            canonical_path("accepted.bin"),
            PathBuf::from("accepted.bin"),
            u64::try_from(payload.len()).expect("payload size should fit"),
            crc32_of(payload),
            Blake3Digest::hash(payload),
            Vec::new(),
        );
        accepted
            .write_chunk(payload)
            .expect("payload write should succeed");
        assert!(
            event_rx.try_recv().is_err(),
            "unverified streamed bytes must not emit a completion event"
        );
        accepted
            .finish(
                &canonical_path("accepted.bin"),
                "game",
                endpoint,
                exact_content_id,
                &event_tx,
            )
            .expect("matching catalog hash should finish");
        assert!(matches!(
            event_rx.try_recv(),
            Ok(PeerEvent::DownloadGameFileChunkFinished {
                peer_id,
                content_id,
                relative_path,
                offset: 0,
                length,
                ..
            }) if peer_id == endpoint.peer_id
                && content_id == exact_content_id
                && relative_path.as_str() == "accepted.bin"
                && length == u64::try_from(payload.len()).expect("payload length should fit")
        ));

        let mut incoming = IncomingFile::new(
            canonical_path("payload.bin"),
            PathBuf::from("payload.bin"),
            u64::try_from(payload.len()).expect("payload size should fit"),
            crc32_of(payload),
            Blake3Digest::hash(b"different trusted bytes"),
            Vec::new(),
        );
        incoming
            .write_chunk(payload)
            .expect("payload write should succeed");

        let error = incoming
            .finish(
                &canonical_path("payload.bin"),
                "game",
                endpoint,
                exact_content_id,
                &event_tx,
            )
            .expect_err("catalog digest mismatch must fail after matching sender CRC");
        assert_eq!(error.kind(), StreamInstallReceiveErrorKind::Integrity);
        assert!(error.to_string().contains("catalog BLAKE3 mismatch"));
        assert!(
            event_rx.try_recv().is_err(),
            "catalog-rejected bytes must never emit completion"
        );
    }

    fn test_catalog_manifest(
        archives: &[&str],
        streamed_install_files: Vec<CatalogExtractedEntry>,
    ) -> Arc<CatalogContentManifest> {
        let archive_digest = Blake3Digest::hash(b"archive");
        let mut files = archives
            .iter()
            .map(|archive| {
                CatalogFileEntry::file(*archive, 7, archive_digest, vec![archive_digest])
                    .expect("archive entry should validate")
            })
            .collect::<Vec<_>>();
        let version = b"20240101";
        let version_digest = Blake3Digest::hash(version);
        files.push(
            CatalogFileEntry::file(
                "version.ini",
                u64::try_from(version.len()).expect("version size should fit"),
                version_digest,
                vec![version_digest],
            )
            .expect("version entry should validate"),
        );
        files.sort_by(|left, right| left.canonical_path().cmp(right.canonical_path()));

        Arc::new(
            CatalogContentManifest::seal(
                CatalogContentManifestBody::new("game", "20240101", files, streamed_install_files)
                    .expect("manifest body should validate"),
            )
            .expect("manifest should seal"),
        )
    }

    #[test]
    fn sender_uses_only_the_exact_catalog_archive_set() {
        let temp = TempDir::new("lanspread-stream-install-sender-archives");
        std::fs::write(temp.path().join("a.eti"), b"archive")
            .expect("catalog archive should be written");
        let extracted = vec![
            CatalogExtractedEntry::file("payload.bin", 1, Blake3Digest::hash(b"x"))
                .expect("extracted file should validate"),
        ];
        let manifest = test_catalog_manifest(&["a.eti"], extracted.clone());

        assert_eq!(
            catalog_stream_archives(temp.path(), "game", &manifest)
                .expect("the exact catalog archive set should be admitted"),
            vec![temp.path().join("a.eti")]
        );

        std::fs::write(temp.path().join("b.eti"), b"extra")
            .expect("extra archive should be written");
        assert!(
            catalog_stream_archives(temp.path(), "game", &manifest).is_err(),
            "a root archive outside the catalog must never be streamed"
        );

        let unsupported = test_catalog_manifest(&["a.eti", "b.eti"], Vec::new());
        assert!(
            catalog_stream_archives(temp.path(), "game", &unsupported).is_err(),
            "a game without verified extracted output must not offer Stream Install"
        );
    }

    #[test]
    fn archive_catalog_name_never_fabricates_a_missing_name() {
        let error = archive_catalog_name(Path::new("/"))
            .expect_err("a path without a file name must be rejected");
        assert!(error.to_string().contains("has no file name"));
    }

    #[cfg(unix)]
    #[test]
    fn archive_catalog_name_rejects_non_utf8_before_provider_work() {
        use std::{ffi::OsString, os::unix::ffi::OsStringExt as _};

        let archive = PathBuf::from(OsString::from_vec(vec![b'a', 0xff, b'.', b'e', b't', b'i']));
        let error =
            archive_catalog_name(&archive).expect_err("a non-UTF-8 archive name must be rejected");
        assert!(error.to_string().contains("not valid UTF-8"));
    }

    #[test]
    fn parses_unrar_technical_listing() {
        let listing = parse_unrar_listing(
            r"
Archive: game.eti
Details: RAR 5, solid

        Name: bin/payload.bin
        Type: File
        Size: 123
       CRC32: 38B488A7

        Name: bin
        Type: Directory
",
        )
        .expect("listing should parse");

        assert!(listing.solid);
        assert_eq!(
            listing.entries,
            vec![
                RarEntry {
                    relative_path: canonical_path("bin/payload.bin"),
                    kind: RarEntryKind::File,
                    size: 123,
                    crc32: Some(0x38B4_88A7),
                },
                RarEntry {
                    relative_path: canonical_path("bin"),
                    kind: RarEntryKind::Directory,
                    size: 0,
                    crc32: None,
                },
            ]
        );
    }

    #[test]
    fn truncated_unrar_listing_metadata_fails_closed() {
        let error = reject_truncated_unrar_listing(true, false, Path::new("large.eti"))
            .expect_err("truncated listing stdout must not authorize a partial manifest");

        assert!(error.to_string().contains("metadata limit"));
    }

    #[test]
    fn rejects_unrar_file_entries_without_crc32() {
        let err = parse_unrar_listing(
            r"
Archive: game.eti
Details: RAR 5

        Name: bin/payload.bin
        Type: File
        Size: 123
",
        )
        .expect_err("file entries without CRC32 should be rejected");

        assert!(err.to_string().contains("has no CRC32"));
    }

    #[test]
    fn accepts_zero_size_unrar_file_entries_without_crc32() {
        let listing = parse_unrar_listing(
            r"
Archive: game.eti
Details: RAR 5

        Name: bin/empty.cfg
        Type: File
        Size: 0
",
        )
        .expect("empty file without CRC32 should parse as CRC32 zero");

        assert_eq!(
            listing.entries,
            vec![RarEntry {
                relative_path: canonical_path("bin/empty.cfg"),
                kind: RarEntryKind::File,
                size: 0,
                crc32: Some(0),
            }]
        );
    }

    #[test]
    fn sender_archive_integrity_accepts_matching_size_and_crc32() {
        let bytes = b"payload";
        let byte_len = u64::try_from(bytes.len()).expect("test payload length should fit in u64");
        let integrity = SenderArchiveIntegrity::new(byte_len, crc32_of(bytes));

        integrity
            .verify(
                &canonical_path("bin/payload.bin"),
                byte_len,
                crc32_of(bytes),
            )
            .expect("matching sender archive metadata should verify");
    }

    #[test]
    fn sender_archive_integrity_rejects_size_mismatch() {
        let integrity = SenderArchiveIntegrity::new(7, crc32_of(b"payload"));
        let err = integrity
            .verify(&canonical_path("bin/payload.bin"), 6, crc32_of(b"payload"))
            .expect_err("truncated file should fail sender archive integrity");

        assert!(err.to_string().contains("size mismatch"));
    }

    #[test]
    fn sender_archive_integrity_rejects_crc32_mismatch() {
        let integrity = SenderArchiveIntegrity::new(7, crc32_of(b"payload"));
        let err = integrity
            .verify(&canonical_path("bin/payload.bin"), 7, crc32_of(b"paylord"))
            .expect_err("mutated file should fail sender archive integrity");

        assert!(err.to_string().contains("sender RAR CRC32 mismatch"));
    }

    fn crc32_of(bytes: &[u8]) -> u32 {
        let mut hasher = Hasher::new();
        hasher.update(bytes);
        hasher.finalize()
    }
}