verbs/
watch_plan.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Pure `heddle watch` planning (no FS / notify / oplog I/O).
3//!
4//! Owns filter validation, relative `--since` duration parsing, notify-class
5//! relevance, and filter matching against event kind strings. Repo open,
6//! oplog drain, notify watcher setup, and RecoveryAdvice mapping stay CLI-owned.
7
8use chrono::{DateTime, Duration as ChronoDuration, Utc};
9
10/// Default debounce interval for the notify tail loop (milliseconds).
11pub const DEFAULT_POLL_INTERVAL_MS: u64 = 200;
12
13/// Hard cap on the in-process recent-entries window.
14pub const MAX_TAIL_WINDOW: usize = 100_000;
15
16/// Invalid `--filter` values.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub enum WatchFilterPlanError {
19    /// A comma-separated token was empty after trim (e.g. `snapshot,,merge`).
20    EmptyToken,
21    /// Token is not in the caller-supplied valid kinds list.
22    UnknownKind { kind: String, valid: Vec<String> },
23}
24
25impl WatchFilterPlanError {
26    pub fn kind(&self) -> &'static str {
27        match self {
28            Self::EmptyToken => "watch_filter_empty_token",
29            Self::UnknownKind { .. } => "watch_filter_invalid",
30        }
31    }
32}
33
34/// Parse `--filter snapshot,merge` into a set of kind strings.
35///
36/// - `None` / blank / only empty tokens after trim → `Ok(None)` (no filter).
37/// - Empty token between commas (e.g. `a,,b`) → `Err(EmptyToken)`.
38/// - Unknown kind → `Err(UnknownKind { .. })` with the valid list for messaging.
39pub fn plan_watch_filter(
40    spec: Option<&str>,
41    valid_kinds: &[&str],
42) -> Result<Option<Vec<String>>, WatchFilterPlanError> {
43    let Some(raw) = spec else {
44        return Ok(None);
45    };
46    let trimmed = raw.trim();
47    if trimmed.is_empty() {
48        return Ok(None);
49    }
50
51    let mut kinds = Vec::new();
52    for token in trimmed.split(',') {
53        let kind = token.trim();
54        if kind.is_empty() {
55            return Err(WatchFilterPlanError::EmptyToken);
56        }
57        if !valid_kinds.contains(&kind) {
58            return Err(WatchFilterPlanError::UnknownKind {
59                kind: kind.to_string(),
60                valid: valid_kinds.iter().map(|s| (*s).to_string()).collect(),
61            });
62        }
63        kinds.push(kind.to_string());
64    }
65    if kinds.is_empty() {
66        return Ok(None);
67    }
68    Ok(Some(kinds))
69}
70
71/// Invalid `--since` duration specs.
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub enum WatchSincePlanError {
74    /// Empty or whitespace-only string.
75    Empty,
76    /// Leading number portion failed to parse.
77    InvalidNumber { spec: String },
78    /// Unit is not `s` / `m` / `h` / `d` (or empty = seconds).
79    UnknownUnit { unit: String },
80}
81
82impl WatchSincePlanError {
83    pub fn kind(&self) -> &'static str {
84        match self {
85            Self::Empty => "watch_since_empty",
86            Self::InvalidNumber { .. } => "watch_since_invalid_number",
87            Self::UnknownUnit { .. } => "watch_since_unknown_unit",
88        }
89    }
90}
91
92/// Parse a relative duration like `30s` / `5m` / `1h` / `2d` into a second count.
93///
94/// Empty unit means seconds. Does not touch wall-clock; pair with
95/// [`plan_watch_since_cutoff`] for a UTC instant.
96pub fn parse_since_duration_secs(spec: &str) -> Result<i64, WatchSincePlanError> {
97    let trimmed = spec.trim();
98    if trimmed.is_empty() {
99        return Err(WatchSincePlanError::Empty);
100    }
101    let (num_part, unit) = trimmed.split_at(
102        trimmed
103            .find(|c: char| !c.is_ascii_digit())
104            .unwrap_or(trimmed.len()),
105    );
106    if num_part.is_empty() {
107        return Err(WatchSincePlanError::InvalidNumber {
108            spec: trimmed.to_string(),
109        });
110    }
111    let n: i64 = num_part
112        .parse()
113        .map_err(|_| WatchSincePlanError::InvalidNumber {
114            spec: trimmed.to_string(),
115        })?;
116    let secs = match unit {
117        "s" | "" => n,
118        "m" => n.saturating_mul(60),
119        "h" => n.saturating_mul(60 * 60),
120        "d" => n.saturating_mul(60 * 60 * 24),
121        other => {
122            return Err(WatchSincePlanError::UnknownUnit {
123                unit: other.to_string(),
124            });
125        }
126    };
127    Ok(secs)
128}
129
130/// Parse `--since` relative to a provided `now` (pure — no ambient clock).
131pub fn plan_watch_since_cutoff(
132    spec: &str,
133    now: DateTime<Utc>,
134) -> Result<DateTime<Utc>, WatchSincePlanError> {
135    let secs = parse_since_duration_secs(spec)?;
136    Ok(now - ChronoDuration::seconds(secs))
137}
138
139/// Notify-side event class without depending on the `notify` crate.
140#[derive(Debug, Clone, Copy, PartialEq, Eq)]
141pub enum WatchNotifyClass {
142    Modify,
143    Create,
144    Remove,
145    Other,
146}
147
148/// Whether a notify event class should trigger an oplog re-read.
149///
150/// Atomic `write_file_atomic` produces Create (temp) and Modify/Remove (rename).
151pub fn is_relevant_watch_event(class: WatchNotifyClass) -> bool {
152    matches!(
153        class,
154        WatchNotifyClass::Modify | WatchNotifyClass::Create | WatchNotifyClass::Remove
155    )
156}
157
158/// UX alias: `--filter merge` matches the wire verb `thread_update`.
159pub fn watch_kind_matches_filter(filter_kind: &str, event_kind: &str) -> bool {
160    filter_kind == event_kind || (filter_kind == "merge" && event_kind == "thread_update")
161}
162
163/// Whether an event kind passes an optional filter list.
164pub fn watch_passes_filter(filter: Option<&[String]>, event_kind: &str) -> bool {
165    match filter {
166        None => true,
167        Some(allowed) => allowed
168            .iter()
169            .any(|k| watch_kind_matches_filter(k.as_str(), event_kind)),
170    }
171}
172
173#[cfg(test)]
174mod tests {
175    use super::*;
176
177    fn valid() -> Vec<&'static str> {
178        vec![
179            "snapshot",
180            "merge",
181            "thread_create",
182            "thread_update",
183            "remote_thread_update",
184            "purge",
185        ]
186    }
187
188    #[test]
189    fn plan_watch_filter_validates_kinds() {
190        assert!(plan_watch_filter(None, &valid()).unwrap().is_none());
191        assert!(plan_watch_filter(Some(""), &valid()).unwrap().is_none());
192        assert!(plan_watch_filter(Some("  "), &valid()).unwrap().is_none());
193        let parsed = plan_watch_filter(Some("snapshot,merge"), &valid())
194            .unwrap()
195            .unwrap();
196        assert_eq!(parsed, vec!["snapshot", "merge"]);
197        assert!(matches!(
198            plan_watch_filter(Some("not_a_real_kind"), &valid()),
199            Err(WatchFilterPlanError::UnknownKind { kind, .. }) if kind == "not_a_real_kind"
200        ));
201        assert_eq!(
202            plan_watch_filter(Some("snapshot,,merge"), &valid()),
203            Err(WatchFilterPlanError::EmptyToken)
204        );
205    }
206
207    #[test]
208    fn filter_accepts_catalog_kinds_when_listed() {
209        for kind in ["remote_thread_update", "purge", "thread_create"] {
210            assert!(
211                plan_watch_filter(Some(kind), &valid()).is_ok(),
212                "filter kind {kind:?} must be accepted when listed as valid"
213            );
214        }
215    }
216
217    #[test]
218    fn parse_since_accepts_common_units() {
219        assert_eq!(parse_since_duration_secs("30s").unwrap(), 30);
220        assert_eq!(parse_since_duration_secs("5m").unwrap(), 5 * 60);
221        assert_eq!(parse_since_duration_secs("2h").unwrap(), 2 * 60 * 60);
222        assert_eq!(parse_since_duration_secs("1d").unwrap(), 86_400);
223        assert_eq!(parse_since_duration_secs("45").unwrap(), 45);
224    }
225
226    #[test]
227    fn parse_since_rejects_bad_input() {
228        assert_eq!(
229            parse_since_duration_secs(""),
230            Err(WatchSincePlanError::Empty)
231        );
232        assert_eq!(
233            parse_since_duration_secs("   "),
234            Err(WatchSincePlanError::Empty)
235        );
236        assert!(matches!(
237            parse_since_duration_secs("5x"),
238            Err(WatchSincePlanError::UnknownUnit { unit }) if unit == "x"
239        ));
240        assert!(matches!(
241            parse_since_duration_secs("m"),
242            Err(WatchSincePlanError::InvalidNumber { .. })
243        ));
244    }
245
246    #[test]
247    fn plan_watch_since_cutoff_subtracts_from_now() {
248        let now = DateTime::parse_from_rfc3339("2026-05-02T12:00:00Z")
249            .unwrap()
250            .with_timezone(&Utc);
251        let cutoff = plan_watch_since_cutoff("5m", now).unwrap();
252        assert_eq!((now - cutoff).num_seconds(), 300);
253    }
254
255    #[test]
256    fn relevant_events_and_filter_matching() {
257        assert!(is_relevant_watch_event(WatchNotifyClass::Modify));
258        assert!(is_relevant_watch_event(WatchNotifyClass::Create));
259        assert!(is_relevant_watch_event(WatchNotifyClass::Remove));
260        assert!(!is_relevant_watch_event(WatchNotifyClass::Other));
261
262        assert!(watch_kind_matches_filter("snapshot", "snapshot"));
263        assert!(watch_kind_matches_filter("merge", "thread_update"));
264        assert!(!watch_kind_matches_filter("snapshot", "thread_create"));
265        assert!(watch_passes_filter(None, "anything"));
266        let filter = vec!["snapshot".into()];
267        assert!(watch_passes_filter(Some(&filter), "snapshot"));
268        assert!(!watch_passes_filter(Some(&filter), "thread_create"));
269    }
270
271    #[test]
272    fn constants_match_historical_cli() {
273        assert_eq!(DEFAULT_POLL_INTERVAL_MS, 200);
274        assert_eq!(MAX_TAIL_WINDOW, 100_000);
275    }
276}
verbs/watch_plan.rs

verbs/
watch_plan.rs
