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;
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
}
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 inner = Arc::clone(&self.inner);
let root_buf = root.to_path_buf();
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, &path);
}
})
.map_err(std::io::Error::other)?;
watcher
.watch(root, notify::RecursiveMode::Recursive)
.map_err(std::io::Error::other)?;
Ok(WatchGuard { _watcher: watcher })
}
}
fn handle_change(inner: &Inner, root: &Path, changed: &Path) {
let is_manifest = changed
.file_name()
.is_some_and(|name| name == "pushkin.toml");
if is_manifest {
let Ok(text) = std::fs::read_to_string(changed) else {
return;
};
let Ok(manifest) = Manifest::parse(&text) else {
return;
};
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();
}
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);
}
}