mod links;
mod paths;
mod posix;
mod watch;
mod workflows;

use std::collections::{HashMap, VecDeque};
use std::convert::Infallible;
use std::ffi::OsString;
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use avril_caps::{Capabilities, PathTarget};
use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
use avril_core::{Extension, Op, PublicModule};
use avril_host::io::{CHUNK, MAX_SLICES, Response, response};
use cap_fs_ext::{DirExt, FollowSymlinks, OpenOptionsFollowExt, OpenOptionsSyncExt};
use cap_std::fs::{Dir, File, Metadata, OpenOptions};
use futures::channel::oneshot;
use tokio::runtime::Handle;

pub const PUBLIC_MODULES: &[PublicModule] = &[
    avril_core::public_module!("avril:fs", "ext:file", "fs"),
    avril_core::public_module!("avril:fs/sync", "ext:file", "fs_sync"),
];

const MAX_ID: u64 = 9_007_199_254_740_991;
const SAFE_INTEGER: u64 = (1 << 53) - 1;
/// Runs against the file and returns the reply's delivery.
type Operation = Box<dyn FnOnce(&mut Option<Opened>, bool) -> Delivery + Send>;
type Delivery = Box<dyn FnOnce() + Send>;
type Reply<T> = oneshot::Receiver<io::Result<T>>;
type Options = (bool, bool, bool, bool, bool, bool, Option<u32>);
type Info = (
    bool,
    bool,
    bool,
    u64,
    Option<f64>,
    Option<f64>,
    Option<f64>,
    Option<u32>,
    Option<u32>,
    Option<u32>,
);

async fn receive<T>(reply: Reply<T>) -> Response<T> {
    response(
        reply
            .await
            .unwrap_or_else(|_| Err(io::Error::other("Filesystem worker stopped"))),
    )
}

fn failed<T>(error: io::Error) -> Reply<T> {
    let (reply, receive) = oneshot::channel();
    let _ = reply.send(Err(error));
    receive
}

fn filesystem_path(caps: &Capabilities, label: &str) -> io::Result<PathBuf> {
    #[cfg(windows)]
    if !Path::new(label).has_root()
        && matches!(
            Path::new(label).components().next(),
            Some(Component::Prefix(_))
        )
    {
        return Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "not yet implemented: drive-relative filesystem paths",
        ));
    }
    Ok(caps.cwd().join(label))
}

fn denied(label: &str, write: bool) -> io::Error {
    io::Error::new(
        io::ErrorKind::PermissionDenied,
        format!("{} {label}", if write { "write" } else { "read" }),
    )
}

fn authorize(caps: &Capabilities, target: &PathTarget, label: &str, write: bool) -> io::Result<()> {
    if write {
        caps.check_write(target)
    } else {
        caps.check_read(target)
    }
    .map_err(|_| denied(label, write))
}

fn bad_resource() -> io::Error {
    io::Error::new(
        io::ErrorKind::NotConnected,
        "File handle is closed or does not allow this operation",
    )
}

#[derive(Clone)]
struct BlockingPool {
    runtime: Handle,
}

impl BlockingPool {
    fn run<T: Send + 'static>(
        &self,
        work: impl FnOnce() -> io::Result<T> + Send + 'static,
    ) -> Reply<T> {
        let (reply, receive) = oneshot::channel();
        self.runtime.spawn_blocking(move || {
            let _ = reply.send(work());
        });
        receive
    }
}

struct Queue {
    operations: VecDeque<Operation>,
    running: bool,
}

struct Resource {
    closed: Arc<AtomicBool>,
    queue: Mutex<Queue>,
    file: Mutex<Option<Opened>>,
    pool: BlockingPool,
}

