pushkin-daemon 0.2.0

//! Warm per-file state + watcher (spec §8.1): verdicts memoized by
//! (path, content-hash), invalidated by keyed path events, cleared
//! wholesale on manifest change. The memo NEVER re-decides — a hit
//! returns the stored envelope, a miss runs the same `check_write` the
//! cold path runs, so warm and cold are identical by construction.

use notify::Watcher as _;
use pushkin_core::envelope::CheckResult;
use pushkin_core::manifest::Manifest;
use pushkin_core::pipeline::{check_write, WriteRequest};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};

/// FNV-1a over the write content — the project's stable-hash convention
/// (spec §7.4 chose it for nudge arms because std hashers are not
/// contractually stable; the memo key reuses that decision).
fn fnv1a(bytes: &[u8]) -> u64 {
    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
    for byte in bytes {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
    }
    hash
}

/// The manifest filename. Only the *resolved* one reloads (F73 phase 3);
/// the name alone is not enough to identify it.
const MANIFEST_NAME: &str = "pushkin.toml";

type Memo = HashMap<String, (u64, CheckResult)>;

struct Inner {
    manifest: RwLock<Manifest>,
    memo: Mutex<Memo>,
    hits: AtomicU64,
}

/// Memoized check state shared between the accept loop and the watcher.
pub struct WarmState {
    inner: Arc<Inner>,
}

/// Keeps the filesystem watcher alive; dropping it stops watching.
pub struct WatchGuard {
    _watcher: notify::RecommendedWatcher,
}

impl WarmState {
    #[must_use]
    pub fn new(manifest: Manifest) -> Self {
        Self {
            inner: Arc::new(Inner {
                manifest: RwLock::new(manifest),
                memo: Mutex::new(HashMap::new()),
                hits: AtomicU64::new(0),
            }),
        }
    }

    /// A handle to the same underlying state (memo, manifest, hit
    /// counter are shared, not copied) — one per connection task.
    #[must_use]
    pub fn share(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }

    /// Warm-or-compute a verdict. A hit returns the stored envelope for
    /// the same (path, content); a miss computes with the SAME
    /// `check_write` the cold path uses and stores the result.
    #[must_use]
    pub fn check(&self, request: &WriteRequest) -> CheckResult {
        let content_hash = fnv1a(request.content.as_bytes());
        if let Ok(memo) = self.inner.memo.lock() {
            if let Some((stored_hash, stored)) = memo.get(&request.file_path) {
                if *stored_hash == content_hash {
                    self.inner.hits.fetch_add(1, Ordering::SeqCst);
                    return stored.clone();
                }
            }
        }
        let result = match self.inner.manifest.read() {
            Ok(manifest) => check_write(&manifest, request),
            Err(poisoned) => check_write(&poisoned.into_inner(), request),
        };
        if let Ok(mut memo) = self.inner.memo.lock() {
            memo.insert(request.file_path.clone(), (content_hash, result.clone()));
        }
        result
    }

    /// Memo hits since construction.
    #[must_use]
    pub fn hits(&self) -> u64 {
        self.inner.hits.load(Ordering::SeqCst)
    }

    /// Replace the manifest and drop every memoized verdict (mappings
    /// shape every verdict; nothing memoized survives a manifest change).
    pub fn swap_manifest(&self, manifest: Manifest) {
        match self.inner.manifest.write() {
            Ok(mut slot) => *slot = manifest,
            Err(poisoned) => *poisoned.into_inner() = manifest,
        }
        if let Ok(mut memo) = self.inner.memo.lock() {
            memo.clear();
        }
    }

    /// Drop memoized verdicts for one path; untouched paths stay warm.
    pub fn invalidate_path(&self, path: &str) {
        if let Ok(mut memo) = self.inner.memo.lock() {
            memo.remove(path);
        }
    }

    /// Watch `root` recursively: a `pushkin.toml` change reloads the
    /// manifest wholesale (parse failures keep the old manifest — a
    /// half-saved edit must not tear down the gate); any other file
    /// change invalidates that file's memo entry, keyed by its
    /// root-relative path.
    ///
    /// # Errors
    /// `std::io::Error` when the watcher cannot be created or attached.
    pub fn watch(&self, root: &Path) -> std::io::Result<WatchGuard> {
        let governing = root.join(MANIFEST_NAME);
        self.watch_governing(root, &governing)
    }

