//! Actor-served files: `/actor/<id>/fs/...` (see PROTOCOL.md, "Actor-served
//! files").
//!
//! This module wraps the `wasi:filesystem` implementation of `wasmtime-wasi`.
//! A path under `/actor/<id>/fs/` does not reach the disk: the host asks the
//! actor `<id>` with a message from `fs` and waits up to `ASK_TIMEOUT` for
//! the reply. Every other path goes to `wasmtime-wasi` unchanged.
//!
//! - `fs_read {path}` -> `fs_file {text | base64}`, `fs_dir {entries: [{name, type}]}`
//! - `fs_write {path, text | base64}`, `fs_mkdir {path}`, `fs_remove {path}`,
//!   `fs_rename {path, to}` -> `fs_done`
//! - any of them -> `fs_error {errno}` (a `wasi:filesystem` error-code name)
//!
//! A file is read once when it is opened. After a change, it is written back
//! in one `fs_write` when it is synced or closed.

use std::collections::HashSet;
use std::hash::{Hash, Hasher};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, Weak};

use bytes::Bytes;
use serde_json::{Value, json};
use wasmtime::component::{HasData, Resource};
use wasmtime_wasi::filesystem::WasiFilesystemCtxView;
use wasmtime_wasi::p2::bindings::filesystem::types::{
    self, DescriptorFlags, DescriptorType, DirectoryEntry, ErrorCode, HostDescriptor, HostDirectoryEntryStream, OpenFlags,
};
use wasmtime_wasi::p2::bindings::filesystem::preopens;
use wasmtime_wasi::p2::pipe::MemoryInputPipe;
use wasmtime_wasi::p2::{DynInputStream, DynOutputStream, FsError, FsResult, OutputStream, Pollable, StreamResult};

use crate::broker::Broker;

/// The `from` of the host's requests. Replies to it go back to the waiting call.
pub const ID: &str = "fs";
pub const ASK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

/// Per-instance state.
pub struct VfsState {
    broker: Weak<Broker>,
    /// The actor of this instance, for notes in its `stderr.log`.
    me: String,
    /// Descriptors of the `/actor` preopen.
    roots: HashSet<u32>,
}

impl VfsState {
    pub fn new(broker: Weak<Broker>, me: String) -> Self {
        Self { broker, me, roots: HashSet::new() }
    }
}

/// [`HasData`] for the linker: the filesystem functions get a [`VfsView`].
pub struct Vfs;

impl HasData for Vfs {
    type Data<'a> = VfsView<'a>;
}

pub struct VfsView<'a> {
    pub inner: WasiFilesystemCtxView<'a>,
    pub state: &'a mut VfsState,
}

/// A descriptor of an actor-served file or folder (in the resource table in
/// place of a `wasmtime-wasi` descriptor).
#[derive(Clone)]
enum Virt {
    Dir { actor: String, path: String },
    File(VFile),
}

#[derive(Clone)]
struct VFile {
    actor: String,
    path: String,
    data: Arc<Mutex<Vec<u8>>>,
    dirty: Arc<AtomicBool>,
    write: bool,
}

/// A directory-entry-stream of an actor-served folder.
struct VDir(std::vec::IntoIter<DirectoryEntry>);

enum Node {
    File(Vec<u8>),
    Dir(Vec<DirectoryEntry>),
}

/// Sends the requests. Owned, so that no borrow of the view lives across an await.
struct Asker {
    broker: Option<Arc<Broker>>,
    me: String,
}

impl Asker {
    async fn ask(&self, actor: &str, body: Value) -> Result<Value, ErrorCode> {
        let broker = self.broker.as_ref().ok_or(ErrorCode::Io)?;
        let what = format!("{} {} to {actor}", body["type"].as_str().unwrap_or(""), body["path"].as_str().unwrap_or(""));
        match broker.ask(actor, body).await {
            None => {
                broker.log(&self.me, &format!("{what}: no reply within {} s", ASK_TIMEOUT.as_secs()));
                Err(ErrorCode::Io)
            }
            Some(reply) => match reply["type"].as_str() {
                Some("fs_error") => Err(errno(&reply["errno"])),
                Some("undeliverable") => Err(ErrorCode::NoEntry),
                _ => Ok(reply),
            },
        }
    }

