bot-forge 1.0.2

//! Filesystem primitives with cancellable locking and atomic replacement.
//!
//! Higher-level modules retain ownership of path policy and persistent schemas. This module owns
//! only reusable I/O mechanics and emits neutral resource-wait lifecycle facts while polling
//! cross-process locks.

use std::fs::{self, File, OpenOptions};
use std::io::{self, Write};
#[cfg(windows)]
use std::os::windows::ffi::OsStrExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(windows)]
use windows_sys::Win32::Storage::FileSystem::{
    MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW,
};

use fs2::FileExt;

use crate::cancellation::requested;
use crate::error::ForgeError;
use crate::events::{LifecycleEvent, emit};
use crate::paths::app_home;
use crate::util::fnv1a;

static WRITE_SEQUENCE: AtomicU64 = AtomicU64::new(0);

pub(crate) fn is_lock_contended(error: &io::Error) -> bool {
    let expected = fs2::lock_contended_error();
    match (error.raw_os_error(), expected.raw_os_error()) {
        (Some(actual), Some(expected)) => actual == expected,
        _ => error.kind() == expected.kind(),
    }
}

pub(crate) struct NamedFileLease {
    file: File,
}

pub(crate) fn acquire_named_lock(
    namespace: &str,
    identity: &str,
) -> Result<NamedFileLease, ForgeError> {
    let directory = app_home().join("locks").join(namespace);
    create_dir_all(&directory)?;
    let path = directory.join(format!(
        "{:016x}-{:016x}.lock",
        fnv1a(namespace),
        fnv1a(identity)
    ));
    acquire_path_lock(&path, &format!("{namespace} lock"))
}

pub(crate) fn acquire_path_lock(path: &Path, label: &str) -> Result<NamedFileLease, ForgeError> {
    if let Some(parent) = path.parent() {
        create_dir_all(parent)?;
    }
    let file = OpenOptions::new()
        .create(true)
        .truncate(false)
        .read(true)
        .write(true)
        .open(path)
        .map_err(|source| ForgeError::Io {
            path: path.to_path_buf(),
            source,
        })?;
    lock_exclusive_cancellable(&file, path, label)?;
    Ok(NamedFileLease { file })
}

pub(crate) fn lock_exclusive_cancellable(
    file: &File,
    path: &Path,
    label: &str,
) -> Result<(), ForgeError> {
    lock_while(file, path, label, requested, false)
}

pub(crate) fn lock_shared_cancellable(
    file: &File,
    path: &Path,
    label: &str,
) -> Result<(), ForgeError> {
    lock_while(file, path, label, requested, true)
}

pub(crate) fn lock_exclusive_while(
    file: &File,
    path: &Path,
    label: &str,
    cancelled: impl Fn() -> bool,
) -> Result<(), ForgeError> {
    lock_while(file, path, label, cancelled, false)
}

fn lock_while(
    file: &File,
    path: &Path,
    label: &str,
    cancelled: impl Fn() -> bool,
    shared: bool,
) -> Result<(), ForgeError> {
    emit(
        None,
        None,
        None,
        LifecycleEvent::ResourceWait {
            resource: label.to_string(),
        },
    );
    let started = std::time::Instant::now();
    loop {
        if cancelled() {
            return Err(ForgeError::Command(format!(
                "cancelled while waiting for {label}"
            )));
        }
        let result = if shared {
            FileExt::try_lock_shared(file)
        } else {
            FileExt::try_lock_exclusive(file)
        };
        match result {
            Ok(()) => {
                emit(
                    None,
                    None,
                    None,
                    LifecycleEvent::ResourceAcquired {
                        resource: label.to_string(),
                        wait_ms: started.elapsed().as_millis(),
                    },
                );
                return Ok(());
            }
            Err(error) if is_lock_contended(&error) => {
                std::thread::sleep(std::time::Duration::from_millis(50));
            }
            Err(source) => {
                return Err(ForgeError::Io {
                    path: path.to_path_buf(),
                    source,
                });
            }
        }
    }
}