    /// `watch`, with the governing manifest named explicitly (F73 phase 3).
    ///
    /// Only `governing` reloads. Every other `pushkin.toml` under the watch
    /// root is a file the daemon does not answer to — adopting one let an agent
    /// disarm the gate by writing a file the gate permits, because
    /// `protected_paths = ["pushkin.toml"]` does not match a nested path.
    ///
    /// # Errors
    /// `std::io::Error` when the watcher cannot be created or attached.
    pub fn watch_governing(&self, root: &Path, governing: &Path) -> std::io::Result<WatchGuard> {
        let inner = Arc::clone(&self.inner);
        let root_buf = root.to_path_buf();
        let governing_buf = canonical_or_owned(governing);
        let mut watcher =
            notify::recommended_watcher(move |event: Result<notify::Event, notify::Error>| {
                let Ok(event) = event else { return };
                for path in event.paths {
                    handle_change(&inner, &root_buf, &governing_buf, &path);
                }
            })
            .map_err(std::io::Error::other)?;
        watcher
            .watch(root, notify::RecursiveMode::Recursive)
            .map_err(std::io::Error::other)?;
        Ok(WatchGuard { _watcher: watcher })
    }
}

/// Resolve symlinks and `..` so the predicate compares real locations, not
/// spellings — macOS hands back `/private/var/...` for a `/var/...` watch. A
/// path that cannot be canonicalized (it may not exist yet) is compared as
/// given, which is the conservative direction: a miss declines a reload rather
/// than granting one.
fn canonical_or_owned(path: &Path) -> PathBuf {
    path.canonicalize().unwrap_or_else(|_| path.to_path_buf())
}

fn handle_change(inner: &Inner, root: &Path, governing: &Path, changed: &Path) {
    if canonical_or_owned(changed) == governing {
        reload_manifest(inner, changed);
        return;
    }
    if changed
        .file_name()
        .is_some_and(|name| name == MANIFEST_NAME)
    {
        // Recorded, not merely skipped (standing rule: every degradation is
        // recorded). A nested manifest changing and nothing happening is
        // indistinguishable from a dead watcher, and telling those two apart by
        // experiment once cost three edit mechanisms and a discriminator.
        eprintln!(
            "pushkin daemon: declined to reload from {} — it is not the governing \
             manifest ({}). Only the resolved manifest reloads.",
            changed.display(),
            governing.display()
        );
        return;
    }
    let key = changed
        .strip_prefix(root)
        .unwrap_or(changed)
        .to_string_lossy()
        .into_owned();
    if let Ok(mut memo) = inner.memo.lock() {
        memo.remove(&key);
    }
}

/// Reload wholesale. A governing manifest that no longer loads STOPS the
/// daemon (F71 Phase B, ruled question 4): a daemon serving rules its
/// manifest no longer states is not fail-open, it is confidently wrong. The
/// stop is recorded on stderr, never silent (standing rule), and clients
/// fall back to the cold pipeline — which denies the same state, so the
/// half-save case costs a daemon restart, not a gate hole.
fn reload_manifest(inner: &Inner, path: &Path) {
    // Transient vs durable, learned on CI (F76's platform shape): inotify
    // delivers the change event MID-WRITE, so the first read can see a
    // truncated file — the half-save the original guard protected. One
    // settle-and-re-read separates that race from a manifest that is durably
    // unloadable; only the second failure stops the daemon.
    let mut outcome = load(path);
    if outcome.is_err() {
        std::thread::sleep(std::time::Duration::from_millis(200));
        outcome = load(path);
    }
    let manifest = match outcome {
        Ok(manifest) => manifest,
        Err(reason) => {
            eprintln!(
                "pushkin daemon: the governing manifest {} no longer parses \
                 ({reason}). Serving STOPPED — stale rules are worse than no \
                 daemon; checks fall back to the cold pipeline until the \
                 manifest is fixed and the daemon restarted.",
                path.display()
            );
            std::process::exit(1);
        }
    };
    match inner.manifest.write() {
        Ok(mut slot) => *slot = manifest,
        Err(poisoned) => *poisoned.into_inner() = manifest,
    }
    // Mappings shape every verdict; nothing memoized survives a manifest change.
    if let Ok(mut memo) = inner.memo.lock() {
        memo.clear();
    }
}

/// One read-and-parse attempt, error as prose for the stop record.
fn load(path: &Path) -> Result<Manifest, String> {
    std::fs::read_to_string(path)
        .map_err(|error| error.to_string())
        .and_then(|text| Manifest::parse(&text).map_err(|error| error.to_string()))
}