car-workflow 0.26.0

//! Persistent content-hash deduplication for workflow items.
//!
//! Powers the [`Dedup`](crate::StageStep::Dedup) stage step and the external-item
//! automation recipe. An automation that polls a source on a schedule must
//! process each item exactly once *across runs*, but workflow state is per-run —
//! so the seen-set lives on disk, keyed by a stable content hash of each item.
//!
//! The store is deliberately separate from the engine (like
//! [`CheckpointStore`](crate::CheckpointStore)) so the hashing/persistence logic
//! stays unit-testable in isolation.

use std::collections::{BTreeMap, HashSet};
use std::fmt::Write as _;
use std::fs;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};

/// A file-backed set of content hashes already seen, scoped to a caller-chosen
/// store namespace. One JSON file per store: `<dir>/<store>.json`.
#[derive(Debug, Clone)]
pub struct DedupStore {
    path: PathBuf,
}

/// On-disk shape: hash → first-seen RFC3339 timestamp. The timestamp drives TTL
/// eviction (see [`DedupStore::prune_older_than`]); membership is all the
/// dedup check itself reads. A `BTreeMap` keeps the file diff-stable (sorted)
/// across writes.
#[derive(Debug, Default, Serialize, Deserialize)]
struct SeenSet {
    #[serde(default)]
    hashes: BTreeMap<String, String>,
}

/// Drop entries whose first-seen timestamp is strictly older than `cutoff`,
/// returning the number removed. An entry whose timestamp can't be parsed is
/// **kept** — fail-safe, so a corrupt/legacy timestamp never causes silent
/// reprocessing. Shared by inline (TTL) and explicit (`prune_older_than`)
/// eviction.
fn prune_in_place(set: &mut SeenSet, cutoff: chrono::DateTime<chrono::Utc>) -> usize {
    let before = set.hashes.len();
    set.hashes.retain(|_, ts| {
        match chrono::DateTime::parse_from_rfc3339(ts) {
            Ok(seen) => seen.with_timezone(&chrono::Utc) >= cutoff,
            Err(_) => true,
        }
    });
    before - set.hashes.len()
}

impl DedupStore {
    /// Open the store for namespace `store` under `dir`, creating `dir` if
    /// needed. `store` is sanitized to a safe filename so an attacker-influenced
    /// namespace can't escape `dir`.
    pub fn open(dir: impl AsRef<Path>, store: &str) -> std::io::Result<Self> {
        let dir = dir.as_ref().to_path_buf();
        fs::create_dir_all(&dir)?;
        Ok(Self {
            path: dir.join(format!("{}.json", sanitize(store))),
        })
    }

