wireshift-fallback 0.1.1

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use wireshift_core::op::{CompletionPayload, OpDescriptor};
use wireshift_core::{Error, Result};

use crate::ops::execute_descriptor_with_fixed;

pub(crate) fn execute_close_fixed(
    slot: u32,
    fixed_files: &mut HashMap<u32, std::fs::File>,
) -> Result<CompletionPayload> {
    let _file = fixed_files.remove(&slot).ok_or_else(|| {
        Error::completion(
            format!("direct descriptor slot {slot} was closed before open"),
            "ensure linked reads only close direct descriptors that were opened successfully",
        )
    })?;
    Ok(CompletionPayload::Unit)
}

pub(crate) fn execute_statx(path: PathBuf) -> Result<CompletionPayload> {
    let metadata = std::fs::metadata(&path).map_err(|error| {
        Error::io(
            "statx failed",
            error,
            "ensure the path exists and is accessible",
        )
    })?;
    Ok(CompletionPayload::Metadata {
        size: metadata.len(),
        is_file: metadata.is_file(),
        is_dir: metadata.is_dir(),
    })
}

pub(crate) fn execute_fsync(file: std::fs::File) -> Result<CompletionPayload> {
    file.sync_all().map_err(|error| {
        Error::io(
            "fsync failed",
            error,
            "ensure the file descriptor remains valid for synchronization",
        )
    })?;
    Ok(CompletionPayload::Unit)
}

pub(crate) fn execute_splice(
    fd_in: std::fs::File,
    off_in: Option<u64>,
    fd_out: std::fs::File,
    _off_out: Option<u64>,
    len: usize,
) -> Result<CompletionPayload> {
    // On Linux, use splice(2) for zero-copy transfer.
    // On other platforms, fall back to read+write.
    #[cfg(target_os = "linux")]
    {
        use rustix::pipe::{splice, SpliceFlags};

        let mut off_in_val = off_in;
        let mut off_out_val = _off_out;

        let result = splice(
            &fd_in,
            off_in_val.as_mut(),
            &fd_out,
            off_out_val.as_mut(),
            len,
            SpliceFlags::MOVE | SpliceFlags::NONBLOCK,
        )
        .map_err(|error| {
            Error::io(
                "splice failed",
                std::io::Error::from(error),
                "ensure at least one fd is a pipe and both fds are valid",
            )
        })?;

        Ok(CompletionPayload::Bytes(result))
    }

    #[cfg(not(target_os = "linux"))]
    {
        let total = splice_fallback_copy(fd_in, off_in, fd_out, len)?;
        Ok(CompletionPayload::Bytes(total))
    }
}

/// Read+write fallback for `execute_splice` on non-Linux platforms.
///
/// Fails CLOSED (Law 10): when an input offset is requested the seek MUST
/// succeed - the previous `let _ = fd_in.seek(..)` discarded a seek failure and
/// then read from the *current* offset, silently producing corrupt splice
/// output. Propagating the seek error surfaces the failure loudly.
///
/// Generic over the reader/writer (rather than hard-wired to `std::fs::File`)
/// so the seek-failure branch is unit-testable on any host, including Linux
/// where the `not(target_os = "linux")` call site above is not compiled.
#[cfg(any(not(target_os = "linux"), test))]
fn splice_fallback_copy<I, O>(
    mut fd_in: I,
    off_in: Option<u64>,
    mut fd_out: O,
    len: usize,
) -> Result<usize>
where
    I: std::io::Read + std::io::Seek,
    O: std::io::Write,
{
    use std::io::SeekFrom;
    if let Some(off) = off_in {
        fd_in.seek(SeekFrom::Start(off)).map_err(|e| {
            Error::io(
                "splice fallback seek failed",
                e,
                "ensure the source fd is seekable when an input offset is provided",
            )
        })?;
    }
    let mut buf = vec![0u8; len.min(65_536)];
    let mut total = 0;
    while total < len {
        let to_read = (len - total).min(buf.len());
        let n = fd_in
            .read(&mut buf[..to_read])
            .map_err(|e| Error::io("splice fallback read failed", e, "check source fd"))?;
        if n == 0 {
            break;
        }
        fd_out
            .write_all(&buf[..n])
            .map_err(|e| Error::io("splice fallback write failed", e, "check dest fd"))?;
        total += n;
    }
    Ok(total)
}

#[cfg(test)]
mod splice_fallback_tests {
    use super::splice_fallback_copy;
    use std::io::{Cursor, Read, Seek, SeekFrom};