impl Resource {
    fn submit<T: Send + 'static>(
        self: &Arc<Self>,
        operation: impl FnOnce(&mut Opened) -> io::Result<T> + Send + 'static,
    ) -> Reply<T> {
        let (reply, receive) = oneshot::channel();
        let mut queue = self.queue.lock().unwrap();
        if self.closed.load(Ordering::Acquire) {
            return failed(bad_resource());
        }
        queue.operations.push_back(Box::new(move |file, closed| {
            let result = if closed {
                Err(bad_resource())
            } else {
                file.as_mut().ok_or_else(bad_resource).and_then(operation)
            };
            Box::new(move || {
                let _ = reply.send(result);
            })
        }));
        if !queue.running {
            queue.running = true;
            let resource = self.clone();
            self.pool.runtime.spawn_blocking(move || resource.run());
        }
        receive
    }

    /// Allows multi-batch operations to observe closure without releasing the queue.
    fn submit_observed<T: Send + 'static>(
        self: &Arc<Self>,
        operation: impl FnOnce(&mut Opened, &AtomicBool) -> io::Result<T> + Send + 'static,
    ) -> Reply<T> {
        let closed = self.closed.clone();
        self.submit(move |opened| operation(opened, &closed))
    }

    fn run(self: Arc<Self>) {
        loop {
            let operation = {
                let mut queue = self.queue.lock().unwrap();
                match queue.operations.pop_front() {
                    Some(operation) => operation,
                    None => {
                        queue.running = false;
                        drop(queue);
                        return self.finish();
                    }
                }
            };
            let delivery = {
                let mut file = self.file.lock().unwrap();
                let closed = self.closed.load(Ordering::Acquire);
                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    operation(&mut file, closed)
                }))
                .inspect_err(|_| {
                    self.closed.store(true, Ordering::Release);
                    file.take();
                })
                .ok()
            };
            // Settle the queue before waking the caller, so code resuming on
            // the reply (such as lending the handle to a subprocess) finds
            // the handle idle rather than busy. Delivering under the queue
            // lock keeps replies in submission order.
            let idle = {
                let mut queue = self.queue.lock().unwrap();
                queue.running = !queue.operations.is_empty();
                if let Some(delivery) = delivery {
                    delivery();
                }
                !queue.running
            };
            if idle {
                return self.finish();
            }
        }
    }

    fn finish(&self) {
        if self.closed.load(Ordering::Acquire) {
            self.file.lock().unwrap().take();
        }
    }

    fn close(self: &Arc<Self>) {
        self.closed.store(true, Ordering::Release);
        let mut queue = self.queue.lock().unwrap();
        if !queue.running {
            queue.running = true;
            let resource = self.clone();
            // Physical closure stays off V8's thread, including finalizer cleanup.
            self.pool.runtime.spawn_blocking(move || resource.run());
        }
    }
}

struct Opened {
    file: File,
    target: PathTarget,
    label: String,
    caps: Arc<Capabilities>,
    read: bool,
    write: bool,
}

impl Opened {
    fn check(&self, write: bool) -> io::Result<()> {
        if (write && !self.write) || (!write && !self.read) {
            return Err(bad_resource());
        }
        authorize(&self.caps, &self.target, &self.label, write)
    }

    fn check_metadata(&self) -> io::Result<()> {
        if self.read {
            let read = self.check(false);
            if read.is_ok() || !self.write {
                return read;
            }
        }
        self.check(true)
    }
}

fn open_file(
    caps: Arc<Capabilities>,
    path: String,
    options: Options,
    non_regular: io::ErrorKind,
) -> io::Result<Opened> {
    if path.is_empty() {
        return Err(io::Error::new(
            io::ErrorKind::NotFound,
            "Empty filesystem path",
        ));
    }
    let (read, write, append, truncate, create, create_new, mode) = options;
    if (!read && !write && !append)
        || ((truncate || create || create_new) && !write && !append)
        || mode.is_some_and(|mode| mode > 0o7777)
    {
        return Err(io::Error::new(
            io::ErrorKind::InvalidInput,
            "Invalid open options",
        ));
    }
    let write = write || append;
    let absolute = filesystem_path(&caps, &path)?;
    let target = match if create_new {
        PathTarget::for_entry(absolute)
    } else {
        PathTarget::for_creation(absolute)
    } {
        Ok(target) => target,
        Err(_) if read && !caps.unrestricted_read() => {
            return Err(denied(&path, false));
        }
        Err(_) if write && !caps.unrestricted_write() => {
            return Err(denied(&path, true));
        }
        Err(error) => return Err(error),
    };
    if read {
        authorize(&caps, &target, &path, false)?;
    }
    if write {
        authorize(&caps, &target, &path, true)?;
    }
    let (directory, name) = parent(target.as_path())?;
    match directory.symlink_metadata(&name) {
        Ok(_) if create_new => {
            return Err(io::Error::new(
                io::ErrorKind::AlreadyExists,
                "File already exists",
            ));
        }
        Ok(metadata) if !metadata.is_file() => {
            return Err(io::Error::new(
                non_regular,
                "not yet implemented: opening non-regular files",
            ));
        }
        Ok(_) => (),
        Err(error) if error.kind() == io::ErrorKind::NotFound => (),
        Err(error) => return Err(error),
    }
    let mut flags = OpenOptions::new();
    flags
        .read(read)
        .write(write)
        .append(append)
        .create(create)
        .create_new(create_new)
        .follow(FollowSymlinks::No)
        .nonblock(true);
    #[cfg(unix)]
    {
        use cap_std::fs::OpenOptionsExt;
        if let Some(mode) = mode {
            flags.mode(mode);
        }
    }
    let file = directory.open_with(name, &flags)?;
    if !file.metadata()?.is_file() {
        return Err(io::Error::new(
            non_regular,
            "not yet implemented: opening non-regular files",
        ));
    }
    let opened = Opened {
        file,
        target,
        label: path,
        caps: caps.clone(),
        read,
        write,
    };
    // Truncation must not happen until the opened object is verified and authorized.
    if truncate {
        opened.check(true)?;
        opened.file.set_len(0)?;
    }
    Ok(opened)
}