    fn read(&self) -> std::io::Result<SeenSet> {
        match fs::read_to_string(&self.path) {
            Ok(json) => serde_json::from_str(&json)
                .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e)),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(SeenSet::default()),
            Err(e) => Err(e),
        }
    }

    /// Write-to-temp + atomic rename so a concurrent reader never sees a
    /// half-written seen-set (matches [`CheckpointStore`](crate::CheckpointStore)).
    fn write(&self, set: &SeenSet) -> std::io::Result<()> {
        let json = serde_json::to_string_pretty(set)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        let parent = self.path.parent().unwrap_or_else(|| Path::new("."));
        let tmp = parent.join(format!(".dedup-{}.tmp", uuid::Uuid::new_v4().simple()));
        fs::write(&tmp, json)?;
        fs::rename(&tmp, &self.path)?;
        Ok(())
    }

    /// Acquire an exclusive advisory lock for this store, held until the
    /// returned handle drops. Serializes the read-modify-write in
    /// [`filter_unseen`](Self::filter_unseen) so two concurrent runs of the same
    /// store can't lost-update each other's newly-recorded hashes (which would
    /// let items reprocess). Uses the same `std::fs::File` locking the registry
    /// supervisor relies on — no extra dependency. Blocking, so an overlapping
    /// run waits rather than racing.
    fn lock(&self) -> std::io::Result<fs::File> {
        let mut p = self.path.as_os_str().to_owned();
        p.push(".lock");
        let f = fs::OpenOptions::new()
            .read(true)
            .write(true)
            .create(true)
            .truncate(false)
            .open(PathBuf::from(p))?;
        f.lock()?;
        Ok(f)
    }

    /// Return the items not seen in any prior call (in input order) and commit
    /// their hashes so a later run skips them. Items duplicated *within* this
    /// call collapse to their first occurrence. The store is rewritten only when
    /// at least one new item is found, so a poll that surfaces nothing new does
    /// no write.
    ///
    /// The full read-modify-write runs under an exclusive lock on a sibling
    /// `<store>.json.lock`, so overlapping runs of the *same* store are
    /// serialized rather than racing (the headline "exactly once across runs"
    /// guarantee). Different stores never contend.
    ///
    /// `hash_fields`: when non-empty and an item is a JSON object, only these
    /// top-level fields contribute to the hash (a stable identity key, e.g.
    /// `["id"]`), so churn in other fields doesn't resurface a handled item.
    /// Empty hashes the whole item canonically. `now` is the timestamp recorded
    /// for newly-seen hashes (passed in so this stays pure/testable).
    ///
    /// `ttl`: when set, entries first seen more than `ttl` before `now` are
    /// evicted *before* the membership check, so (a) the file stays bounded to
    /// the active window and (b) an item that aged out is treated as new again.
    /// `None` keeps every hash forever (unbounded — fine for low-volume or
    /// short-lived stores). Eviction happens under the same lock as the insert,
    /// so it never races a concurrent run.
    pub fn filter_unseen(
        &self,
        items: &[Value],
        hash_fields: &[String],
        now: chrono::DateTime<chrono::Utc>,
        ttl: Option<chrono::Duration>,
    ) -> std::io::Result<Vec<Value>> {
        let _lock = self.lock()?;
        let mut set = self.read()?;
        let evicted = match ttl {
            Some(ttl) => prune_in_place(&mut set, now - ttl),
            None => 0,
        };
        let now_str = now.to_rfc3339();
        let mut unseen = Vec::new();
        let mut batch: HashSet<String> = HashSet::new();
        for item in items {
            let h = content_hash(item, hash_fields);
            if set.hashes.contains_key(&h) || !batch.insert(h.clone()) {
                continue;
            }
            set.hashes.insert(h, now_str.clone());
            unseen.push(item.clone());
        }
        // Persist when anything changed — new hashes recorded, or stale ones
        // evicted (so the bound is actually applied to disk).
        if !unseen.is_empty() || evicted > 0 {
            self.write(&set)?;
        }
        Ok(unseen)
    }

    /// Evict every entry first seen strictly before `cutoff`, returning the
    /// number removed. A standalone GC entry point — a maintenance pass can call
    /// this directly (e.g. `prune_older_than(Utc::now() - Duration::days(90))`)
    /// without running a poll. [`filter_unseen`](Self::filter_unseen) does the
    /// same eviction inline when given a `ttl`.
    pub fn prune_older_than(
        &self,
        cutoff: chrono::DateTime<chrono::Utc>,
    ) -> std::io::Result<usize> {
        let _lock = self.lock()?;
        let mut set = self.read()?;
        let removed = prune_in_place(&mut set, cutoff);
        if removed > 0 {
            self.write(&set)?;
        }
        Ok(removed)
    }
}

