par2-rs 0.4.1

use std::collections::HashSet;
use std::fs::{self, File, Metadata};
use std::io::{self, Read, Seek, SeekFrom};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

use rayon::prelude::*;

use crate::checksum::{self, FileHashState, Md5State, SliceChecksumState};
use crate::error::{Par2Error, Result};
use crate::md5_simd;
use crate::path::translate_par2_name_to_relative;
use crate::types::{
    CancellationToken, FileId, MAX_FILES_PER_SET, MAX_SLICES_PER_FILE, MAX_TOTAL_INPUT_SLICES,
    ProgressCallback, ProgressStage, SliceChecksum,
};

use super::encode::ForwardSourceProvider;

const FIRST_HASH_BYTES: u64 = 16 * 1024;
const READ_BUFFER_BYTES: usize = 256 * 1024;
const MAX_PAR2_NAME_BYTES: usize = 100_000;
/// Staging budget for one file's multi-buffer slice-hash batch.
///
/// Matches the verifier's and the repair scanner's equivalent budgets so the
/// three hashing paths admit the same per-task working set. `plan.rs` accounts
/// this exact number, so the two must move together.
pub(crate) const CREATE_MD5_BATCH_MEMORY_BYTES: usize = 4 * 1024 * 1024;

/// How many consecutive slices of one file to hash per multi-buffer MD5 call.
///
/// Consecutive slices are independent messages, so they lane directly. The
/// width is the narrower of what the kernel offers ([`md5_simd::max_lanes`]:
/// 8 on AVX2, 4 on NEON/SSE2/simd128, 1 scalar) and what the staging budget
/// affords. A block size large enough to leave only one lane falls back to the
/// streaming scan, which needs just one `READ_BUFFER_BYTES` chunk.
pub(crate) fn create_md5_batch_lanes(block_size: usize) -> usize {
    if block_size == 0 {
        return 1;
    }
    (CREATE_MD5_BATCH_MEMORY_BYTES / block_size).clamp(1, md5_simd::max_lanes())
}

/// A validated explicit source file and the metadata needed by critical PAR2
/// packets.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CreationSource {
    /// Canonical path read during creation.
    pub path: PathBuf,
    /// Safe path recorded in PAR2 packets, using `/` separators.
    pub par2_name: String,
    /// PAR2 file identifier.
    pub file_id: FileId,
    /// Source length in bytes.
    pub file_length: u64,
    /// MD5 of the complete source file.
    pub hash_full: [u8; 16],
    /// MD5 of the first 16KiB, or the complete file when shorter.
    pub hash_16k: [u8; 16],
    /// Zero-padded per-slice CRC32 and MD5 pairs.
    pub slice_checksums: Vec<SliceChecksum>,
}

impl CreationSource {
    /// Number of source slices represented by this source.
    pub fn slice_count(&self) -> u32 {
        self.slice_checksums.len() as u32
    }
}

#[derive(Debug, Clone)]
pub(crate) struct InputLength {
    pub length: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct SourceFingerprint {
    length: u64,
    modified: Option<std::time::SystemTime>,
    #[cfg(unix)]
    device: u64,
    #[cfg(unix)]
    inode: u64,
}

/// Per-creator memo of the source scan, so one `Par2Creator` reads and hashes
/// its inputs once instead of once per `build_plan` call.
///
/// `Par2Creator::create` rebuilds the plan from the current inputs and compares
/// it with the caller's, which means the whole input set is otherwise scanned
/// twice for one creation: once in `plan()` and once in `create()`. An entry is
/// only reused when a fresh `stat` of the same canonical path still yields the
/// identical [`SourceFingerprint`] (length, mtime, and on unix device+inode)
/// *and* the same block size, so what changes is which read the hashes come
/// from, not whether the inputs are re-validated. A file that moved, was
/// replaced, was rewritten in place, or whose length changed misses the memo
/// and is rehashed exactly as before.
///
/// Entries are removed as they are used: the memory plan admits two live copies
/// of the source metadata (`source_metadata_bytes * 2`) for the create phase,
/// and draining is what keeps that true while the rebuilt plan is assembled.
pub(crate) struct SourceScanCache {
    entries: std::sync::Mutex<Vec<CachedScan>>,
}

struct CachedScan {
    fingerprint: SourceFingerprint,
    block_size: u64,
    source: CreationSource,
}

impl SourceScanCache {
    pub(crate) fn new() -> Self {
        Self {
            entries: std::sync::Mutex::new(Vec::new()),
        }
    }