    async fn read(&self, actor: &str, path: &str) -> Result<Node, ErrorCode> {
        let reply = self.ask(actor, json!({"type": "fs_read", "path": path})).await?;
        match reply["type"].as_str() {
            Some("fs_file") => content(&reply).map(Node::File),
            Some("fs_dir") => Ok(Node::Dir(
                reply["entries"]
                    .as_array()
                    .into_iter()
                    .flatten()
                    .filter_map(|e| {
                        let type_ = if e["type"] == "dir" { DescriptorType::Directory } else { DescriptorType::RegularFile };
                        Some(DirectoryEntry { type_, name: e["name"].as_str()?.to_string() })
                    })
                    .collect(),
            )),
            _ => {
                if let Some(b) = &self.broker {
                    b.log(&self.me, &format!("fs_read {path} to {actor}: unexpected reply {reply}"));
                }
                Err(ErrorCode::Io)
            }
        }
    }

    async fn change(&self, actor: &str, body: Value) -> FsResult<()> {
        self.ask(actor, body).await.map(|_| ()).map_err(FsError::from)
    }

    async fn flush(&self, f: &VFile) -> FsResult<()> {
        if f.dirty.swap(false, Ordering::SeqCst)
            && let Err(e) = self.ask(&f.actor, write_body(f)).await
        {
            f.dirty.store(true, Ordering::SeqCst);
            return Err(e.into());
        }
        Ok(())
    }
}

fn errno(v: &Value) -> ErrorCode {
    match v.as_str().unwrap_or("") {
        "no-entry" => ErrorCode::NoEntry,
        "exist" => ErrorCode::Exist,
        "not-directory" => ErrorCode::NotDirectory,
        "is-directory" => ErrorCode::IsDirectory,
        "not-empty" => ErrorCode::NotEmpty,
        "access" => ErrorCode::Access,
        "not-permitted" => ErrorCode::NotPermitted,
        "invalid" => ErrorCode::Invalid,
        "unsupported" => ErrorCode::Unsupported,
        "busy" => ErrorCode::Busy,
        _ => ErrorCode::Io,
    }
}

fn content(v: &Value) -> Result<Vec<u8>, ErrorCode> {
    if let Some(text) = v["text"].as_str() {
        return Ok(text.as_bytes().to_vec());
    }
    match v["base64"].as_str() {
        Some(b) => base64_decode(b).ok_or(ErrorCode::IllegalByteSequence),
        None => Ok(Vec::new()),
    }
}

fn write_body(f: &VFile) -> Value {
    let data = f.data.lock().unwrap().clone();
    match String::from_utf8(data) {
        Ok(text) => json!({"type": "fs_write", "path": f.path, "text": text}),
        Err(e) => json!({"type": "fs_write", "path": f.path, "base64": base64_encode(e.as_bytes())}),
    }
}

/// `rel` resolved against the folder `base`, as an absolute path.
fn join(base: &str, rel: &str) -> String {
    let mut parts: Vec<&str> = base.split('/').filter(|p| !p.is_empty()).collect();
    for p in rel.split('/') {
        match p {
            "" | "." => {}
            ".." => {
                parts.pop();
            }
            p => parts.push(p),
        }
    }
    format!("/{}", parts.join("/"))
}

fn stat(type_: DescriptorType, size: u64) -> types::DescriptorStat {
    types::DescriptorStat {
        type_,
        link_count: 1,
        size,
        data_access_timestamp: None,
        data_modification_timestamp: None,
        status_change_timestamp: None,
    }
}

fn node_stat(node: &Node) -> types::DescriptorStat {
    match node {
        Node::File(data) => stat(DescriptorType::RegularFile, data.len() as u64),
        Node::Dir(_) => stat(DescriptorType::Directory, 0),
    }
}

fn hash(actor: &str, path: &str) -> types::MetadataHashValue {
    let mut h = std::collections::hash_map::DefaultHasher::new();
    (actor, path).hash(&mut h);
    types::MetadataHashValue { lower: h.finish(), upper: 0 }
}

fn write_at(data: &mut Vec<u8>, at: usize, bytes: &[u8]) {
    if data.len() < at + bytes.len() {
        data.resize(at + bytes.len(), 0);
    }
    data[at..at + bytes.len()].copy_from_slice(bytes);
}