/// Stable SHA-256 (hex) of an item's canonical JSON. With `hash_fields`, only
/// those top-level object fields are hashed (a missing field contributes
/// `null`), giving a content hash keyed to item *identity* rather than its full
/// contents.
pub fn content_hash(item: &Value, hash_fields: &[String]) -> String {
    let basis = if hash_fields.is_empty() {
        item.clone()
    } else {
        let mut obj = serde_json::Map::new();
        for f in hash_fields {
            obj.insert(f.clone(), item.get(f).cloned().unwrap_or(Value::Null));
        }
        Value::Object(obj)
    };
    let mut hasher = Sha256::new();
    hasher.update(canonicalize(&basis).as_bytes());
    to_hex(&hasher.finalize())
}

/// Deterministic string form of a JSON value with object keys sorted, so two
/// objects that differ only in key order hash identically. (`serde_json`'s
/// default serialization preserves insertion order, which is not stable for
/// hashing.)
fn canonicalize(v: &Value) -> String {
    match v {
        Value::Object(map) => {
            let mut entries: Vec<(&String, &Value)> = map.iter().collect();
            entries.sort_by(|a, b| a.0.cmp(b.0));
            let inner: Vec<String> = entries
                .iter()
                .map(|(k, val)| {
                    format!(
                        "{}:{}",
                        serde_json::to_string(k).unwrap_or_default(),
                        canonicalize(val)
                    )
                })
                .collect();
            format!("{{{}}}", inner.join(","))
        }
        Value::Array(arr) => {
            let inner: Vec<String> = arr.iter().map(canonicalize).collect();
            format!("[{}]", inner.join(","))
        }
        other => other.to_string(),
    }
}

fn to_hex(bytes: &[u8]) -> String {
    let mut s = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        let _ = write!(s, "{b:02x}");
    }
    s
}