    fn take(
        &self,
        path: &Path,
        fingerprint: &SourceFingerprint,
        block_size: u64,
    ) -> Option<CreationSource> {
        let mut entries = self
            .entries
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let index = entries.iter().position(|entry| {
            entry.block_size == block_size
                && entry.fingerprint == *fingerprint
                && entry.source.path == path
        })?;
        Some(entries.swap_remove(index).source)
    }

    fn store(&self, fingerprint: SourceFingerprint, block_size: u64, source: &CreationSource) {
        let mut entries = self
            .entries
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        entries.retain(|entry| entry.source.path != source.path);
        entries.push(CachedScan {
            fingerprint,
            block_size,
            source: source.clone(),
        });
    }
}

impl SourceFingerprint {
    fn from_metadata(metadata: &Metadata) -> Self {
        #[cfg(unix)]
        use std::os::unix::fs::MetadataExt;

        Self {
            length: metadata.len(),
            modified: metadata.modified().ok(),
            #[cfg(unix)]
            device: metadata.dev(),
            #[cfg(unix)]
            inode: metadata.ino(),
        }
    }
}

/// Resolve, validate, and hash explicit source files in input order.
pub(crate) fn collect_sources(
    base_path: &Path,
    inputs: &[PathBuf],
    block_size: u64,
    cancellation: &CancellationToken,
    progress: Option<&ProgressCallback>,
    total_bytes: u64,
    cache: Option<&SourceScanCache>,
) -> Result<Vec<CreationSource>> {
    if inputs.is_empty() {
        return Err(Par2Error::InvalidCreationOptions {
            reason: "at least one source file is required".to_string(),
        });
    }
    let base = fs::canonicalize(base_path).map_err(Par2Error::Io)?;
    let metadata = fs::metadata(&base).map_err(Par2Error::Io)?;
    if !metadata.is_dir() {
        return Err(Par2Error::UnsafeCreationSource {
            path: base.display().to_string(),
            reason: "base path is not a directory".to_string(),
        });
    }

    let mut active_inputs = Vec::with_capacity(inputs.len());
    for input in inputs {
        if cancellation.is_cancelled() {
            return Err(Par2Error::Cancelled);
        }
        let path = resolve_input_path(&base, input)?;
        let metadata = fs::metadata(&path).map_err(Par2Error::Io)?;
        if !metadata.is_file() {
            return Err(Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "source is not a regular file".to_string(),
            });
        }
        let relative = path
            .strip_prefix(&base)
            .map_err(|_| Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "source is outside the base directory".to_string(),
            })?;
        validate_relative_path(relative, input)?;
        if relative.to_str().is_none() {
            return Err(Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "source filename is not valid UTF-8".to_string(),
            });
        }
        if metadata.len() > 0 {
            active_inputs.push(input);
        }
    }
    if active_inputs.is_empty() {
        return Err(Par2Error::InvalidCreationOptions {
            reason: "at least one non-empty source file is required".to_string(),
        });
    }
    validate_main_file_count(active_inputs.len())?;

    let mut names = HashSet::with_capacity(active_inputs.len());
    let mut ids = HashSet::with_capacity(active_inputs.len());
    let bytes_processed = AtomicU64::new(0);
    let file_total =
        u32::try_from(active_inputs.len()).map_err(|_| Par2Error::ResourceLimitExceeded {
            reason: "source file count exceeds the supported progress range".to_string(),
        })?;
    let mut total_slices = 0usize;

    let hash_one = |(index, input): (usize, &&PathBuf)| -> Result<CreationSource> {
        if cancellation.is_cancelled() {
            return Err(Par2Error::Cancelled);
        }
        resolve_and_hash_source(
            &base,
            input,
            block_size,
            cancellation,
            progress,
            index as u32,
            file_total,
            &bytes_processed,
            total_bytes,
            cache,
        )
    };
    // Hash one file per rayon task; each file's scan is independent. The
    // shared byte counter is monotonic, but callbacks fire concurrently, so
    // DELIVERY to the progress callback is not ordered. The sequential arm
    // is the single-threaded-wasm path (no worker pool, matching the crate's
    // other guards; `wasm32-wasip1-threads` probes `true` and hashes in
    // parallel like native)
    // and the WEAVER_PAR2_CREATE_THREADS=1 pre-banding escape hatch.
    let scan_parallel = reedsolomon_rs::threading::parallel_enabled()
        && active_inputs.len() > 1
        && super::encode::configured_create_threads() != 1;
    let scanned: Vec<CreationSource> = if scan_parallel {
        // Collect every per-file Result, then surface the FIRST error by
        // input order: rayon's Result collection reports an arbitrary
        // racer's error, which would let a concurrent I/O error mask
        // `Cancelled` (or make the reported path vary run to run). The
        // deliberate cost is losing error short-circuiting — in-flight
        // files hash to completion before the error surfaces.
        active_inputs
            .par_iter()
            .enumerate()
            .map(hash_one)
            .collect::<Vec<Result<_>>>()
            .into_iter()
            .collect::<Result<Vec<_>>>()?
    } else {
        active_inputs
            .iter()
            .enumerate()
            .map(hash_one)
            .collect::<Result<Vec<_>>>()?
    };
    // The creation memory plan accounts this vector by LENGTH (plan.rs passes
    // `sources.len()` into `estimate_source_metadata_bytes`), so a collect
    // through a Result adapter cannot skew the estimate. The exact-capacity
    // rebuild stays so the allocation itself matches what the plan admits.
    let mut sources = Vec::with_capacity(active_inputs.len());
    sources.extend(scanned);

    for (source, input) in sources.iter().zip(active_inputs.iter()) {
        if !names.insert(source.par2_name.clone()) {
            return Err(Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "duplicate relative PAR2 name".to_string(),
            });
        }
        if !ids.insert(source.file_id) {
            return Err(Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "duplicate PAR2 file identifier".to_string(),
            });
        }
        total_slices = total_slices
            .checked_add(source.slice_checksums.len())
            .ok_or_else(|| Par2Error::ResourceLimitExceeded {
                reason: "total source slice count overflows".to_string(),
            })?;
        if total_slices > MAX_TOTAL_INPUT_SLICES {
            return Err(Par2Error::ResourceLimitExceeded {
                reason: format!(
                    "total source slice count {total_slices} exceeds {MAX_TOTAL_INPUT_SLICES}"
                ),
            });
        }
    }
    Ok(sources)
}

