runite 0.1.0

//! macOS filesystem backend.

use std::collections::VecDeque;
use std::ffi::CString;
use std::future::poll_fn;
use std::io;
use std::os::fd::{FromRawFd, OwnedFd, RawFd};
use std::os::unix::ffi::OsStrExt;
use std::os::unix::fs::{FileTypeExt, MetadataExt, OpenOptionsExt};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};

use crate::op::completion::completion_for_current_thread;
use crate::op::fs::{FileType, FsOp, MetadataTarget, RawDirEntry, RawMetadata};
use crate::platform::current::runtime::{ThreadHandle, current_thread_handle};
use crate::sys::blocking::spawn_blocking;

pub async fn open(op: FsOp) -> io::Result<OwnedFd> {
    let FsOp::Open { path, options } = op else {
        unreachable!("open backend called with non-open op");
    };

    offload(move || {
        let mut open = std::fs::OpenOptions::new();
        open.read(options.read)
            .write(options.write)
            .append(options.append)
            .truncate(options.truncate)
            .create(options.create)
            .create_new(options.create_new)
            .mode(0o666);
        let file = open.open(path)?;
        Ok(unsafe { OwnedFd::from_raw_fd(std::os::fd::IntoRawFd::into_raw_fd(file)) })
    })
    .await
}

pub async fn read(op: FsOp) -> io::Result<Vec<u8>> {
    let FsOp::Read { fd, offset, len } = op else {
        unreachable!("read backend called with non-read op");
    };

    offload(move || {
        let mut buffer = vec![0; len];
        let read = match offset {
            Some(offset) => unsafe {
                libc::pread(
                    fd,
                    buffer.as_mut_ptr().cast::<libc::c_void>(),
                    len,
                    offset as libc::off_t,
                )
            },
            None => unsafe { libc::read(fd, buffer.as_mut_ptr().cast::<libc::c_void>(), len) },
        };
        if read < 0 {
            return Err(io::Error::last_os_error());
        }
        buffer.truncate(read as usize);
        Ok(buffer)
    })
    .await
}

pub async fn write(op: FsOp) -> io::Result<usize> {
    let FsOp::Write { fd, offset, data } = op else {
        unreachable!("write backend called with non-write op");
    };

    offload(move || {
        let written = match offset {
            Some(offset) => unsafe {
                libc::pwrite(
                    fd,
                    data.as_ptr().cast::<libc::c_void>(),
                    data.len(),
                    offset as libc::off_t,
                )
            },
            None => unsafe { libc::write(fd, data.as_ptr().cast::<libc::c_void>(), data.len()) },
        };
        if written < 0 {
            return Err(io::Error::last_os_error());
        }
        Ok(written as usize)
    })
    .await
}

pub async fn metadata(op: FsOp) -> io::Result<RawMetadata> {
    let FsOp::Metadata {
        target,
        follow_symlinks,
    } = op
    else {
        unreachable!("metadata backend called with non-metadata op");
    };

    offload(move || {
        let metadata = match target {
            MetadataTarget::Path(path) => {
                if follow_symlinks {
                    std::fs::metadata(path)
                } else {
                    std::fs::symlink_metadata(path)
                }
            }
            MetadataTarget::File(fd) => {
                let mut stat = unsafe { std::mem::zeroed::<libc::stat>() };
                let result = unsafe { libc::fstat(fd, &mut stat) };
                if result < 0 {
                    return Err(io::Error::last_os_error());
                }
                return Ok(raw_metadata_from_stat(&stat));
            }
        }?;
        Ok(raw_metadata_from_std(&metadata))
    })
    .await
}

/// Durably flushes a file to disk on macOS.
///
/// Plain `fsync(2)` on macOS does **not** flush the drive's write cache, so it
/// is not a real durability barrier; `fcntl(F_FULLFSYNC)` is. `std::fs` uses
/// `F_FULLFSYNC` for both `sync_all` and `sync_data` on Apple targets for this
/// reason, falling back to `fsync` on filesystems (e.g. some network mounts)
/// that do not support it. We match that behavior.
fn full_fsync(fd: RawFd) -> io::Result<()> {
    match cvt(unsafe { libc::fcntl(fd, libc::F_FULLFSYNC) }) {
        Ok(_) => Ok(()),
        Err(error)
            if matches!(
                error.raw_os_error(),
                Some(libc::ENOTSUP) | Some(libc::EINVAL) | Some(libc::ENOTTY)
            ) =>
        {
            cvt(unsafe { libc::fsync(fd) }).map(|_| ())
        }
        Err(error) => Err(error),
    }
}

pub async fn sync_all(op: FsOp) -> io::Result<()> {
    let FsOp::SyncAll { fd } = op else {
        unreachable!("sync_all backend called with non-sync_all op");
    };

    offload(move || full_fsync(fd)).await
}

pub async fn sync_data(op: FsOp) -> io::Result<()> {
    let FsOp::SyncData { fd } = op else {
        unreachable!("sync_data backend called with non-sync_data op");
    };

    offload(move || full_fsync(fd)).await
}