#[cfg_attr(
    not(any(target_os = "linux", target_os = "macos", windows)),
    allow(dead_code)
)]
pub fn process_file(
    caps: Arc<Capabilities>,
    path: String,
    options: Options,
) -> io::Result<std::fs::File> {
    Ok(open_file(caps, path, options, io::ErrorKind::Unsupported)?
        .file
        .into_std())
}

pub fn response_file(
    caps: Arc<Capabilities>,
    path: String,
) -> io::Result<(std::fs::File, avril_http::file::CheckFile)> {
    let opened = open_file(
        caps,
        path,
        (true, false, false, false, false, false, None),
        io::ErrorKind::NotFound,
    )?;
    opened.check(false)?;
    let Opened { file, target, label, caps, .. } = opened;
    let check = Arc::new(move || authorize(&caps, &target, &label, false));
    Ok((file.into_std(), check))
}

pub struct Files {
    pool: BlockingPool,
    resources: Mutex<HashMap<u64, Arc<Resource>>>,
    next: AtomicU64,
    caps: Arc<Capabilities>,
    interrupted: Arc<AtomicBool>,
}

impl Drop for Files {
    fn drop(&mut self) {
        for resource in self
            .resources
            .get_mut()
            .unwrap()
            .drain()
            .map(|(_, resource)| resource)
        {
            resource.close();
        }
    }
}

impl Files {
    fn io_ops<A, R, F>(
        &self,
        name: &'static str,
        sync_name: &'static str,
        function: impl Fn(A) -> F + Clone + 'static,
    ) -> Vec<Op>
    where
        A: serde::de::DeserializeOwned,
        R: serde::Serialize + 'static,
        F: std::future::Future<Output = Result<R, Infallible>> + 'static,
    {
        let blocking = function.clone();
        let interrupted = self.interrupted.clone();
        vec![
            Op::asynchronous(name, function),
            Op::with_scope(sync_name, move |scope, args| {
                avril_host::wait(blocking(args), || {
                    avril_host::interrupted(scope, &interrupted)
                })
                .transpose()
            }),
        ]
    }