/// Resolve explicit inputs and collect their lengths before selecting a slice size.
pub(crate) fn collect_input_lengths(
    base_path: &Path,
    inputs: &[PathBuf],
    cancellation: &CancellationToken,
) -> Result<Vec<InputLength>> {
    if inputs.is_empty() {
        return Err(Par2Error::InvalidCreationOptions {
            reason: "at least one source file is required".to_string(),
        });
    }
    let base = fs::canonicalize(base_path).map_err(Par2Error::Io)?;
    let base_metadata = fs::metadata(&base).map_err(Par2Error::Io)?;
    if !base_metadata.is_dir() {
        return Err(Par2Error::UnsafeCreationSource {
            path: base.display().to_string(),
            reason: "base path is not a directory".to_string(),
        });
    }

    let mut lengths = Vec::with_capacity(inputs.len());
    for input in inputs {
        if cancellation.is_cancelled() {
            return Err(Par2Error::Cancelled);
        }
        let path = resolve_input_path(&base, input)?;
        let metadata = fs::metadata(&path).map_err(Par2Error::Io)?;
        if !metadata.is_file() {
            return Err(Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "source is not a regular file".to_string(),
            });
        }
        let relative = path
            .strip_prefix(&base)
            .map_err(|_| Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "source is outside the base directory".to_string(),
            })?;
        validate_relative_path(relative, input)?;
        if relative.to_str().is_none() {
            return Err(Par2Error::UnsafeCreationSource {
                path: input.display().to_string(),
                reason: "source filename is not valid UTF-8".to_string(),
            });
        }
        if metadata.len() > 0 {
            lengths.push(InputLength {
                length: metadata.len(),
            });
        }
    }
    validate_main_file_count(lengths.len())?;
    Ok(lengths)
}