pub async fn set_len(op: FsOp) -> io::Result<()> {
    let FsOp::SetLen { fd, len } = op else {
        unreachable!("set_len backend called with non-set_len op");
    };

    offload(move || cvt(unsafe { libc::ftruncate(fd, len as libc::off_t) }).map(|_| ())).await
}

/// Repositions the file's kernel cursor. `lseek(2)` on a regular file is a fast
/// metadata operation that does not block, so it runs inline on the event loop.
pub fn seek(fd: RawFd, pos: std::io::SeekFrom) -> io::Result<u64> {
    let (whence, offset) = match pos {
        std::io::SeekFrom::Start(n) => (libc::SEEK_SET, n as libc::off_t),
        std::io::SeekFrom::End(n) => (libc::SEEK_END, n as libc::off_t),
        std::io::SeekFrom::Current(n) => (libc::SEEK_CUR, n as libc::off_t),
    };
    // SAFETY: `lseek` takes only a descriptor and integer arguments.
    let result = unsafe { libc::lseek(fd, offset, whence) };
    if result < 0 {
        Err(io::Error::last_os_error())
    } else {
        Ok(result as u64)
    }
}

pub async fn try_clone(op: FsOp) -> io::Result<OwnedFd> {
    let FsOp::Duplicate { fd } = op else {
        unreachable!("try_clone backend called with non-duplicate op");
    };

    offload(move || {
        let duplicated = cvt(unsafe { libc::fcntl(fd, libc::F_DUPFD_CLOEXEC, 0) })?;
        Ok(unsafe { OwnedFd::from_raw_fd(duplicated) })
    })
    .await
}

pub async fn create_dir(op: FsOp) -> io::Result<()> {
    let FsOp::CreateDir { path, mode } = op else {
        unreachable!("create_dir backend called with non-create_dir op");
    };

    offload(move || {
        let c_path = path_to_c_string(path)?;
        cvt(unsafe { libc::mkdir(c_path.as_ptr(), mode as libc::mode_t) }).map(|_| ())
    })
    .await
}

pub async fn remove_file(op: FsOp) -> io::Result<()> {
    let FsOp::RemoveFile { path } = op else {
        unreachable!("remove_file backend called with non-remove_file op");
    };

    offload(move || std::fs::remove_file(path)).await
}

pub async fn remove_dir(op: FsOp) -> io::Result<()> {
    let FsOp::RemoveDir { path } = op else {
        unreachable!("remove_dir backend called with non-remove_dir op");
    };

    offload(move || std::fs::remove_dir(path)).await
}

pub async fn rename(op: FsOp) -> io::Result<()> {
    let FsOp::Rename { from, to } = op else {
        unreachable!("rename backend called with non-rename op");
    };

    offload(move || std::fs::rename(from, to)).await
}

pub fn read_dir(op: FsOp) -> io::Result<ReadDirStream> {
    let FsOp::ReadDir { path } = op else {
        unreachable!("read_dir backend called with non-read_dir op");
    };

    ReadDirStream::new(path)
}

pub struct ReadDirStream {
    state: Arc<ReadDirState>,
}

impl ReadDirStream {
    fn new(path: PathBuf) -> io::Result<Self> {
        let state = Arc::new(ReadDirState::new(current_thread_handle()));
        let producer = Arc::clone(&state);

        if let Err(error) = spawn_blocking(move || produce_dir_entries(path, producer)) {
            state.release_pending();
            return Err(error);
        }

        Ok(Self { state })
    }

    pub async fn next_entry(&mut self) -> io::Result<Option<RawDirEntry>> {
        poll_fn(|cx| self.state.poll_next(cx)).await
    }
}

struct ReadDirState {
    owner: ThreadHandle,
    queue: Mutex<VecDeque<io::Result<RawDirEntry>>>,
    done: AtomicBool,
    pending: AtomicBool,
    wake_queued: AtomicBool,
    waker: Mutex<Option<Waker>>,
}

impl ReadDirState {
    fn new(owner: ThreadHandle) -> Self {
        owner.begin_async_operation();
        Self {
            owner,
            queue: Mutex::new(VecDeque::new()),
            done: AtomicBool::new(false),
            pending: AtomicBool::new(true),
            wake_queued: AtomicBool::new(false),
            waker: Mutex::new(None),
        }
    }

    fn push(self: &Arc<Self>, entry: io::Result<RawDirEntry>) {
        self.queue.lock().unwrap().push_back(entry);
        self.notify();
    }

    fn finish(self: &Arc<Self>) {
        self.done.store(true, Ordering::Release);
        // Enqueue the consumer's wake BEFORE releasing the pending op. In the
        // reverse order there is a window where pending_ops is 0 and the
        // remote queue is empty; run()'s exit protocol can commit `closed` in
        // that window, after which the wake enqueue fails and the consumer is
        // stranded (its task never resumes, and callers observe the stream's
        // work as silently lost). With the wake enqueued first, the exit
        // commit's remote-queue check sees it and keeps the loop alive.
        // (Mirrors the ordering contract in op/completion.rs `finish`.)
        self.notify();
        self.release_pending();
    }