    #[cfg_attr(
        not(any(target_os = "linux", target_os = "macos", windows)),
        allow(dead_code)
    )]
    pub fn duplicate_for_process(
        &self,
        id: u64,
        read: bool,
        write: bool,
    ) -> io::Result<std::fs::File> {
        let resource = self
            .resources
            .lock()
            .unwrap()
            .get(&id)
            .cloned()
            .ok_or_else(bad_resource)?;
        if resource.closed.load(Ordering::SeqCst) {
            return Err(bad_resource());
        }
        let queue = resource.queue.lock().unwrap();
        if queue.running || !queue.operations.is_empty() {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                "File handle is busy",
            ));
        }
        let file = resource
            .file
            .try_lock()
            .map_err(|_| io::Error::new(io::ErrorKind::WouldBlock, "File handle is busy"))?;
        let opened = file.as_ref().ok_or_else(bad_resource)?;
        if read {
            opened.check(false)?;
        }
        if write {
            opened.check(true)?;
        }
        Ok(opened.file.try_clone()?.into_std())
    }

    pub fn response_handle(
        &self,
        id: u64,
        consume: bool,
    ) -> io::Result<(String, std::fs::File, avril_http::file::CheckFile)> {
        let mut resources = self.resources.lock().unwrap();
        let resource = resources.get(&id).cloned().ok_or_else(bad_resource)?;
        let queue = resource.queue.lock().unwrap();
        if resource.closed.load(Ordering::Acquire) {
            return Err(bad_resource());
        }
        if queue.running || !queue.operations.is_empty() {
            return Err(io::Error::new(
                io::ErrorKind::WouldBlock,
                "File handle is busy",
            ));
        }
        let mut file = resource
            .file
            .try_lock()
            .map_err(|_| io::Error::new(io::ErrorKind::WouldBlock, "File handle is busy"))?;
        let opened = file.as_ref().ok_or_else(bad_resource)?;
        opened.check(false)?;
        let name = opened.label.clone();
        let target = opened.target.clone();
        let caps = opened.caps.clone();
        let label = name.clone();
        let check = Arc::new(move || authorize(&caps, &target, &label, false));
        let file = if consume {
            resource.closed.store(true, Ordering::Release);
            let opened = file.take().unwrap();
            resources.remove(&id);
            opened.file.into_std()
        } else {
            #[cfg(unix)]
            let duplicate = opened.file.try_clone()?;
            #[cfg(windows)]
            let duplicate = {
                use cap_fs_ext::Reopen;
                // Windows positional reads still change a duplicated handle's shared cursor.
                opened.file.reopen(OpenOptions::new().read(true))?
            };
            #[cfg(not(any(unix, windows)))]
            return Err(io::Error::new(
                io::ErrorKind::Unsupported,
                "Retaining file response handles is not supported on this platform",
            ));
            #[cfg(any(unix, windows))]
            duplicate.into_std()
        };
        Ok((name, file, check))
    }

    fn submit<T: Send + 'static>(
        &self,
        id: u64,
        operation: impl FnOnce(&mut Opened) -> io::Result<T> + Send + 'static,
    ) -> Reply<T> {
        let resource = self.resources.lock().unwrap().get(&id).cloned();
        match resource {
            Some(resource) => resource.submit(operation),
            None => failed(bad_resource()),
        }
    }

    fn submit_observed<T: Send + 'static>(
        &self,
        id: u64,
        operation: impl FnOnce(&mut Opened, &AtomicBool) -> io::Result<T> + Send + 'static,
    ) -> Reply<T> {
        let resource = self.resources.lock().unwrap().get(&id).cloned();
        match resource {
            Some(resource) => resource.submit_observed(operation),
            None => failed(bad_resource()),
        }
    }

    fn open(self: &Arc<Self>, path: String, options: Options) -> Reply<u64> {
        let state = self.clone();
        self.pool.run(move || {
            let opened = open_file(
                state.caps.clone(),
                path,
                options,
                io::ErrorKind::Unsupported,
            )?;
            let id = state
                .next
                .fetch_update(Ordering::AcqRel, Ordering::Acquire, |id| {
                    (id <= MAX_ID).then_some(id + 1)
                })
                .map_err(|_| io::Error::other("File handle identifiers exhausted"))?;
            state.resources.lock().unwrap().insert(
                id,
                Arc::new(Resource {
                    closed: Arc::new(AtomicBool::new(false)),
                    queue: Mutex::new(Queue {
                        operations: VecDeque::new(),
                        running: false,
                    }),
                    file: Mutex::new(Some(opened)),
                    pool: state.pool.clone(),
                }),
            );
            Ok(id)
        })
    }
}

pub(crate) use avril_caps::fs::parent;