impl Drop for NamedFileLease {
    fn drop(&mut self) {
        let _ = FileExt::unlock(&self.file);
    }
}

pub(crate) fn copy_dir(source: &Path, target: &Path) -> Result<(), ForgeError> {
    if target.exists() {
        return Err(ForgeError::Config(format!(
            "{} already exists; refusing to overwrite the target",
            target.display()
        )));
    }
    create_dir_all(target)?;
    for entry in read_dir(source)? {
        let path = entry?;
        let file_name = path
            .file_name()
            .ok_or_else(|| ForgeError::Config(format!("invalid path: {}", path.display())))?;
        let dest = target.join(file_name);
        if path.is_dir() {
            copy_dir(&path, &dest)?;
        } else {
            copy_file(&path, &dest)?;
        }
    }
    Ok(())
}

pub(crate) fn copy_file(source: &Path, target: &Path) -> Result<(), ForgeError> {
    if let Some(parent) = target.parent() {
        create_dir_all(parent)?;
    }
    fs::copy(source, target).map_err(|source_error| ForgeError::Io {
        path: source.to_path_buf(),
        source: source_error,
    })?;
    Ok(())
}

pub(crate) fn read_dir(path: &Path) -> Result<Vec<Result<PathBuf, ForgeError>>, ForgeError> {
    let entries = fs::read_dir(path)
        .map_err(|source| ForgeError::Io {
            path: path.to_path_buf(),
            source,
        })?
        .map(|entry| {
            entry
                .map(|entry| entry.path())
                .map_err(|source| ForgeError::Io {
                    path: path.to_path_buf(),
                    source,
                })
        })
        .collect();
    Ok(entries)
}

pub(crate) fn create_dir_all(path: &Path) -> Result<(), ForgeError> {
    fs::create_dir_all(path).map_err(|source| ForgeError::Io {
        path: path.to_path_buf(),
        source,
    })
}

pub(crate) fn remove_dir_all(path: &Path) -> Result<(), ForgeError> {
    fs::remove_dir_all(path).map_err(|source| ForgeError::Io {
        path: path.to_path_buf(),
        source,
    })
}

pub(crate) fn remove_file(path: &Path) -> Result<(), ForgeError> {
    fs::remove_file(path).map_err(|source| ForgeError::Io {
        path: path.to_path_buf(),
        source,
    })
}

pub(crate) fn read_to_string(path: &Path) -> Result<String, ForgeError> {
    fs::read_to_string(path).map_err(|source| ForgeError::Io {
        path: path.to_path_buf(),
        source,
    })
}

pub(crate) fn write_file(path: &Path, contents: &str) -> Result<(), ForgeError> {
    fs::write(path, contents).map_err(|source| ForgeError::Io {
        path: path.to_path_buf(),
        source,
    })
}

pub(crate) fn atomic_write_file(path: &Path, contents: &[u8]) -> Result<(), ForgeError> {
    atomic_write_file_with(path, contents, |file, contents| file.write_all(contents))
}

fn atomic_write_file_with(
    path: &Path,
    contents: &[u8],
    write: impl FnOnce(&mut File, &[u8]) -> std::io::Result<()>,
) -> Result<(), ForgeError> {
    if let Some(parent) = path.parent() {
        create_dir_all(parent)?;
    }
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .ok_or_else(|| ForgeError::Config(format!("invalid write path: {}", path.display())))?;
    let sequence = WRITE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
    let temp = path.with_file_name(format!(
        ".{file_name}.{}-{sequence}.tmp",
        std::process::id()
    ));
    let result = (|| {
        let mut file = OpenOptions::new()
            .create(true)
            .truncate(true)
            .write(true)
            .open(&temp)
            .map_err(|source| ForgeError::Io {
                path: temp.clone(),
                source,
            })?;
        write(&mut file, contents).map_err(|source| ForgeError::Io {
            path: temp.clone(),
            source,
        })?;
        file.sync_all().map_err(|source| ForgeError::Io {
            path: temp.clone(),
            source,
        })?;
        replace_file(&temp, path).map_err(|source| ForgeError::Io {
            path: path.to_path_buf(),
            source,
        })?;
        Ok(())
    })();
    if result.is_err() {
        let _ = fs::remove_file(temp);
    }
    result
}