/// An output stream into an actor-served file (`pos: None` appends).
struct VWriter {
    file: VFile,
    pos: Option<usize>,
}

#[wasmtime_wasi::async_trait]
impl Pollable for VWriter {
    async fn ready(&mut self) {}
}

#[wasmtime_wasi::async_trait]
impl OutputStream for VWriter {
    fn write(&mut self, bytes: Bytes) -> StreamResult<()> {
        let mut data = self.file.data.lock().unwrap();
        let at = self.pos.unwrap_or(data.len());
        write_at(&mut data, at, &bytes);
        if let Some(p) = &mut self.pos {
            *p += bytes.len();
        }
        self.file.dirty.store(true, Ordering::SeqCst);
        Ok(())
    }
    fn flush(&mut self) -> StreamResult<()> {
        Ok(())
    }
    fn check_write(&mut self) -> StreamResult<usize> {
        Ok(64 * 1024)
    }
}

impl VfsView<'_> {
    fn asker(&self) -> Asker {
        Asker { broker: self.state.broker.upgrade(), me: self.state.me.clone() }
    }

    fn virt(&mut self, fd: &Resource<types::Descriptor>) -> Option<Virt> {
        self.inner.table.get_any_mut(fd.rep()).ok()?.downcast_ref::<Virt>().cloned()
    }

    fn file(&mut self, fd: &Resource<types::Descriptor>) -> Option<FsResult<VFile>> {
        match self.virt(fd)? {
            Virt::File(f) => Some(Ok(f)),
            Virt::Dir { .. } => Some(Err(ErrorCode::IsDirectory.into())),
        }
    }

    /// The actor and its path, if `path` relative to `fd` is actor-served.
    fn target(&mut self, fd: &Resource<types::Descriptor>, path: &str) -> Option<(String, String)> {
        match self.virt(fd) {
            Some(Virt::Dir { actor, path: base }) => Some((actor, join(&base, path))),
            Some(Virt::File(f)) => Some((f.actor, join(&f.path, path))),
            None if self.state.roots.contains(&fd.rep()) => {
                let full = join("/", path);
                let mut parts = full.splitn(4, '/').skip(1);
                let (actor, fs, rest) = (parts.next()?, parts.next()?, parts.next().unwrap_or(""));
                (fs == "fs" && !actor.is_empty()).then(|| (actor.to_string(), format!("/{rest}")))
            }
            None => None,
        }
    }

    fn push(&mut self, v: Virt) -> FsResult<Resource<types::Descriptor>> {
        let r = self.inner.table.push(v)?;
        Ok(Resource::new_own(r.rep()))
    }

    async fn open(&mut self, actor: String, path: String, oflags: OpenFlags, flags: DescriptorFlags) -> FsResult<Resource<types::Descriptor>> {
        let write = flags.contains(DescriptorFlags::WRITE);
        let create = oflags.contains(OpenFlags::CREATE);
        let exclusive = create && oflags.contains(OpenFlags::EXCLUSIVE);
        let virt = match self.asker().read(&actor, &path).await {
            Ok(_) if exclusive => return Err(ErrorCode::Exist.into()),
            Ok(Node::Dir(_)) if write => return Err(ErrorCode::IsDirectory.into()),
            Ok(Node::Dir(_)) => Virt::Dir { actor, path },
            Ok(Node::File(_)) if oflags.contains(OpenFlags::DIRECTORY) => return Err(ErrorCode::NotDirectory.into()),
            Ok(Node::File(data)) => {
                let truncate = oflags.contains(OpenFlags::TRUNCATE);
                let data = if truncate { Vec::new() } else { data };
                Virt::File(VFile { actor, path, data: Arc::new(Mutex::new(data)), dirty: Arc::new(AtomicBool::new(truncate)), write })
            }
            Err(ErrorCode::NoEntry) if create && !oflags.contains(OpenFlags::DIRECTORY) => Virt::File(VFile {
                actor,
                path,
                data: Arc::default(),
                dirty: Arc::new(AtomicBool::new(true)),
                write,
            }),
            Err(e) => return Err(e.into()),
        };
        self.push(virt)
    }

    /// Send the changed file without waiting (the descriptor is gone). The
    /// request leaves now, so later requests of this instance come after it.
    fn write_back_later(&self, f: &VFile) {
        let Some(broker) = self.state.broker.upgrade() else { return };
        let reply = broker.ask(&f.actor, write_body(f));
        let (me, what) = (self.state.me.clone(), format!("fs_write {} to {}", f.path, f.actor));
        tokio::spawn(async move {
            match reply.await {
                Some(r) if r["type"] != "fs_error" && r["type"] != "undeliverable" => {}
                r => broker.log(&me, &format!("{what} failed: {}", r.map_or("no reply".to_string(), |r| r.to_string()))),
            }
        });
    }
}