fn validate_main_file_count(file_count: usize) -> Result<()> {
    if file_count > MAX_FILES_PER_SET {
        return Err(Par2Error::ResourceLimitExceeded {
            reason: format!(
                "source file count {file_count} exceeds the Main packet limit of {MAX_FILES_PER_SET}"
            ),
        });
    }
    Ok(())
}

/// Seek/read source slices for one forward-encoding pass.
pub(crate) struct DiskSourceProvider<'a> {
    sources: &'a [CreationSource],
    files: Vec<File>,
    fingerprints: Vec<SourceFingerprint>,
    slice_starts: Vec<usize>,
    slice_size: usize,
    cancellation: &'a CancellationToken,
}

impl<'a> DiskSourceProvider<'a> {
    pub(crate) fn open(
        sources: &'a [CreationSource],
        slice_size: usize,
        cancellation: &'a CancellationToken,
    ) -> Result<Self> {
        if slice_size == 0 {
            return Err(Par2Error::ResourceLimitExceeded {
                reason: "source provider slice size is zero".to_string(),
            });
        }
        let mut files = Vec::with_capacity(sources.len());
        let mut fingerprints = Vec::with_capacity(sources.len());
        let mut slice_starts = Vec::with_capacity(sources.len());
        let mut next_start = 0usize;
        for source in sources {
            if cancellation.is_cancelled() {
                return Err(Par2Error::Cancelled);
            }
            let metadata = fs::metadata(&source.path).map_err(Par2Error::Io)?;
            if !metadata.is_file() || metadata.len() != source.file_length {
                return Err(Par2Error::CreationSourceChanged {
                    path: source.path.display().to_string(),
                });
            }
            let slice_count = source.slice_checksums.len();
            slice_starts.push(next_start);
            next_start = next_start.checked_add(slice_count).ok_or_else(|| {
                Par2Error::ResourceLimitExceeded {
                    reason: "source slice provider index overflows".to_string(),
                }
            })?;
            fingerprints.push(SourceFingerprint::from_metadata(&metadata));
            files.push(File::open(&source.path).map_err(Par2Error::Io)?);
        }
        Ok(Self {
            sources,
            files,
            fingerprints,
            slice_starts,
            slice_size,
            cancellation,
        })
    }

    pub(crate) fn verify_unchanged(&self) -> Result<()> {
        for (source, fingerprint) in self.sources.iter().zip(&self.fingerprints) {
            let metadata = fs::metadata(&source.path).map_err(Par2Error::Io)?;
            if SourceFingerprint::from_metadata(&metadata) != *fingerprint {
                return Err(Par2Error::CreationSourceChanged {
                    path: source.path.display().to_string(),
                });
            }
        }
        Ok(())
    }

