//! Adapts VFS observation to bounded, pollable world-path change queues.
use super::{
    State,
    bindings::sausage::process::host::{Change, ChangeKind, HostWatcher},
};
use sausage_vfs::observe::{ChangeKind as Kind, under};
use tokio::sync::broadcast::{Receiver, error::TryRecvError};
use wasmtime::component::Resource;
use wasmtime_wasi::p2::{DynPollable, Pollable, subscribe};

pub struct Watcher {
    paths: Vec<String>,
    events: Receiver<(String, Kind)>,
    pending: Vec<Change>,
}
impl Watcher {
    fn push(&mut self, path: String, kind: Kind) {
        if self.paths.iter().any(|base| under(&path, base)) {
            if self
                .pending
                .iter()
                .any(|c| matches!(c.kind, ChangeKind::Overflow))
            {
                return;
            }
            if self.pending.len() >= 256 {
                self.overflow();
                return;
            }
            self.pending.push(Change {
                path,
                kind: match kind {
                    Kind::Created => ChangeKind::Created,
                    Kind::Modified => ChangeKind::Modified,
                    Kind::Removed => ChangeKind::Removed,
                },
            });
        }
    }
    fn overflow(&mut self) {
        self.pending.clear();
        self.pending.push(Change {
            path: "/".into(),
            kind: ChangeKind::Overflow,
        });
    }
    fn drain(&mut self) {
        loop {
            match self.events.try_recv() {
                Ok((path, kind)) => self.push(path, kind),
                Err(TryRecvError::Lagged(_)) => self.overflow(),
                Err(_) => break,
            }
        }
    }
}
#[async_trait::async_trait]
impl Pollable for Watcher {
    async fn ready(&mut self) {
        loop {
            self.drain();
            if !self.pending.is_empty() {
                return;
            }
            match self.events.recv().await {
                Ok((path, kind)) => self.push(path, kind),
                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => self.overflow(),
                Err(_) => return,
            }
        }
    }
}
impl HostWatcher for State {
    async fn new(&mut self, paths: Vec<String>) -> wasmtime::Result<Resource<Watcher>> {
        let observer = self
            .runtime
            .world
            .root
            .observer()
            .ok_or_else(|| wasmtime::format_err!("filesystem has no observer"))?;
        let paths: Vec<String> = paths
            .iter()
            .map(|p| {
                sausage_vfs::resolve("/", p)
                    .trim_end_matches('/')
                    .to_owned()
            })
            .map(|p| if p.is_empty() { "/".into() } else { p })
            .collect();
        Ok(self.table.push(Watcher {
            events: observer.subscribe(paths.clone()),
            paths,
            pending: Vec::new(),
        })?)
    }
    async fn subscribe(
        &mut self,
        watcher: Resource<Watcher>,
    ) -> wasmtime::Result<Resource<DynPollable>> {
        subscribe(&mut self.table, watcher)
    }
    async fn changes(&mut self, watcher: Resource<Watcher>) -> wasmtime::Result<Vec<Change>> {
        let watcher = self.table.get_mut(&watcher)?;
        watcher.drain();
        Ok(std::mem::take(&mut watcher.pending))
    }
    async fn drop(&mut self, watcher: Resource<Watcher>) -> wasmtime::Result<()> {
        self.table.delete(watcher)?;
        Ok(())
    }
}