impl preopens::Host for VfsView<'_> {
    fn get_directories(&mut self) -> wasmtime::Result<Vec<(Resource<types::Descriptor>, String)>> {
        let dirs = preopens::Host::get_directories(&mut self.inner)?;
        for (fd, name) in &dirs {
            if name == "/actor" {
                self.state.roots.insert(fd.rep());
            }
        }
        Ok(dirs)
    }
}

impl types::Host for VfsView<'_> {
    fn convert_error_code(&mut self, err: FsError) -> wasmtime::Result<ErrorCode> {
        types::Host::convert_error_code(&mut self.inner, err)
    }

    fn filesystem_error_code(&mut self, err: Resource<wasmtime::Error>) -> wasmtime::Result<Option<ErrorCode>> {
        types::Host::filesystem_error_code(&mut self.inner, err)
    }
}

impl HostDescriptor for VfsView<'_> {
    async fn advise(&mut self, fd: Resource<types::Descriptor>, offset: types::Filesize, len: types::Filesize, advice: types::Advice) -> FsResult<()> {
        if self.virt(&fd).is_some() {
            return Ok(());
        }
        HostDescriptor::advise(&mut self.inner, fd, offset, len, advice).await
    }

    async fn sync_data(&mut self, fd: Resource<types::Descriptor>) -> FsResult<()> {
        match self.virt(&fd) {
            Some(Virt::File(f)) => self.asker().flush(&f).await,
            Some(Virt::Dir { .. }) => Ok(()),
            None => HostDescriptor::sync_data(&mut self.inner, fd).await,
        }
    }

    async fn get_flags(&mut self, fd: Resource<types::Descriptor>) -> FsResult<DescriptorFlags> {
        match self.virt(&fd) {
            Some(Virt::File(f)) if f.write => Ok(DescriptorFlags::READ | DescriptorFlags::WRITE),
            Some(Virt::File(_)) => Ok(DescriptorFlags::READ),
            Some(Virt::Dir { .. }) => Ok(DescriptorFlags::READ | DescriptorFlags::MUTATE_DIRECTORY),
            None => HostDescriptor::get_flags(&mut self.inner, fd).await,
        }
    }

    async fn get_type(&mut self, fd: Resource<types::Descriptor>) -> FsResult<DescriptorType> {
        match self.virt(&fd) {
            Some(Virt::File(_)) => Ok(DescriptorType::RegularFile),
            Some(Virt::Dir { .. }) => Ok(DescriptorType::Directory),
            None => HostDescriptor::get_type(&mut self.inner, fd).await,
        }
    }

    async fn set_size(&mut self, fd: Resource<types::Descriptor>, size: types::Filesize) -> FsResult<()> {
        match self.file(&fd) {
            Some(f) => {
                let f = f?;
                if !f.write {
                    return Err(ErrorCode::NotPermitted.into());
                }
                f.data.lock().unwrap().resize(size as usize, 0);
                f.dirty.store(true, Ordering::SeqCst);
                Ok(())
            }
            None => HostDescriptor::set_size(&mut self.inner, fd, size).await,
        }
    }

    async fn set_times(&mut self, fd: Resource<types::Descriptor>, atim: types::NewTimestamp, mtim: types::NewTimestamp) -> FsResult<()> {
        if self.virt(&fd).is_some() {
            return Ok(());
        }
        HostDescriptor::set_times(&mut self.inner, fd, atim, mtim).await
    }

    async fn read(&mut self, fd: Resource<types::Descriptor>, len: types::Filesize, offset: types::Filesize) -> FsResult<(Vec<u8>, bool)> {
        match self.file(&fd) {
            Some(f) => {
                let f = f?;
                let data = f.data.lock().unwrap();
                let start = usize::try_from(offset).unwrap_or(usize::MAX).min(data.len());
                let end = start.saturating_add(usize::try_from(len).unwrap_or(usize::MAX)).min(data.len());
                Ok((data[start..end].to_vec(), start == end))
            }
            None => HostDescriptor::read(&mut self.inner, fd, len, offset).await,
        }
    }

    async fn write(&mut self, fd: Resource<types::Descriptor>, buf: Vec<u8>, offset: types::Filesize) -> FsResult<types::Filesize> {
        match self.file(&fd) {
            Some(f) => {
                let f = f?;
                if !f.write {
                    return Err(ErrorCode::NotPermitted.into());
                }
                write_at(&mut f.data.lock().unwrap(), offset as usize, &buf);
                f.dirty.store(true, Ordering::SeqCst);
                Ok(buf.len() as u64)
            }
            None => HostDescriptor::write(&mut self.inner, fd, buf, offset).await,
        }
    }

    async fn read_directory(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Resource<types::DirectoryEntryStream>> {
        match self.virt(&fd) {
            Some(Virt::Dir { actor, path }) => match self.asker().read(&actor, &path).await? {
                Node::Dir(entries) => {
                    let r = self.inner.table.push(VDir(entries.into_iter()))?;
                    Ok(Resource::new_own(r.rep()))
                }
                Node::File(_) => Err(ErrorCode::NotDirectory.into()),
            },
            Some(Virt::File(_)) => Err(ErrorCode::NotDirectory.into()),
            None => HostDescriptor::read_directory(&mut self.inner, fd).await,
        }
    }

    async fn sync(&mut self, fd: Resource<types::Descriptor>) -> FsResult<()> {
        match self.virt(&fd) {
            Some(Virt::File(f)) => self.asker().flush(&f).await,
            Some(Virt::Dir { .. }) => Ok(()),
            None => HostDescriptor::sync(&mut self.inner, fd).await,
        }
    }

    async fn create_directory_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<()> {
        match self.target(&fd, &path) {
            Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_mkdir", "path": path})).await,
            None => HostDescriptor::create_directory_at(&mut self.inner, fd, path).await,
        }
    }

    async fn stat(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::DescriptorStat> {
        match self.virt(&fd) {
            Some(Virt::File(f)) => Ok(stat(DescriptorType::RegularFile, f.data.lock().unwrap().len() as u64)),
            Some(Virt::Dir { .. }) => Ok(stat(DescriptorType::Directory, 0)),
            None => HostDescriptor::stat(&mut self.inner, fd).await,
        }
    }

    async fn stat_at(&mut self, fd: Resource<types::Descriptor>, path_flags: types::PathFlags, path: String) -> FsResult<types::DescriptorStat> {
        match self.target(&fd, &path) {
            Some((actor, path)) => Ok(node_stat(&self.asker().read(&actor, &path).await?)),
            None => HostDescriptor::stat_at(&mut self.inner, fd, path_flags, path).await,
        }
    }

    async fn set_times_at(
        &mut self,
        fd: Resource<types::Descriptor>,
        path_flags: types::PathFlags,
        path: String,
        atim: types::NewTimestamp,
        mtim: types::NewTimestamp,
    ) -> FsResult<()> {
        if self.target(&fd, &path).is_some() {
            return Ok(());
        }
        HostDescriptor::set_times_at(&mut self.inner, fd, path_flags, path, atim, mtim).await
    }

    async fn link_at(
        &mut self,
        fd: Resource<types::Descriptor>,
        old_path_flags: types::PathFlags,
        old_path: String,
        new_descriptor: Resource<types::Descriptor>,
        new_path: String,
    ) -> FsResult<()> {
        if self.target(&fd, &old_path).is_some() || self.target(&new_descriptor, &new_path).is_some() {
            return Err(ErrorCode::Unsupported.into());
        }
        HostDescriptor::link_at(&mut self.inner, fd, old_path_flags, old_path, new_descriptor, new_path).await
    }

    async fn open_at(
        &mut self,
        fd: Resource<types::Descriptor>,
        path_flags: types::PathFlags,
        path: String,
        oflags: OpenFlags,
        flags: DescriptorFlags,
    ) -> FsResult<Resource<types::Descriptor>> {
        match self.target(&fd, &path) {
            Some((actor, path)) => self.open(actor, path, oflags, flags).await,
            None => HostDescriptor::open_at(&mut self.inner, fd, path_flags, path, oflags, flags).await,
        }
    }

    fn drop(&mut self, fd: Resource<types::Descriptor>) -> wasmtime::Result<()> {
        if self.virt(&fd).is_some() {
            if let Virt::File(f) = self.inner.table.delete(Resource::<Virt>::new_own(fd.rep()))?
                && f.dirty.load(Ordering::SeqCst)
            {
                self.write_back_later(&f);
            }
            return Ok(());
        }
        self.state.roots.remove(&fd.rep());
        HostDescriptor::drop(&mut self.inner, fd)
    }

    async fn readlink_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<String> {
        if self.target(&fd, &path).is_some() {
            return Err(ErrorCode::Invalid.into());
        }
        HostDescriptor::readlink_at(&mut self.inner, fd, path).await
    }

    async fn remove_directory_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<()> {
        match self.target(&fd, &path) {
            Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_remove", "path": path})).await,
            None => HostDescriptor::remove_directory_at(&mut self.inner, fd, path).await,
        }
    }

    async fn rename_at(&mut self, fd: Resource<types::Descriptor>, old_path: String, new_fd: Resource<types::Descriptor>, new_path: String) -> FsResult<()> {
        match (self.target(&fd, &old_path), self.target(&new_fd, &new_path)) {
            (None, None) => HostDescriptor::rename_at(&mut self.inner, fd, old_path, new_fd, new_path).await,
            (Some((a, from)), Some((b, to))) if a == b => {
                self.asker().change(&a, json!({"type": "fs_rename", "path": from, "to": to})).await
            }
            _ => Err(ErrorCode::Unsupported.into()),
        }
    }

    async fn symlink_at(&mut self, fd: Resource<types::Descriptor>, src_path: String, dest_path: String) -> FsResult<()> {
        if self.target(&fd, &dest_path).is_some() {
            return Err(ErrorCode::Unsupported.into());
        }
        HostDescriptor::symlink_at(&mut self.inner, fd, src_path, dest_path).await
    }

    async fn unlink_file_at(&mut self, fd: Resource<types::Descriptor>, path: String) -> FsResult<()> {
        match self.target(&fd, &path) {
            Some((actor, path)) => self.asker().change(&actor, json!({"type": "fs_remove", "path": path})).await,
            None => HostDescriptor::unlink_file_at(&mut self.inner, fd, path).await,
        }
    }

    fn read_via_stream(&mut self, fd: Resource<types::Descriptor>, offset: types::Filesize) -> FsResult<Resource<DynInputStream>> {
        match self.file(&fd) {
            Some(f) => {
                let f = f?;
                let data = f.data.lock().unwrap();
                let start = usize::try_from(offset).unwrap_or(usize::MAX).min(data.len());
                let stream: DynInputStream = Box::new(MemoryInputPipe::new(data[start..].to_vec()));
                std::mem::drop(data);
                Ok(self.inner.table.push(stream)?)
            }
            None => HostDescriptor::read_via_stream(&mut self.inner, fd, offset),
        }
    }

    fn write_via_stream(&mut self, fd: Resource<types::Descriptor>, offset: types::Filesize) -> FsResult<Resource<DynOutputStream>> {
        match self.file(&fd) {
            Some(f) => {
                let f = f?;
                if !f.write {
                    return Err(ErrorCode::NotPermitted.into());
                }
                let stream: DynOutputStream = Box::new(VWriter { file: f, pos: Some(offset as usize) });
                Ok(self.inner.table.push(stream)?)
            }
            None => HostDescriptor::write_via_stream(&mut self.inner, fd, offset),
        }
    }

    fn append_via_stream(&mut self, fd: Resource<types::Descriptor>) -> FsResult<Resource<DynOutputStream>> {
        match self.file(&fd) {
            Some(f) => {
                let f = f?;
                if !f.write {
                    return Err(ErrorCode::NotPermitted.into());
                }
                let stream: DynOutputStream = Box::new(VWriter { file: f, pos: None });
                Ok(self.inner.table.push(stream)?)
            }
            None => HostDescriptor::append_via_stream(&mut self.inner, fd),
        }
    }

    async fn is_same_object(&mut self, a: Resource<types::Descriptor>, b: Resource<types::Descriptor>) -> wasmtime::Result<bool> {
        let key = |v: Virt| match v {
            Virt::Dir { actor, path } => (actor, path),
            Virt::File(f) => (f.actor, f.path),
        };
        match (self.virt(&a), self.virt(&b)) {
            (None, None) => HostDescriptor::is_same_object(&mut self.inner, a, b).await,
            (Some(x), Some(y)) => Ok(key(x) == key(y)),
            _ => Ok(false),
        }
    }

    async fn metadata_hash(&mut self, fd: Resource<types::Descriptor>) -> FsResult<types::MetadataHashValue> {
        match self.virt(&fd) {
            Some(Virt::Dir { actor, path }) => Ok(hash(&actor, &path)),
            Some(Virt::File(f)) => Ok(hash(&f.actor, &f.path)),
            None => HostDescriptor::metadata_hash(&mut self.inner, fd).await,
        }
    }

    async fn metadata_hash_at(&mut self, fd: Resource<types::Descriptor>, path_flags: types::PathFlags, path: String) -> FsResult<types::MetadataHashValue> {
        match self.target(&fd, &path) {
            Some((actor, path)) => Ok(hash(&actor, &path)),
            None => HostDescriptor::metadata_hash_at(&mut self.inner, fd, path_flags, path).await,
        }
    }
}

impl HostDirectoryEntryStream for VfsView<'_> {
    async fn read_directory_entry(&mut self, stream: Resource<types::DirectoryEntryStream>) -> FsResult<Option<DirectoryEntry>> {
        if let Some(dir) = self.inner.table.get_any_mut(stream.rep()).ok().and_then(|a| a.downcast_mut::<VDir>()) {
            return Ok(dir.0.next());
        }
        HostDirectoryEntryStream::read_directory_entry(&mut self.inner, stream).await
    }

    fn drop(&mut self, stream: Resource<types::DirectoryEntryStream>) -> wasmtime::Result<()> {
        let ours = self.inner.table.get_any_mut(stream.rep()).ok().is_some_and(|a| a.is::<VDir>());
        if ours {
            self.inner.table.delete(Resource::<VDir>::new_own(stream.rep()))?;
            return Ok(());
        }
        HostDirectoryEntryStream::drop(&mut self.inner, stream)
    }
}