    fn source_location(&self, source_index: usize) -> Result<(usize, usize)> {
        let file_index = self
            .slice_starts
            .partition_point(|&start| start <= source_index)
            .checked_sub(1)
            .ok_or_else(|| Par2Error::CreationSourceChanged {
                path: "source slice index is out of range".to_string(),
            })?;
        let local_index = source_index
            .checked_sub(self.slice_starts[file_index])
            .ok_or_else(|| Par2Error::ResourceLimitExceeded {
                reason: "source slice index underflows".to_string(),
            })?;
        if local_index >= self.sources[file_index].slice_checksums.len() {
            return Err(Par2Error::CreationSourceChanged {
                path: self.sources[file_index].path.display().to_string(),
            });
        }
        Ok((file_index, local_index))
    }
}

impl ForwardSourceProvider for DiskSourceProvider<'_> {
    fn source_count(&self) -> usize {
        self.sources
            .iter()
            .map(|source| source.slice_checksums.len())
            .sum()
    }

    fn source_slice_len(&self, source_index: usize) -> Result<usize> {
        let (file_index, local_index) = self.source_location(source_index)?;
        let offset = (local_index as u64)
            .checked_mul(self.slice_size as u64)
            .ok_or_else(|| Par2Error::ResourceLimitExceeded {
                reason: "source slice offset overflows".to_string(),
            })?;
        usize::try_from(
            self.sources[file_index]
                .file_length
                .saturating_sub(offset)
                .min(self.slice_size as u64),
        )
        .map_err(|_| Par2Error::ResourceLimitExceeded {
            reason: "source slice length exceeds addressable memory".to_string(),
        })
    }

    fn read_source_chunk(
        &mut self,
        source_index: usize,
        offset: usize,
        destination: &mut [u8],
    ) -> Result<usize> {
        if self.cancellation.is_cancelled() {
            return Err(Par2Error::Cancelled);
        }
        let (file_index, local_index) = self.source_location(source_index)?;
        let slice_len = self.source_slice_len(source_index)?;
        let start = offset.min(slice_len);
        let take = destination.len().min(slice_len.saturating_sub(start));
        if take == 0 {
            return Ok(0);
        }
        let slice_offset = (local_index as u64)
            .checked_mul(self.slice_size as u64)
            .and_then(|value| value.checked_add(start as u64))
            .ok_or_else(|| Par2Error::ResourceLimitExceeded {
                reason: "source read offset overflows".to_string(),
            })?;
        self.files[file_index]
            .seek(SeekFrom::Start(slice_offset))
            .map_err(Par2Error::Io)?;
        self.files[file_index]
            .read_exact(&mut destination[..take])
            .map_err(|error| {
                if error.kind() == io::ErrorKind::UnexpectedEof {
                    Par2Error::CreationSourceChanged {
                        path: self.sources[file_index].path.display().to_string(),
                    }
                } else {
                    Par2Error::Io(error)
                }
            })?;
        Ok(take)
    }
}