    /// A reader that is `Seek` but whose `seek` always fails - models a
    /// non-seekable/streaming source fd for which an input offset is invalid.
    struct SeekFailReader(Cursor<Vec<u8>>);
    impl Read for SeekFailReader {
        fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
            self.0.read(buf)
        }
    }
    impl Seek for SeekFailReader {
        fn seek(&mut self, _pos: SeekFrom) -> std::io::Result<u64> {
            Err(std::io::Error::new(
                std::io::ErrorKind::Unsupported,
                "underlying fd is not seekable",
            ))
        }
    }

    #[test]
    fn seek_failure_with_input_offset_is_surfaced_not_silently_ignored() {
        // With an input offset requested, a failing seek must abort with a loud
        // error rather than silently reading from the current offset (which
        // corrupts the splice output). Pins the Law 10 fail-closed fix.
        let src = SeekFailReader(Cursor::new(b"payload-bytes".to_vec()));
        let mut sink: Vec<u8> = Vec::new();
        let result = splice_fallback_copy(src, Some(4), &mut sink, 13);
        let err = result.expect_err("failing seek must surface as an error");
        assert!(
            err.to_string().to_lowercase().contains("seek"),
            "error must identify the seek failure, got: {err}"
        );
        assert!(
            sink.is_empty(),
            "no bytes may be written after a seek failure (output must not be corrupted)"
        );
    }

    #[test]
    fn successful_seek_copies_from_the_requested_offset() {
        // Guard against over-failing: a seekable source with an offset copies
        // exactly the bytes from that offset onward.
        let src = Cursor::new(b"0123456789".to_vec());
        let mut sink: Vec<u8> = Vec::new();
        let copied = splice_fallback_copy(src, Some(3), &mut sink, 7)
            .expect("seekable source must copy successfully");
        assert_eq!(copied, 7, "must copy the requested 7 bytes from offset 3");
        assert_eq!(sink, b"3456789", "bytes must come from the requested offset");
    }

    #[test]
    fn no_offset_copies_from_current_position() {
        // Without an offset, no seek occurs and the whole buffer is copied.
        let src = Cursor::new(b"abcdef".to_vec());
        let mut sink: Vec<u8> = Vec::new();
        let copied = splice_fallback_copy(src, None, &mut sink, 6)
            .expect("no-offset copy must succeed");
        assert_eq!(copied, 6);
        assert_eq!(sink, b"abcdef");
    }
}

pub(crate) fn execute_madvise(
    file: std::fs::File,
    advice: wireshift_core::op::MadviseAdvice,
) -> Result<CompletionPayload> {
    #[cfg(target_os = "linux")]
    {
        let metadata = file
            .metadata()
            .map_err(|e| Error::io("madvise stat failed", e, "ensure file is valid"))?;
        let len = metadata.len() as usize;
        if len == 0 {
            return Ok(CompletionPayload::Bytes(0));
        }

        let advice_val = match advice {
            wireshift_core::op::MadviseAdvice::Sequential => rustix::fs::Advice::Sequential,
            wireshift_core::op::MadviseAdvice::Random => rustix::fs::Advice::Random,
            wireshift_core::op::MadviseAdvice::DontNeed => rustix::fs::Advice::DontNeed,
            wireshift_core::op::MadviseAdvice::Normal => rustix::fs::Advice::Normal,
            _ => rustix::fs::Advice::Normal,
        };

        rustix::fs::fadvise(&file, 0, std::num::NonZeroU64::new(len as u64), advice_val).map_err(|error| {
            Error::io(
                "posix_fadvise failed",
                std::io::Error::from(error),
                "ensure the fd is valid and the kernel supports fadvise",
            )
        })?;

        let advice_code = match advice {
            wireshift_core::op::MadviseAdvice::Sequential => 0,
            wireshift_core::op::MadviseAdvice::Random => 1,
            wireshift_core::op::MadviseAdvice::DontNeed => 2,
            wireshift_core::op::MadviseAdvice::Normal => 3,
            _ => 3,
        };

        Ok(CompletionPayload::Bytes(advice_code))
    }

    #[cfg(not(target_os = "linux"))]
    {
        let _ = (file, advice);
        Ok(CompletionPayload::Bytes(3)) // Normal
    }
}

pub(crate) fn execute_linked(
    descriptors: Vec<OpDescriptor>,
    canceled: Arc<AtomicBool>,
    timeout: Option<std::time::Duration>,
) -> Result<CompletionPayload> {
    let mut fixed_files = HashMap::new();
    let mut results = Vec::with_capacity(descriptors.len());
    for descriptor in descriptors {
        results.push(execute_descriptor_with_fixed(
            descriptor,
            canceled.clone(),
            &mut fixed_files,
            timeout,
        )?);
    }
    drop(fixed_files);
    Ok(CompletionPayload::LinkedChain(results))
}