fn info(metadata: Metadata) -> Info {
    fn time(time: io::Result<cap_std::time::SystemTime>) -> Option<f64> {
        // Integer milliseconds stay exact through Date's float conversion;
        // float seconds would smear sub-millisecond remainders into the
        // millisecond field. Float arithmetic also cannot overflow: times
        // past the Date range read back as Invalid Date, as before.
        fn millis(duration: std::time::Duration) -> f64 {
            duration.as_secs() as f64 * 1000.0 + f64::from(duration.subsec_millis())
        }
        let time = time.ok()?.into_std();
        Some(match time.duration_since(std::time::UNIX_EPOCH) {
            Ok(duration) => millis(duration),
            Err(error) => -millis(error.duration()),
        })
    }
    #[cfg(unix)]
    let (mode, uid, gid) = {
        use cap_std::fs::MetadataExt;
        (
            Some(metadata.mode()),
            Some(metadata.uid()),
            Some(metadata.gid()),
        )
    };
    #[cfg(not(unix))]
    let (mode, uid, gid) = (None, None, None);
    (
        metadata.is_file(),
        metadata.is_dir(),
        metadata.is_symlink(),
        metadata.len(),
        time(metadata.modified()),
        time(metadata.accessed()),
        time(metadata.created()),
        mode,
        uid,
        gid,
    )
}