#[allow(clippy::too_many_arguments)]
fn resolve_and_hash_source(
    base: &Path,
    input: &Path,
    block_size: u64,
    cancellation: &CancellationToken,
    progress: Option<&ProgressCallback>,
    file_index: u32,
    file_total: u32,
    bytes_processed: &AtomicU64,
    total_bytes: u64,
    cache: Option<&SourceScanCache>,
) -> Result<CreationSource> {
    if block_size == 0 || !block_size.is_multiple_of(4) {
        return Err(Par2Error::InvalidCreationOptions {
            reason: format!("source block size {block_size} is not a positive multiple of four"),
        });
    }
    let block_size_usize =
        usize::try_from(block_size).map_err(|_| Par2Error::ResourceLimitExceeded {
            reason: format!("source block size {block_size} exceeds addressable memory"),
        })?;

    let path = resolve_input_path(base, input)?;
    let metadata = fs::metadata(&path).map_err(Par2Error::Io)?;
    if !metadata.is_file() {
        return Err(Par2Error::UnsafeCreationSource {
            path: input.display().to_string(),
            reason: "source is not a regular file".to_string(),
        });
    }
    let fingerprint = SourceFingerprint::from_metadata(&metadata);
    let relative = path
        .strip_prefix(base)
        .map_err(|_| Par2Error::UnsafeCreationSource {
            path: input.display().to_string(),
            reason: "source is outside the base directory".to_string(),
        })?;
    validate_relative_path(relative, input)?;
    let relative_name = relative
        .to_str()
        .ok_or_else(|| Par2Error::UnsafeCreationSource {
            path: input.display().to_string(),
            reason: "source filename is not valid UTF-8".to_string(),
        })?;
    let par2_name = relative_name.replace('\\', "/");
    let par2_name = translate_par2_name_to_relative(&par2_name).map_err(|error| {
        Par2Error::UnsafeCreationSource {
            path: input.display().to_string(),
            reason: error.to_string(),
        }
    })?;
    if par2_name.is_empty()
        || par2_name.len() > MAX_PAR2_NAME_BYTES
        || par2_name.as_bytes().contains(&0)
    {
        return Err(Par2Error::UnsafeCreationSource {
            path: input.display().to_string(),
            reason: "PAR2 name is empty, contains NUL, or exceeds 100000 bytes".to_string(),
        });
    }

    // Memo hit: this exact path, at this exact fingerprint and block size, was
    // already read and hashed by this creator, and the `stat` above is the
    // guard that says so. The bytes are not read again; progress is still
    // reported one slice at a time, with the same byte steps a real scan of
    // this file would produce, so the callback stream a caller sees does not
    // depend on which read the hashes came from.
    if let Some(source) = cache.and_then(|cache| cache.take(&path, &fingerprint, block_size)) {
        debug_assert_eq!(source.par2_name, par2_name);
        let mut remaining = fingerprint.length;
        for _ in 0..source.slice_checksums.len() {
            if cancellation.is_cancelled() {
                return Err(Par2Error::Cancelled);
            }
            let step = remaining.min(block_size);
            remaining -= step;
            let processed_total = bytes_processed
                .fetch_add(step, Ordering::Relaxed)
                .saturating_add(step);
            report_progress(
                progress,
                file_index,
                file_total,
                processed_total,
                total_bytes,
            );
        }
        return Ok(source);
    }

    let mut file = File::open(&path).map_err(Par2Error::Io)?;
    let slice_count = if fingerprint.length == 0 {
        0
    } else {
        fingerprint
            .length
            .checked_add(block_size - 1)
            .ok_or_else(|| Par2Error::ResourceLimitExceeded {
                reason: "source slice count overflows".to_string(),
            })?
            / block_size
    };
    if slice_count > MAX_SLICES_PER_FILE as u64 {
        return Err(Par2Error::ResourceLimitExceeded {
            reason: format!("source slice count {slice_count} exceeds {MAX_SLICES_PER_FILE}"),
        });
    }
    let slice_count_usize =
        usize::try_from(slice_count).map_err(|_| Par2Error::ResourceLimitExceeded {
            reason: "source slice count exceeds addressable memory".to_string(),
        })?;

    let mut full_hash = FileHashState::new();
    let mut first_hash = Md5State::new();
    let mut first_bytes = 0u64;
    let mut checksums = Vec::with_capacity(slice_count_usize);

    // Per-slice length, shared by both scan shapes: every slice is a full
    // block except the file's last, which is short and zero-padded to the
    // block size for checksum purposes.
    let slice_len = |slice_index: usize| -> Result<usize> {
        let offset = (slice_index as u64)
            .checked_mul(block_size)
            .ok_or_else(|| Par2Error::ResourceLimitExceeded {
                reason: "source slice offset overflows".to_string(),
            })?;
        usize::try_from(fingerprint.length.saturating_sub(offset).min(block_size)).map_err(|_| {
            Par2Error::ResourceLimitExceeded {
                reason: "source slice length exceeds addressable memory".to_string(),
            }
        })
    };

    let lanes = create_md5_batch_lanes(block_size_usize);
    if lanes >= 2 {
        // Batched scan. Consecutive slices are independent MD5 messages, so a
        // batch of them goes through the multi-buffer kernel in one pass. The
        // whole-file MD5 stays a single serial stream over the same bytes: it
        // is a different message and cannot be laned within one file (see the
        // module note on why it is also not laned *across* files).
        let mut batch = vec![0u8; lanes * block_size_usize];
        let mut digests = vec![[0u8; 16]; lanes];
        let mut lens = vec![0usize; lanes];
        let mut crcs = vec![0u32; lanes];
        let mut slice_index = 0usize;

        while slice_index < slice_count_usize {
            let batch_slices = lanes.min(slice_count_usize - slice_index);

            for lane in 0..batch_slices {
                if cancellation.is_cancelled() {
                    return Err(Par2Error::Cancelled);
                }
                let actual_len = slice_len(slice_index + lane)?;
                let start = lane * block_size_usize;
                let slice = &mut batch[start..start + actual_len];
                read_exact_or_changed(&mut file, slice, input)?;
                lens[lane] = actual_len;

                // Both single-stream digests run here, while this slice is
                // still in the cache the read just filled, instead of after a
                // second whole-batch walk. Only the file's final slice can be
                // short, so lane order within a batch is file order and the
                // serial stream sees exactly the bytes — and the byte order —
                // that one `batch[..batch_bytes]` absorb produced.
                let slice = &batch[start..start + actual_len];
                absorb_file_stream(&mut full_hash, &mut first_hash, &mut first_bytes, slice);
                crcs[lane] = checksum::crc32_padded(slice, block_size);

                let processed_total = bytes_processed
                    .fetch_add(actual_len as u64, Ordering::Relaxed)
                    .saturating_add(actual_len as u64);
                report_progress(
                    progress,
                    file_index,
                    file_total,
                    processed_total,
                    total_bytes,
                );
            }

            let inputs = (0..batch_slices)
                .map(|lane| {
                    let start = lane * block_size_usize;
                    &batch[start..start + lens[lane]]
                })
                .collect::<Vec<_>>();
            md5_simd::md5_multi_into(&inputs, Some(block_size), &mut digests[..batch_slices]);

            for lane in 0..batch_slices {
                checksums.push(SliceChecksum {
                    crc32: crcs[lane],
                    md5: digests[lane],
                });
            }

            slice_index += batch_slices;
        }
    } else {
        // Streaming scan: one slice at a time through a single read buffer.
        // Selected when one block already fills the staging budget, so there
        // is no second lane to fill anyway.
        let mut buffer = vec![0u8; READ_BUFFER_BYTES.min(block_size_usize.max(1))];
        for slice_index in 0..slice_count_usize {
            if cancellation.is_cancelled() {
                return Err(Par2Error::Cancelled);
            }
            let actual_len = slice_len(slice_index)?;
            let mut remaining = actual_len;
            let mut slice_hash = SliceChecksumState::new();
            while remaining > 0 {
                if cancellation.is_cancelled() {
                    return Err(Par2Error::Cancelled);
                }
                let take = remaining.min(buffer.len());
                read_exact_or_changed(&mut file, &mut buffer[..take], input)?;
                let chunk = &buffer[..take];
                slice_hash.update(chunk);
                absorb_file_stream(&mut full_hash, &mut first_hash, &mut first_bytes, chunk);
                remaining -= take;
                let processed_total = bytes_processed
                    .fetch_add(take as u64, Ordering::Relaxed)
                    .saturating_add(take as u64);
                report_progress(
                    progress,
                    file_index,
                    file_total,
                    processed_total,
                    total_bytes,
                );
            }
            let pad_to = ((actual_len as u64) < block_size).then_some(block_size);
            let (crc32, md5) = slice_hash.finalize(pad_to);
            checksums.push(SliceChecksum { crc32, md5 });
        }
    }

    if full_hash.bytes_fed() != fingerprint.length {
        return Err(Par2Error::CreationSourceChanged {
            path: input.display().to_string(),
        });
    }
    let after = fs::metadata(&path).map_err(Par2Error::Io)?;
    if SourceFingerprint::from_metadata(&after) != fingerprint {
        return Err(Par2Error::CreationSourceChanged {
            path: input.display().to_string(),
        });
    }

    let hash_full = full_hash.finalize();
    let hash_16k = first_hash.finalize();
    let mut file_id_hash = Md5State::new();
    file_id_hash.update(&hash_16k);
    file_id_hash.update(&fingerprint.length.to_le_bytes());
    file_id_hash.update(par2_name.as_bytes());
    let file_id = FileId::from_bytes(file_id_hash.finalize());

    let source = CreationSource {
        path,
        par2_name,
        file_id,
        file_length: fingerprint.length,
        hash_full,
        hash_16k,
        slice_checksums: checksums,
    };
    if let Some(cache) = cache {
        cache.store(fingerprint, block_size, &source);
    }
    Ok(source)
}