#[cfg(not(windows))]
pub(crate) fn replace_file(source: &Path, target: &Path) -> std::io::Result<()> {
    fs::rename(source, target)
}

#[cfg(windows)]
pub(crate) fn replace_file(source: &Path, target: &Path) -> std::io::Result<()> {
    let source = source
        .as_os_str()
        .encode_wide()
        .chain(std::iter::once(0))
        .collect::<Vec<_>>();
    let target = target
        .as_os_str()
        .encode_wide()
        .chain(std::iter::once(0))
        .collect::<Vec<_>>();
    // SAFETY: Both buffers are NUL-terminated and remain alive for the duration of the call.
    // The flags request replacement and synchronous metadata persistence only.
    let result = unsafe {
        MoveFileExW(
            source.as_ptr(),
            target.as_ptr(),
            MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
        )
    };
    if result == 0 {
        Err(std::io::Error::last_os_error())
    } else {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::fs::{self, OpenOptions};
    use std::sync::atomic::{AtomicUsize, Ordering};

    use crate::fsutil::{
        atomic_write_file, atomic_write_file_with, is_lock_contended, lock_exclusive_while,
    };
    use crate::util::now_secs;
    use fs2::FileExt;

    #[test]
    fn recognizes_only_the_platform_lock_contention_error() {
        assert!(is_lock_contended(&fs2::lock_contended_error()));
        assert!(!is_lock_contended(&std::io::Error::new(
            std::io::ErrorKind::PermissionDenied,
            "ordinary permission failure",
        )));
    }

    #[test]
    fn atomic_write_replaces_an_existing_file() {
        let root = std::env::temp_dir().join(format!(
            "bot-forge-atomic-write-{}-{}",
            std::process::id(),
            now_secs()
        ));
        let path = root.join("state.json");
        atomic_write_file(&path, b"first").unwrap();
        atomic_write_file(&path, b"second").unwrap();
        assert_eq!(fs::read(&path).unwrap(), b"second");
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn atomic_write_preserves_target_and_removes_temp_on_disk_full() {
        let root = std::env::temp_dir().join(format!(
            "bot-forge-atomic-disk-full-{}-{}",
            std::process::id(),
            now_secs()
        ));
        let path = root.join("state.json");
        atomic_write_file(&path, b"stable").unwrap();
        let error = atomic_write_file_with(&path, b"replacement", |_, _| {
            Err(std::io::Error::new(
                std::io::ErrorKind::StorageFull,
                "injected disk full",
            ))
        })
        .unwrap_err();
        assert!(error.to_string().contains("injected disk full"));
        assert_eq!(fs::read(&path).unwrap(), b"stable");
        assert_eq!(fs::read_dir(&root).unwrap().count(), 1);
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn cancellable_file_lock_exits_without_waiting_for_the_holder() {
        let root = std::env::temp_dir().join(format!(
            "bot-forge-cancellable-lock-{}-{}",
            std::process::id(),
            now_secs()
        ));
        fs::create_dir_all(&root).unwrap();
        let path = root.join("lock");
        let holder = OpenOptions::new()
            .create(true)
            .truncate(false)
            .read(true)
            .write(true)
            .open(&path)
            .unwrap();
        holder.lock_exclusive().unwrap();
        let waiter = OpenOptions::new()
            .read(true)
            .write(true)
            .open(&path)
            .unwrap();
        let polls = AtomicUsize::new(0);
        let started = std::time::Instant::now();
        let error = lock_exclusive_while(&waiter, &path, "test lock", || {
            polls.fetch_add(1, Ordering::SeqCst) >= 1
        })
        .unwrap_err();
        assert!(error.to_string().contains("cancelled"));
        assert!(started.elapsed() < std::time::Duration::from_millis(150));
        assert!(polls.load(Ordering::SeqCst) >= 2);
        let _ = FileExt::unlock(&holder);
        fs::remove_dir_all(root).unwrap();
    }
}