/// Reduce a store namespace to `[A-Za-z0-9_-]`, collapsing everything else to
/// `_`, so it can never traverse out of the store directory. Empty → `default`.
/// The collapse is lossy: two distinct names that differ only in disallowed
/// characters (e.g. `a.b` and `a_b`) map to the same file and share a seen-set.
/// Store names are operator-chosen, so keep them within `[A-Za-z0-9_-]`.
fn sanitize(store: &str) -> String {
    let cleaned: String = store
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect();
    if cleaned.is_empty() {
        "default".to_string()
    } else {
        cleaned
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::{Duration, TimeZone, Utc};
    use serde_json::json;

    fn tmp_dir(tag: &str) -> PathBuf {
        std::env::temp_dir().join(format!(
            "car-dedup-{tag}-{}",
            &uuid::Uuid::new_v4().simple().to_string()[..8]
        ))
    }

    /// A fixed epoch timestamp, for tests that don't care about the exact value.
    fn at(secs: i64) -> chrono::DateTime<Utc> {
        Utc.timestamp_opt(secs, 0).unwrap()
    }

    #[test]
    fn first_run_passes_all_then_skips_seen() {
        let dir = tmp_dir("basic");
        let store = DedupStore::open(&dir, "src").unwrap();
        let items = vec![json!({"id": 1}), json!({"id": 2})];

        let first = store.filter_unseen(&items, &[], at(0), None).unwrap();
        assert_eq!(first.len(), 2, "all new on first run");

        // Re-open (simulates a later scheduled run) — both already seen.
        let store2 = DedupStore::open(&dir, "src").unwrap();
        let second = store2.filter_unseen(&items, &[], at(1), None).unwrap();
        assert!(second.is_empty(), "nothing new on second run");

        // A third item slips through; the prior two stay suppressed.
        let mut more = items.clone();
        more.push(json!({"id": 3}));
        let third = store2.filter_unseen(&more, &[], at(2), None).unwrap();
        assert_eq!(third, vec![json!({"id": 3})]);

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn within_batch_duplicates_collapse() {
        let dir = tmp_dir("batch");
        let store = DedupStore::open(&dir, "s").unwrap();
        let items = vec![json!({"id": 1}), json!({"id": 1}), json!({"id": 2})];
        let unseen = store.filter_unseen(&items, &[], at(0), None).unwrap();
        assert_eq!(unseen, vec![json!({"id": 1}), json!({"id": 2})]);
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn hash_fields_key_on_identity_not_full_content() {
        let dir = tmp_dir("fields");
        let store = DedupStore::open(&dir, "s").unwrap();
        let v1 = vec![json!({"id": 7, "title": "old"})];
        assert_eq!(store.filter_unseen(&v1, &["id".into()], at(0), None).unwrap().len(), 1);
        // Same id, changed title → still considered seen.
        let v2 = vec![json!({"id": 7, "title": "new"})];
        assert!(store.filter_unseen(&v2, &["id".into()], at(1), None).unwrap().is_empty());
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn key_order_does_not_affect_hash() {
        assert_eq!(
            content_hash(&json!({"a": 1, "b": 2}), &[]),
            content_hash(&json!({"b": 2, "a": 1}), &[]),
        );
    }

    #[test]
    fn no_write_when_nothing_new() {
        let dir = tmp_dir("nowrite");
        let store = DedupStore::open(&dir, "s").unwrap();
        // Empty input → no file created.
        assert!(store.filter_unseen(&[], &[], at(0), None).unwrap().is_empty());
        assert!(!dir.join("s.json").exists(), "no write on empty input");
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn namespace_is_sanitized() {
        let dir = tmp_dir("sanitize");
        let store = DedupStore::open(&dir, "../escape/../x").unwrap();
        store.filter_unseen(&[json!(1)], &[], at(0), None).unwrap();
        // The file lands inside `dir`, not traversed out of it.
        let entries: Vec<_> = fs::read_dir(&dir).unwrap().filter_map(|e| e.ok()).collect();
        assert!(entries.iter().all(|e| e.path().starts_with(&dir)));
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn ttl_evicts_expired_and_allows_reprocess() {
        let dir = tmp_dir("ttl");
        let store = DedupStore::open(&dir, "s").unwrap();
        let ttl = Duration::seconds(3600);
        let items = vec![json!({"id": 1})];

        // First seen at t0.
        assert_eq!(store.filter_unseen(&items, &[], at(1_000_000), Some(ttl)).unwrap().len(), 1);
        // Within the TTL window → still suppressed.
        assert!(store
            .filter_unseen(&items, &[], at(1_001_800), Some(ttl))
            .unwrap()
            .is_empty());
        // Past the TTL → evicted and reprocessed as new.
        assert_eq!(
            store.filter_unseen(&items, &[], at(1_007_200), Some(ttl)).unwrap().len(),
            1
        );
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn prune_older_than_removes_only_expired() {
        let dir = tmp_dir("prune");
        let store = DedupStore::open(&dir, "s").unwrap();
        store.filter_unseen(&[json!({"id": 1})], &[], at(1_000_000), None).unwrap();
        store.filter_unseen(&[json!({"id": 2})], &[], at(2_000_000), None).unwrap();

        // Cutoff between the two first-seen times: drops the old one only.
        assert_eq!(store.prune_older_than(at(1_500_000)).unwrap(), 1);
        // id=2 stays suppressed; id=1 aged out and is reprocessable.
        assert!(store
            .filter_unseen(&[json!({"id": 2})], &[], at(2_000_001), None)
            .unwrap()
            .is_empty());
        assert_eq!(
            store.filter_unseen(&[json!({"id": 1})], &[], at(2_000_002), None).unwrap().len(),
            1
        );
        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn prune_keeps_unparseable_timestamps() {
        // Fail-safe: a legacy/corrupt timestamp is never silently dropped.
        let mut set = SeenSet::default();
        set.hashes.insert("good".into(), at(0).to_rfc3339());
        set.hashes.insert("weird".into(), "not-a-timestamp".into());
        let removed = prune_in_place(&mut set, at(1_000_000));
        assert_eq!(removed, 1);
        assert!(set.hashes.contains_key("weird"));
        assert!(!set.hashes.contains_key("good"));
    }
}