const B64: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";

fn base64_encode(bytes: &[u8]) -> String {
    let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
    for chunk in bytes.chunks(3) {
        let n = chunk.iter().enumerate().fold(0u32, |n, (i, b)| n | (*b as u32) << (16 - 8 * i));
        for i in 0..4 {
            out.push(if i <= chunk.len() { B64[(n >> (18 - 6 * i) & 63) as usize] as char } else { '=' });
        }
    }
    out
}

fn base64_decode(text: &str) -> Option<Vec<u8>> {
    let (mut out, mut n, mut bits) = (Vec::new(), 0u32, 0);
    for c in text.bytes().filter(|c| !c.is_ascii_whitespace() && *c != b'=') {
        n = (n << 6 | B64.iter().position(|b| *b == c)? as u32) & 0xFF_FFFF;
        bits += 6;
        if bits >= 8 {
            bits -= 8;
            out.push((n >> bits) as u8);
        }
    }
    Some(out)
}

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

    #[test]
    fn base64_round_trip() {
        for s in ["", "a", "ab", "abc", "abcd", "\u{0}\u{ff}binary"] {
            let enc = base64_encode(s.as_bytes());
            assert_eq!(base64_decode(&enc).unwrap(), s.as_bytes());
        }
        assert_eq!(base64_encode(b"hello"), "aGVsbG8=");
    }

    #[test]
    fn join_resolves_dots() {
        assert_eq!(join("/a/b", "../c/./d"), "/a/c/d");
        assert_eq!(join("/", "x/fs"), "/x/fs");
        assert_eq!(join("/", ".."), "/");
    }
}