pub fn init(
    runtime: Handle,
    caps: Arc<Capabilities>,
    interrupted: Arc<AtomicBool>,
) -> (Extension, Arc<Files>) {
    let state = Arc::new(Files {
        pool: BlockingPool { runtime },
        resources: Mutex::new(HashMap::new()),
        next: AtomicU64::new(1),
        caps,
        interrupted,
    });
    let ops = state.clone();
    let shared = state.clone();
    let open = state.clone();
    let read = state.clone();
    let write = state.clone();
    let writev = state.clone();
    let seek = state.clone();
    let truncate = state.clone();
    let sync = state.clone();
    let stat = state.clone();
    let path_stat = state.clone();
    let link_ops = links::ops(state.clone());
    let path_ops = paths::ops(state.clone());
    let posix_ops = posix::ops(state.clone());
    let workflow_ops = workflows::ops(state.clone());
    let watch_ops = watch::ops(state.clone());
    let mut extension = Extension {
        name: "ext:file",
        dependencies: &["ext:url", "ext:web", "ext:webstreams"],
        ops: vec![
            vec![Op::new("op_file_url_path", |value: String| {
                let parsed = (|| {
                    let url = url::Url::parse(&value).map_err(io::Error::other)?;
                    if url.scheme() != "file" || url.query().is_some() || url.fragment().is_some() {
                        return Err(io::Error::new(
                            io::ErrorKind::InvalidInput,
                            "Expected a file URL without query or fragment",
                        ));
                    }
                    url.to_file_path()
                        .map_err(|_| {
                            io::Error::new(io::ErrorKind::InvalidInput, "Invalid file URL")
                        })?
                        .into_os_string()
                        .into_string()
                        .map_err(|_| {
                            io::Error::new(io::ErrorKind::InvalidInput, "Non-UTF-8 file URL")
                        })
                })();
                Ok::<_, Infallible>(response(parsed))
            })],
            ops.io_ops(
                "op_file_open",
                "op_file_open_sync",
                move |(path, options): (String, Options)| {
                    let pending = open.open(path, options);
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            ops.io_ops(
                "op_file_read",
                "op_file_read_sync",
                move |(id, limit): (u64, usize)| {
                    let pending = read.submit(id, move |opened| {
                        opened.check(false)?;
                        let mut bytes = vec![0; limit.min(CHUNK)];
                        if limit == 0 {
                            return Ok(Some(ToJsBuffer::from(bytes)));
                        }
                        let count = loop {
                            match opened.file.read(&mut bytes) {
                                Err(error) if error.kind() == io::ErrorKind::Interrupted => {
                                    continue;
                                }
                                result => break result?,
                            }
                        };
                        bytes.truncate(count);
                        Ok(if count == 0 {
                            None
                        } else {
                            Some(ToJsBuffer::from(bytes))
                        })
                    });
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            ops.io_ops(
                "op_file_write",
                "op_file_write_sync",
                move |(id, bytes): (u64, JsBuffer)| {
                    let bytes = bytes[..bytes.len().min(CHUNK)].to_vec();
                    let pending = write.submit(id, move |opened| {
                        opened.check(true)?;
                        loop {
                            match opened.file.write(&bytes) {
                                Err(error) if error.kind() == io::ErrorKind::Interrupted => {
                                    continue;
                                }
                                result => return result,
                            }
                        }
                    });
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            ops.io_ops(
                "op_file_writev",
                "op_file_writev_sync",
                move |(id, buffers): (u64, Vec<JsBuffer>)| {
                    let pending = writev.submit_observed(id, move |opened, closed| {
                        opened.check(true)?;
                        let mut total = 0;
                        let mut elem = 0;
                        let mut inner = 0;
                        while elem < buffers.len() {
                            while elem < buffers.len() && inner >= buffers[elem].len() {
                                elem += 1;
                                inner = 0;
                            }
                            if elem >= buffers.len() {
                                break;
                            }
                            if closed.load(Ordering::Acquire) {
                                return Err(bad_resource());
                            }
                            opened.check(true)?;
                            let mut slices = Vec::new();
                            let mut budget = CHUNK;
                            let mut scan = elem;
                            let mut scan_inner = inner;
                            while scan < buffers.len() && slices.len() < MAX_SLICES && budget > 0 {
                                let bytes: &[u8] = &buffers[scan];
                                let available = bytes.len().saturating_sub(scan_inner);
                                if available == 0 {
                                    scan += 1;
                                    scan_inner = 0;
                                    continue;
                                }
                                let take = available.min(budget);
                                slices
                                    .push(io::IoSlice::new(&bytes[scan_inner..scan_inner + take]));
                                budget -= take;
                                scan_inner += take;
                                if scan_inner >= bytes.len() {
                                    scan += 1;
                                    scan_inner = 0;
                                }
                            }
                            if slices.is_empty() {
                                break;
                            }
                            let count = loop {
                                match opened.file.write_vectored(&slices) {
                                    Err(error) if error.kind() == io::ErrorKind::Interrupted => {
                                        continue;
                                    }
                                    result => break result?,
                                }
                            };
                            if count == 0 {
                                return Err(io::Error::new(
                                    io::ErrorKind::WriteZero,
                                    "File write made no progress",
                                ));
                            }
                            total += count;
                            let mut skip = count;
                            while skip > 0 {
                                let remaining = buffers[elem].len() - inner;
                                if remaining == 0 {
                                    elem += 1;
                                    inner = 0;
                                    continue;
                                }
                                if skip < remaining {
                                    inner += skip;
                                    break;
                                }
                                skip -= remaining;
                                elem += 1;
                                inner = 0;
                            }
                        }
                        Ok(total)
                    });
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            ops.io_ops(
                "op_file_seek",
                "op_file_seek_sync",
                move |(id, offset, whence): (u64, i64, u8)| {
                    let pending = seek.submit(id, move |opened| {
                        opened.check_metadata()?;
                        if offset.unsigned_abs() > SAFE_INTEGER {
                            return Err(io::Error::new(
                                io::ErrorKind::InvalidInput,
                                "Unsafe seek offset",
                            ));
                        }
                        let origin = match whence {
                            0 => 0,
                            1 => opened.file.stream_position()?,
                            2 => opened.file.metadata()?.len(),
                            _ => {
                                return Err(io::Error::new(
                                    io::ErrorKind::InvalidInput,
                                    "Invalid seek mode",
                                ));
                            }
                        };
                        let position = origin
                            .checked_add_signed(offset)
                            .filter(|position| *position <= SAFE_INTEGER)
                            .ok_or_else(|| {
                                io::Error::new(io::ErrorKind::InvalidInput, "Unsafe seek position")
                            })?;
                        let position = opened.file.seek(SeekFrom::Start(position))?;
                        Ok(position)
                    });
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            ops.io_ops(
                "op_file_truncate",
                "op_file_truncate_sync",
                move |(id, length): (u64, u64)| {
                    let pending = truncate.submit(id, move |opened| {
                        opened.check(true)?;
                        if length > SAFE_INTEGER {
                            return Err(io::Error::new(
                                io::ErrorKind::InvalidInput,
                                "Unsafe truncate length",
                            ));
                        }
                        opened.file.set_len(length).map(|_| true)
                    });
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            ops.io_ops(
                "op_file_sync",
                "op_file_sync_sync",
                move |(id, data): (u64, bool)| {
                    let pending = sync.submit(id, move |opened| {
                        opened.check(true)?;
                        if data {
                            opened.file.sync_data()?;
                        } else {
                            opened.file.sync_all()?;
                        }
                        Ok(true)
                    });
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            ops.io_ops("op_file_stat", "op_file_stat_sync", move |id: u64| {
                let pending = stat.submit(id, |opened| {
                    opened.check_metadata()?;
                    opened.file.metadata().map(info)
                });
                async move { Ok::<_, Infallible>(receive(pending).await) }
            }),
            ops.io_ops(
                "op_file_path_stat",
                "op_file_path_stat_sync",
                move |(path, follow): (String, bool)| {
                    let caps = path_stat.caps.clone();
                    let pending = path_stat.pool.run(move || {
                        if path.is_empty() {
                            return Err(io::Error::new(
                                io::ErrorKind::NotFound,
                                "Empty filesystem path",
                            ));
                        }
                        let follow = follow || paths::follows_link(&path);
                        let absolute = filesystem_path(&caps, &path)?;
                        let target = if follow {
                            PathTarget::for_creation(&absolute)
                        } else {
                            PathTarget::for_entry(&absolute)
                        };
                        let target = match target {
                            Ok(target) => target,
                            Err(_) if !caps.unrestricted_read() => {
                                return Err(denied(&path, false));
                            }
                            Err(error) => return Err(error),
                        };
                        authorize(&caps, &target, &path, false)?;
                        let (directory, name) = parent(target.as_path())?;
                        let metadata = directory.symlink_metadata(name)?;
                        if follow && metadata.is_symlink() {
                            return Err(io::Error::new(
                                io::ErrorKind::NotFound,
                                "Symbolic link changed or has a missing target",
                            ));
                        }
                        Ok(info(metadata))
                    });
                    async move { Ok::<_, Infallible>(receive(pending).await) }
                },
            ),
            vec![Op::new("op_file_close", move |(id, quiet): (u64, bool)| {
                let resource = state.resources.lock().unwrap().remove(&id);
                let result = match resource {
                    Some(resource) => {
                        resource.close();
                        Ok(true)
                    }
                    None if quiet => Ok(true),
                    None => Err(bad_resource()),
                };
                Ok::<_, Infallible>(response(result))
            })],
        ]
        .into_iter()
        .flatten()
        .collect(),
        modules: const {
            &[
                (
                    "ext:file/main.js",
                    include_str!(concat!(env!("OUT_DIR"), "/main.js")),
                ),
                (
                    "ext:file/errors.js",
                    include_str!(concat!(env!("OUT_DIR"), "/errors.js")),
                ),
                (
                    "ext:file/sync.js",
                    include_str!(concat!(env!("OUT_DIR"), "/sync.js")),
                ),
                PUBLIC_MODULES[0].module,
                PUBLIC_MODULES[1].module,
            ]
        },
        entry_point: None,
    };
    extension.ops.extend(link_ops);
    extension.ops.extend(path_ops);
    extension.ops.extend(posix_ops);
    extension.ops.extend(workflow_ops);
    extension.ops.extend(watch_ops);
    (extension, shared)
}

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

    #[tokio::test]
    async fn response_handles_reject_busy_files_without_consuming_them() {
        let root = tempfile::tempdir().unwrap();
        std::fs::write(root.path().join("file.txt"), b"content").unwrap();
        let caps = Capabilities::new(avril_caps::Initial::new(false), root.path().to_owned());
        let (_extension, files) = init(Handle::current(), caps, Arc::new(AtomicBool::new(false)));
        let id = files
            .open(
                "file.txt".into(),
                (true, false, false, false, false, false, None),
            )
            .await
            .unwrap()
            .unwrap();
        let (started, ready) = oneshot::channel();
        let (release, wait) = std::sync::mpsc::channel();
        let operation = files.submit(id, move |_| {
            started.send(()).unwrap();
            wait.recv().unwrap();
            Ok(())
        });
        ready.await.unwrap();
        for consume in [false, true] {
            assert_eq!(
                files.response_handle(id, consume).err().unwrap().kind(),
                io::ErrorKind::WouldBlock,
            );
        }
        release.send(()).unwrap();
        operation.await.unwrap().unwrap();
        let (name, file, check) = files.response_handle(id, true).unwrap();
        assert_eq!(name, "file.txt");
        check().unwrap();
        assert_eq!(file.metadata().unwrap().len(), 7);
        assert_eq!(
            files.response_handle(id, false).err().unwrap().kind(),
            io::ErrorKind::NotConnected,
        );
    }
}