//! Armed plugin-file watches. A Notify permit coalesces changes without losing a later change.
use crate::{Host, RpcResult, error};
use sausage_vfs::observe::under;
use serde_json::{Value, json};
use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
};
use tokio::sync::Notify;
use uuid::Uuid;

pub(crate) struct Watch {
    path: String,
    changed: Arc<Notify>,
}
pub(crate) type Watches = Arc<Mutex<HashMap<Uuid, Watch>>>;
pub(crate) fn changed(watches: &Watches, path: &str) {
    let targets: Vec<_> = watches
        .lock()
        .unwrap()
        .values()
        // Removing/moving an ancestor also invalidates a watch inside that subtree.
        .filter(|w| under(path, &w.path) || under(&w.path, path))
        .map(|w| w.changed.clone())
        .collect();
    for changed in targets {
        changed.notify_one();
    }
}
struct Registration {
    watches: Watches,
    key: Uuid,
}
impl Drop for Registration {
    fn drop(&mut self) {
        self.watches.lock().unwrap().remove(&self.key);
    }
}
fn path(caller: &str, path: &str) -> String {
    let path = sausage_vfs::resolve("/", path);
    let path = path.trim_end_matches('/');
    if path.is_empty() {
        return "/".into();
    }
    if under(path, "/self") {
        format!("/proc/{caller}{}", &path[5..])
    } else {
        path.into()
    }
}
impl Host {
    pub(crate) async fn watch(&self, caller: &str, id: Option<Value>, target: &str) -> RpcResult {
        let id = id.ok_or_else(|| error(-32600, "watch needs a request ID"))?;
        let mut ended = self
            .0
            .state
            .lock()
            .unwrap()
            .processes
            .get(caller)
            .filter(|p| p.info.state == "running")
            .map(|p| p.ended.subscribe())
            .ok_or_else(|| error(-32600, "watch needs a running caller process"))?;
        let changed = Arc::new(Notify::new());
        let registration = Registration {
            watches: self.0.watches.clone(),
            key: Uuid::now_v7(),
        };
        self.0.watches.lock().unwrap().insert(
            registration.key,
            Watch {
                path: path(caller, target),
                changed: changed.clone(),
            },
        );
        self.call_notify(caller, &id, "armed", json!({}));
        loop {
            tokio::select! {
                biased;
                _ = ended.wait_for(|ended| *ended) => return Ok(Value::Null),
                _ = changed.notified() => self.call_notify(caller, &id, "changed", json!({})),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    /// Pending changes coalesce; a change after consuming the last permit is retained.
    #[tokio::test]
    async fn coalescing_and_cleanup() {
        let watches = Watches::default();
        let n = Arc::new(Notify::new());
        let r = Registration {
            watches: watches.clone(),
            key: Uuid::now_v7(),
        };
        watches.lock().unwrap().insert(
            r.key,
            Watch {
                path: "/mailbox".into(),
                changed: n.clone(),
            },
        );
        changed(&watches, "/mailboxes/a");
        assert!(
            tokio::time::timeout(std::time::Duration::ZERO, n.notified())
                .await
                .is_err()
        );
        changed(&watches, "/mailbox/a");
        changed(&watches, "/mailbox/b");
        n.notified().await;
        assert!(
            tokio::time::timeout(std::time::Duration::ZERO, n.notified())
                .await
                .is_err()
        );
        changed(&watches, "/mailbox/c");
        n.notified().await;
        drop(r);
        assert!(watches.lock().unwrap().is_empty());
        changed(&watches, "/mailbox/d");
        assert!(
            tokio::time::timeout(std::time::Duration::ZERO, n.notified())
                .await
                .is_err()
        );
    }
    #[test]
    fn self_alias_and_normalization() {
        assert_eq!(path("writer", "/self/../self/a/"), "/proc/writer/a");
        assert_eq!(path("writer", "/selfish/a"), "/selfish/a");
        assert_eq!(path("writer", "/"), "/");
    }
}

#[cfg(test)]
mod lifetime_tests {
    use super::*;
    /// The real caller's ended event releases the registration, without a cancellation RPC.
    #[tokio::test]
    async fn caller_death_removes_watch() {
        let temp = tempfile::tempdir().unwrap();
        let runtime = crate::runtime::Runtime::new(&temp.path().join("cache")).unwrap();
        let host = Host::test_world(&temp.path().join("world"), runtime)
            .await
            .unwrap();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        host.register_native(
            "holder",
            Arc::new(move |ctx, _, _| {
                let tx = tx.clone();
                Box::pin(async move {
                    tx.send(ctx.pid.clone()).unwrap();
                    std::future::pending().await
                })
            }),
        );
        let h = host.clone();
        let task = tokio::spawn(async move { h.call("holder.hold", json!({})).await });
        let pid = rx.recv().await.unwrap();
        let mut watch = Box::pin(host.watch(&pid, Some(json!("watch")), "/mailbox"));
        assert!(
            watch
                .as_mut()
                .poll(&mut std::task::Context::from_waker(std::task::Waker::noop()))
                .is_pending()
        );
        assert_eq!(host.0.watches.lock().unwrap().len(), 1);
        host.call("core.kill", json!({"pid":pid})).await.unwrap();
        watch.await.unwrap();
        assert!(host.0.watches.lock().unwrap().is_empty());
        assert!(task.await.unwrap().is_err());
        host.stop().await;
    }
}