fn resolve_input_path(base: &Path, input: &Path) -> Result<PathBuf> {
    if input.is_absolute() {
        return fs::canonicalize(input).map_err(Par2Error::Io);
    }

    fs::canonicalize(base.join(input)).map_err(Par2Error::Io)
}

fn validate_relative_path(relative: &Path, input: &Path) -> Result<()> {
    if relative.as_os_str().is_empty()
        || relative.components().any(|component| {
            matches!(
                component,
                Component::ParentDir | Component::RootDir | Component::Prefix(_)
            )
        })
    {
        return Err(Par2Error::UnsafeCreationSource {
            path: input.display().to_string(),
            reason: "source does not have a safe relative path".to_string(),
        });
    }
    Ok(())
}

/// Feed one contiguous, in-file-order run of source bytes to the two
/// whole-file digests.
///
/// Both are single serial MD5 streams over the entire file, so neither can be
/// laned: multi-buffer widens a batch of independent messages, not one message.
/// They stay scalar (the aws-lc backend when built with `native-crypto`) while
/// the per-slice digests beside them go through the SIMD kernel.
fn absorb_file_stream(
    full_hash: &mut FileHashState,
    first_hash: &mut Md5State,
    first_bytes: &mut u64,
    chunk: &[u8],
) {
    full_hash.update(chunk);
    if *first_bytes < FIRST_HASH_BYTES {
        let take = (FIRST_HASH_BYTES - *first_bytes).min(chunk.len() as u64) as usize;
        first_hash.update(&chunk[..take]);
        *first_bytes += take as u64;
    }
}

fn read_exact_or_changed(file: &mut File, buffer: &mut [u8], input: &Path) -> Result<()> {
    file.read_exact(buffer).map_err(|error| {
        if error.kind() == io::ErrorKind::UnexpectedEof {
            Par2Error::CreationSourceChanged {
                path: input.display().to_string(),
            }
        } else {
            Par2Error::Io(error)
        }
    })
}

fn report_progress(
    progress: Option<&ProgressCallback>,
    current: u32,
    total: u32,
    bytes_processed: u64,
    total_bytes: u64,
) {
    if let Some(progress) = progress {
        progress(crate::types::ProgressUpdate {
            stage: ProgressStage::Creating,
            current,
            total,
            bytes_processed,
            total_bytes: Some(total_bytes),
            phase: crate::types::ProgressPhase::SourceScan,
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn main_file_count_boundary_matches_parser_limit() {
        assert!(validate_main_file_count(MAX_FILES_PER_SET).is_ok());
        assert!(matches!(
            validate_main_file_count(MAX_FILES_PER_SET + 1),
            Err(Par2Error::ResourceLimitExceeded { .. })
        ));
    }
}