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};
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
}
const MANIFEST_NAME: &str = "pushkin.toml";
type Memo = HashMap<String, (u64, CheckResult)>;
struct Inner {
manifest: RwLock<Manifest>,
memo: Mutex<Memo>,
hits: AtomicU64,
}
pub struct WarmState {
inner: Arc<Inner>,
}
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),
}),
}
}
#[must_use]
pub fn share(&self) -> Self {
Self {
inner: Arc::clone(&self.inner),
}
}
#[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
}
#[must_use]
pub fn hits(&self) -> u64 {
self.inner.hits.load(Ordering::SeqCst)
}
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();
}
}
pub fn invalidate_path(&self, path: &str) {
if let Ok(mut memo) = self.inner.memo.lock() {
memo.remove(path);
}
}
pub fn watch(&self, root: &Path) -> std::io::Result<WatchGuard> {
let governing = root.join(MANIFEST_NAME);
self.watch_governing(root, &governing)
}
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 })
}
}
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)
{
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);
}
}
fn reload_manifest(inner: &Inner, path: &Path) {
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,
}
if let Ok(mut memo) = inner.memo.lock() {
memo.clear();
}
}
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()))
}