    fn release_pending(&self) {
        if self.pending.swap(false, Ordering::AcqRel) {
            self.owner.finish_async_operation();
        }
    }

    fn notify(self: &Arc<Self>) {
        if self.wake_queued.swap(true, Ordering::AcqRel) {
            return;
        }

        // Internal wake: routed through the capacity-bypassing path (like
        // completion and task wakes) so a full user queue can never drop it —
        // a dropped stream wake strands the consumer. The only failure is a
        // closed owner thread, where nothing can be scheduled anyway.
        let state = Arc::clone(self);
        if self
            .owner
            .queue_internal_wake(move || {
                state.wake_queued.store(false, Ordering::Release);
                if let Some(waker) = state.waker.lock().unwrap().take() {
                    waker.wake();
                }
            })
            .is_err()
        {
            self.wake_queued.store(false, Ordering::Release);
        }
    }

    fn poll_next(&self, cx: &mut Context<'_>) -> Poll<io::Result<Option<RawDirEntry>>> {
        if let Some(entry) = self.queue.lock().unwrap().pop_front() {
            return Poll::Ready(entry.map(Some));
        }

        if self.done.load(Ordering::Acquire) {
            return Poll::Ready(Ok(None));
        }

        *self.waker.lock().unwrap() = Some(cx.waker().clone());

        if let Some(entry) = self.queue.lock().unwrap().pop_front() {
            let _ = self.waker.lock().unwrap().take();
            return Poll::Ready(entry.map(Some));
        }

        if self.done.load(Ordering::Acquire) {
            let _ = self.waker.lock().unwrap().take();
            return Poll::Ready(Ok(None));
        }

        Poll::Pending
    }
}

impl Drop for ReadDirStream {
    fn drop(&mut self) {
        self.state.release_pending();
    }
}

fn produce_dir_entries(path: PathBuf, state: Arc<ReadDirState>) {
    match std::fs::read_dir(path) {
        Ok(entries) => {
            for entry in entries {
                match entry {
                    Ok(entry) => {
                        let file_name = entry.file_name();
                        state.push(Ok(RawDirEntry {
                            path: entry.path(),
                            file_name,
                        }));
                    }
                    Err(error) => state.push(Err(error)),
                }
            }
            state.finish();
        }
        Err(error) => {
            state.push(Err(error));
            state.finish();
        }
    }
}

async fn offload<T: Send + 'static>(
    work: impl FnOnce() -> io::Result<T> + Send + 'static,
) -> io::Result<T> {
    let (future, handle) = completion_for_current_thread::<io::Result<T>>();
    let handle_for_task = handle.clone();
    if let Err(error) = spawn_blocking(move || handle_for_task.complete(work())) {
        handle.complete(Err(error));
    }
    future.await
}

fn path_to_c_string(path: PathBuf) -> io::Result<CString> {
    CString::new(path.as_os_str().as_bytes()).map_err(|_| {
        io::Error::new(
            io::ErrorKind::InvalidInput,
            "path contains interior NUL bytes",
        )
    })
}

fn raw_metadata_from_std(metadata: &std::fs::Metadata) -> RawMetadata {
    let file_type = metadata.file_type();
    let kind = if file_type.is_file() {
        FileType::File
    } else if file_type.is_dir() {
        FileType::Directory
    } else if file_type.is_symlink() {
        FileType::Symlink
    } else if file_type.is_block_device() {
        FileType::BlockDevice
    } else if file_type.is_char_device() {
        FileType::CharacterDevice
    } else if file_type.is_fifo() {
        FileType::Fifo
    } else if file_type.is_socket() {
        FileType::Socket
    } else {
        FileType::Unknown
    };

    RawMetadata {
        file_type: kind,
        // Full st_mode (type + permission bits), matching std and the Linux
        // backend rather than masking to permission bits only.
        mode: metadata.mode(),
        len: metadata.len(),
        platform: Default::default(),
    }
}

fn raw_metadata_from_stat(stat: &libc::stat) -> RawMetadata {
    let kind = match stat.st_mode & libc::S_IFMT {
        libc::S_IFREG => FileType::File,
        libc::S_IFDIR => FileType::Directory,
        libc::S_IFLNK => FileType::Symlink,
        libc::S_IFBLK => FileType::BlockDevice,
        libc::S_IFCHR => FileType::CharacterDevice,
        libc::S_IFIFO => FileType::Fifo,
        libc::S_IFSOCK => FileType::Socket,
        _ => FileType::Unknown,
    };

    RawMetadata {
        file_type: kind,
        mode: u32::from(stat.st_mode),
        len: stat.st_size as u64,
        platform: Default::default(),
    }
}

fn cvt(value: libc::c_int) -> io::Result<libc::c_int> {
    if value < 0 {
        Err(io::Error::last_os_error())
    } else {
        Ok(value)
    }
}