oximedia_workflow/
pause_resume.rs

1//! Workflow pause/resume support with in-memory checkpoint serialisation.
2//!
3//! [`PauseCheckpoint`] captures the state of a workflow at the moment it is
4//! paused (step index, completed steps, pending steps, and arbitrary key-value
5//! state data). [`PauseCheckpointStore`] manages a collection of such
6//! checkpoints keyed by workflow ID.
7//!
8//! Serialisation is implemented by hand (no `serde` derive) so there are no
9//! additional dependencies.
10//!
11//! # Example
12//!
13//! ```rust
14//! use oximedia_workflow::pause_resume::{PauseCheckpoint, PauseCheckpointStore};
15//!
16//! let cp = PauseCheckpoint::new("wf-001", 2)
17//!     .with_completed("ingest")
18//!     .with_completed("analyse")
19//!     .with_pending("transcode")
20//!     .with_pending("upload")
21//!     .with_state("source_path", "/mnt/raw/clip.mxf")
22//!     .with_reason("Awaiting approval");
23//!
24//! let json = cp.to_json();
25//! let restored = PauseCheckpoint::from_json(&json).expect("round-trip");
26//! assert_eq!(restored.workflow_id, "wf-001");
27//! assert_eq!(restored.remaining_steps(), 2);
28//! ```
29
30#![allow(dead_code)]
31
32use std::collections::HashMap;
33use std::time::{SystemTime, UNIX_EPOCH};
34
35// ---------------------------------------------------------------------------
36// PauseCheckpoint
37// ---------------------------------------------------------------------------
38
39/// A snapshot of workflow state recorded when the workflow is paused.
40#[derive(Debug, Clone, PartialEq)]
41pub struct PauseCheckpoint {
42    /// Identifier of the paused workflow.
43    pub workflow_id: String,
44    /// Zero-based index of the step at which execution was paused.
45    pub step_index: usize,
46    /// Names of steps that had already completed before pausing.
47    pub completed_steps: Vec<String>,
48    /// Names of steps still remaining to be executed.
49    pub pending_steps: Vec<String>,
50    /// Arbitrary serialised outputs or metadata from completed steps.
51    pub state_data: HashMap<String, String>,
52    /// Unix timestamp (milliseconds) when the checkpoint was recorded.
53    pub paused_at: u64,
54    /// Human-readable reason the workflow was paused.
55    pub pause_reason: String,
56}
57
58impl PauseCheckpoint {
59    /// Create a new checkpoint for `workflow_id` paused at `step_index`.
60    ///
61    /// `paused_at` is set to the current wall-clock time in milliseconds.
62    #[must_use]
63    pub fn new(workflow_id: impl Into<String>, step_index: usize) -> Self {
64        let paused_at = SystemTime::now()
65            .duration_since(UNIX_EPOCH)
66            .map(|d| d.as_millis() as u64)
67            .unwrap_or(0);
68        Self {
69            workflow_id: workflow_id.into(),
70            step_index,
71            completed_steps: Vec::new(),
72            pending_steps: Vec::new(),
73            state_data: HashMap::new(),
74            paused_at,
75            pause_reason: String::new(),
76        }
77    }
78
79    /// Add a completed step name (builder pattern).
80    #[must_use]
81    pub fn with_completed(mut self, step: impl Into<String>) -> Self {
82        self.completed_steps.push(step.into());
83        self
84    }
85
86    /// Add a pending step name (builder pattern).
87    #[must_use]
88    pub fn with_pending(mut self, step: impl Into<String>) -> Self {
89        self.pending_steps.push(step.into());
90        self
91    }
92
93    /// Insert a key-value pair into the state data (builder pattern).
94    #[must_use]
95    pub fn with_state(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
96        self.state_data.insert(key.into(), value.into());
97        self
98    }
99
100    /// Set the human-readable pause reason (builder pattern).
101    #[must_use]
102    pub fn with_reason(mut self, reason: impl Into<String>) -> Self {
103        self.pause_reason = reason.into();
104        self
105    }
106
107    /// Number of steps still pending (not yet executed).
108    #[must_use]
109    pub fn remaining_steps(&self) -> usize {
110        self.pending_steps.len()
111    }
112
113    /// Progress as a fraction in `[0.0, 1.0]`.
114    ///
115    /// Returns `0.0` when no steps have been completed and no steps are pending.
116    #[must_use]
117    pub fn progress_percent(&self) -> f32 {
118        let done = self.completed_steps.len();
119        let total = done + self.pending_steps.len();
120        if total == 0 {
121            return 0.0;
122        }
123        done as f32 / total as f32
124    }
125
126    // -----------------------------------------------------------------------
127    // Hand-written JSON serialisation
128    // -----------------------------------------------------------------------
129
130    /// Serialise the checkpoint to a JSON string.
131    ///
132    /// The format is stable and suitable for long-term storage.
133    #[must_use]
134    pub fn to_json(&self) -> String {
135        let completed = json_string_array(&self.completed_steps);
136        let pending = json_string_array(&self.pending_steps);
137        let state = json_string_map(&self.state_data);
138
139        format!(
140            concat!(
141                "{{",
142                "\"workflow_id\":{wid},",
143                "\"step_index\":{si},",
144                "\"completed_steps\":{completed},",
145                "\"pending_steps\":{pending},",
146                "\"state_data\":{state},",
147                "\"paused_at\":{pa},",
148                "\"pause_reason\":{reason}",
149                "}}"
150            ),
151            wid = json_escape_string(&self.workflow_id),
152            si = self.step_index,
153            completed = completed,
154            pending = pending,
155            state = state,
156            pa = self.paused_at,
157            reason = json_escape_string(&self.pause_reason),
158        )
159    }
160
161    /// Deserialise a checkpoint from a JSON string produced by [`to_json`].
162    ///
163    /// # Errors
164    ///
165    /// Returns `Err(String)` when the JSON is malformed or a required field
166    /// is missing.
167    ///
168    /// [`to_json`]: PauseCheckpoint::to_json
169    pub fn from_json(json: &str) -> Result<Self, String> {
170        let trimmed = json.trim();
171        if !trimmed.starts_with('{') || !trimmed.ends_with('}') {
172            return Err("Not a JSON object".to_string());
173        }
174
175        let workflow_id =
176            parse_json_string(trimmed, "workflow_id").ok_or("Missing field: workflow_id")?;
177        let step_index =
178            parse_json_u64(trimmed, "step_index").ok_or("Missing field: step_index")? as usize;
179        let completed_steps =
180            parse_json_string_array(trimmed, "completed_steps").unwrap_or_default();
181        let pending_steps = parse_json_string_array(trimmed, "pending_steps").unwrap_or_default();
182        let state_data = parse_json_string_map(trimmed, "state_data").unwrap_or_default();
183        let paused_at = parse_json_u64(trimmed, "paused_at").unwrap_or(0);
184        let pause_reason = parse_json_string(trimmed, "pause_reason").unwrap_or_default();
185
186        Ok(Self {
187            workflow_id,
188            step_index,
189            completed_steps,
190            pending_steps,
191            state_data,
192            paused_at,
193            pause_reason,
194        })
195    }
196}
197
198// ---------------------------------------------------------------------------
199// PauseCheckpointStore
200// ---------------------------------------------------------------------------
201
202/// In-memory store for [`PauseCheckpoint`] instances.
203///
204/// Checkpoints are keyed by `workflow_id`.  Saving a checkpoint with an
205/// existing key overwrites the previous one.
206#[derive(Debug, Default)]
207pub struct PauseCheckpointStore {
208    checkpoints: HashMap<String, PauseCheckpoint>,
209}
210
211impl PauseCheckpointStore {
212    /// Create an empty store.
213    #[must_use]
214    pub fn new() -> Self {
215        Self::default()
216    }
217
218    /// Save (insert or overwrite) a checkpoint.
219    pub fn save(&mut self, checkpoint: PauseCheckpoint) {
220        self.checkpoints
221            .insert(checkpoint.workflow_id.clone(), checkpoint);
222    }
223
224    /// Load a checkpoint by workflow ID.
225    #[must_use]
226    pub fn load(&self, workflow_id: &str) -> Option<&PauseCheckpoint> {
227        self.checkpoints.get(workflow_id)
228    }
229
230    /// Delete a checkpoint.  Returns `true` if the checkpoint existed.
231    pub fn delete(&mut self, workflow_id: &str) -> bool {
232        self.checkpoints.remove(workflow_id).is_some()
233    }
234
235    /// Return references to all stored checkpoints.
236    #[must_use]
237    pub fn list_paused(&self) -> Vec<&PauseCheckpoint> {
238        self.checkpoints.values().collect()
239    }
240
241    /// Number of stored checkpoints.
242    #[must_use]
243    pub fn count(&self) -> usize {
244        self.checkpoints.len()
245    }
246}
247
248// ---------------------------------------------------------------------------
249// PauseResumeController
250// ---------------------------------------------------------------------------
251
252/// High-level controller for pausing and resuming workflows by numeric ID.
253///
254/// Internally maps numeric workflow IDs to [`PauseCheckpoint`]s stored in a
255/// [`PauseCheckpointStore`].  Numeric IDs are converted to strings for
256/// compatibility with the checkpoint store's string-keyed API.
257///
258/// # Example
259///
260/// ```rust
261/// use oximedia_workflow::pause_resume::PauseResumeController;
262///
263/// let mut ctrl = PauseResumeController::new();
264/// ctrl.pause(42);
265/// assert!(ctrl.is_paused(42));
266/// ctrl.resume(42);
267/// assert!(!ctrl.is_paused(42));
268/// ```
269#[derive(Debug, Default)]
270pub struct PauseResumeController {
271    store: PauseCheckpointStore,
272}
273
274impl PauseResumeController {
275    /// Create an empty controller.
276    #[must_use]
277    pub fn new() -> Self {
278        Self::default()
279    }
280
281    /// Pause a workflow by its numeric ID.
282    ///
283    /// If the workflow is already paused this is a no-op (the checkpoint is
284    /// updated with a fresh timestamp by replacing the entry).
285    pub fn pause(&mut self, workflow_id: u64) {
286        let key = workflow_id.to_string();
287        let cp = PauseCheckpoint::new(key, 0).with_reason("Paused by PauseResumeController");
288        self.store.save(cp);
289    }
290
291    /// Resume a workflow by removing its pause checkpoint.
292    ///
293    /// If the workflow is not currently paused this is a no-op.
294    pub fn resume(&mut self, workflow_id: u64) {
295        self.store.delete(&workflow_id.to_string());
296    }
297
298    /// Return `true` if a pause checkpoint exists for the given workflow ID.
299    #[must_use]
300    pub fn is_paused(&self, workflow_id: u64) -> bool {
301        self.store.load(&workflow_id.to_string()).is_some()
302    }
303
304    /// Return the number of currently paused workflows.
305    #[must_use]
306    pub fn paused_count(&self) -> usize {
307        self.store.count()
308    }
309
310    /// List all currently paused workflow IDs.
311    #[must_use]
312    pub fn paused_ids(&self) -> Vec<u64> {
313        self.store
314            .list_paused()
315            .iter()
316            .filter_map(|cp| cp.workflow_id.parse::<u64>().ok())
317            .collect()
318    }
319}
320
321// ---------------------------------------------------------------------------
322// Hand-written JSON helpers (no serde dependency)
323// ---------------------------------------------------------------------------
324
325/// Produce a JSON-escaped quoted string: `"value"`.
326fn json_escape_string(s: &str) -> String {
327    let mut out = String::with_capacity(s.len() + 2);
328    out.push('"');
329    for c in s.chars() {
330        match c {
331            '"' => out.push_str("\\\""),
332            '\\' => out.push_str("\\\\"),
333            '\n' => out.push_str("\\n"),
334            '\r' => out.push_str("\\r"),
335            '\t' => out.push_str("\\t"),
336            c => out.push(c),
337        }
338    }
339    out.push('"');
340    out
341}
342
343/// Produce a JSON array of strings: `["a","b"]`.
344fn json_string_array(items: &[String]) -> String {
345    let mut out = String::from("[");
346    for (i, item) in items.iter().enumerate() {
347        if i > 0 {
348            out.push(',');
349        }
350        out.push_str(&json_escape_string(item));
351    }
352    out.push(']');
353    out
354}
355
356/// Produce a JSON object from a string map: `{"k":"v"}`.
357fn json_string_map(map: &HashMap<String, String>) -> String {
358    let mut out = String::from("{");
359    let mut first = true;
360    for (k, v) in map {
361        if !first {
362            out.push(',');
363        }
364        first = false;
365        out.push_str(&json_escape_string(k));
366        out.push(':');
367        out.push_str(&json_escape_string(v));
368    }
369    out.push('}');
370    out
371}
372
373// ---------------------------------------------------------------------------
374// Minimal JSON parsing helpers
375// These are intentionally simple — they support only the subset of JSON that
376// `to_json` produces.
377// ---------------------------------------------------------------------------
378
379/// Find `"key":` in `json` and return the position just after the `:`.
380fn find_value_start(json: &str, key: &str) -> Option<usize> {
381    let search = format!("\"{}\":", key);
382    json.find(&search).map(|pos| pos + search.len())
383}
384
385/// Parse a quoted string value for `key` from `json`.
386fn parse_json_string(json: &str, key: &str) -> Option<String> {
387    let after_colon = find_value_start(json, key)?;
388    let rest = json[after_colon..].trim_start();
389    if !rest.starts_with('"') {
390        return None;
391    }
392    let inner = &rest[1..]; // skip opening quote
393    let mut result = String::new();
394    let mut chars = inner.chars().peekable();
395    loop {
396        match chars.next()? {
397            '"' => break,
398            '\\' => match chars.next()? {
399                '"' => result.push('"'),
400                '\\' => result.push('\\'),
401                'n' => result.push('\n'),
402                'r' => result.push('\r'),
403                't' => result.push('\t'),
404                c => result.push(c),
405            },
406            c => result.push(c),
407        }
408    }
409    Some(result)
410}
411
412/// Parse an unsigned integer value for `key` from `json`.
413fn parse_json_u64(json: &str, key: &str) -> Option<u64> {
414    let after_colon = find_value_start(json, key)?;
415    let rest = json[after_colon..].trim_start();
416    let end = rest
417        .find(|c: char| !c.is_ascii_digit())
418        .unwrap_or(rest.len());
419    rest[..end].parse().ok()
420}
421
422/// Parse an array of strings for `key` from `json`.
423///
424/// Supports the subset produced by [`json_string_array`].
425fn parse_json_string_array(json: &str, key: &str) -> Option<Vec<String>> {
426    let after_colon = find_value_start(json, key)?;
427    let rest = json[after_colon..].trim_start();
428    if !rest.starts_with('[') {
429        return None;
430    }
431    let bracket_end = find_matching_bracket(rest, '[', ']')?;
432    let inner = &rest[1..bracket_end]; // between [ and ]
433    let mut result = Vec::new();
434    let mut remaining = inner.trim();
435    while !remaining.is_empty() {
436        if !remaining.starts_with('"') {
437            break;
438        }
439        let (s, consumed) = parse_one_json_string(remaining)?;
440        result.push(s);
441        remaining = remaining[consumed..].trim();
442        if remaining.starts_with(',') {
443            remaining = remaining[1..].trim();
444        }
445    }
446    Some(result)
447}
448
449/// Parse a `{"k":"v", ...}` object for `key` from `json`.
450///
451/// Supports the subset produced by [`json_string_map`].
452fn parse_json_string_map(json: &str, key: &str) -> Option<HashMap<String, String>> {
453    let after_colon = find_value_start(json, key)?;
454    let rest = json[after_colon..].trim_start();
455    if !rest.starts_with('{') {
456        return None;
457    }
458    let brace_end = find_matching_bracket(rest, '{', '}')?;
459    let inner = &rest[1..brace_end]; // between { and }
460    let mut result = HashMap::new();
461    let mut remaining = inner.trim();
462    while !remaining.is_empty() {
463        if !remaining.starts_with('"') {
464            break;
465        }
466        let (k, k_consumed) = parse_one_json_string(remaining)?;
467        remaining = remaining[k_consumed..].trim();
468        if !remaining.starts_with(':') {
469            break;
470        }
471        remaining = remaining[1..].trim();
472        let (v, v_consumed) = parse_one_json_string(remaining)?;
473        result.insert(k, v);
474        remaining = remaining[v_consumed..].trim();
475        if remaining.starts_with(',') {
476            remaining = remaining[1..].trim();
477        }
478    }
479    Some(result)
480}
481
482/// Parse one `"..."` JSON string from the start of `s`.
483///
484/// Returns `(value, bytes_consumed)` including the surrounding quotes.
485fn parse_one_json_string(s: &str) -> Option<(String, usize)> {
486    if !s.starts_with('"') {
487        return None;
488    }
489    let mut result = String::new();
490    let mut chars = s[1..].char_indices();
491    let mut consumed = 1usize; // opening quote
492    loop {
493        let (i, c) = chars.next()?;
494        match c {
495            '"' => {
496                consumed += i + c.len_utf8();
497                break;
498            }
499            '\\' => {
500                let (_, next) = chars.next()?;
501                consumed += 1;
502                match next {
503                    '"' => result.push('"'),
504                    '\\' => result.push('\\'),
505                    'n' => result.push('\n'),
506                    'r' => result.push('\r'),
507                    't' => result.push('\t'),
508                    c => result.push(c),
509                }
510            }
511            c => result.push(c),
512        }
513    }
514    Some((result, consumed))
515}
516
517/// Find the index of the matching closing bracket/brace, handling nesting.
518fn find_matching_bracket(s: &str, open: char, close: char) -> Option<usize> {
519    let mut depth: usize = 0;
520    let mut in_string = false;
521    let mut prev_backslash = false;
522    for (i, c) in s.char_indices() {
523        if prev_backslash {
524            prev_backslash = false;
525            continue;
526        }
527        if in_string {
528            if c == '\\' {
529                prev_backslash = true;
530            } else if c == '"' {
531                in_string = false;
532            }
533            continue;
534        }
535        if c == '"' {
536            in_string = true;
537        } else if c == open {
538            depth += 1;
539        } else if c == close {
540            depth -= 1;
541            if depth == 0 {
542                return Some(i);
543            }
544        }
545    }
546    None
547}
548
549// ===========================================================================
550// Tests
551// ===========================================================================
552
553#[cfg(test)]
554mod tests {
555    use super::*;
556
557    #[test]
558    fn test_new_checkpoint_fields() {
559        let cp = PauseCheckpoint::new("wf-001", 3);
560        assert_eq!(cp.workflow_id, "wf-001");
561        assert_eq!(cp.step_index, 3);
562        assert!(cp.completed_steps.is_empty());
563        assert!(cp.pending_steps.is_empty());
564        assert!(cp.state_data.is_empty());
565        assert!(cp.paused_at > 0);
566        assert!(cp.pause_reason.is_empty());
567    }
568
569    #[test]
570    fn test_progress_percent_partial() {
571        let cp = PauseCheckpoint::new("wf-002", 2)
572            .with_completed("step-a")
573            .with_completed("step-b")
574            .with_pending("step-c")
575            .with_pending("step-d");
576        let progress = cp.progress_percent();
577        // 2 done / 4 total = 0.5
578        assert!(
579            (progress - 0.5).abs() < f32::EPSILON,
580            "expected 0.5, got {progress}"
581        );
582    }
583
584    #[test]
585    fn test_progress_percent_all_done() {
586        let cp = PauseCheckpoint::new("wf-003", 2)
587            .with_completed("a")
588            .with_completed("b");
589        assert!((cp.progress_percent() - 1.0).abs() < f32::EPSILON);
590    }
591
592    #[test]
593    fn test_progress_percent_none_done() {
594        let cp = PauseCheckpoint::new("wf-004", 0)
595            .with_pending("a")
596            .with_pending("b");
597        assert!((cp.progress_percent() - 0.0).abs() < f32::EPSILON);
598    }
599
600    #[test]
601    fn test_to_json_from_json_round_trip() {
602        let original = PauseCheckpoint::new("wf-roundtrip", 1)
603            .with_completed("ingest")
604            .with_pending("transcode")
605            .with_pending("upload")
606            .with_state("source", "/data/clip.mxf")
607            .with_reason("Waiting for budget approval");
608
609        let json = original.to_json();
610        let restored = PauseCheckpoint::from_json(&json).expect("from_json should succeed");
611
612        assert_eq!(restored.workflow_id, original.workflow_id);
613        assert_eq!(restored.step_index, original.step_index);
614        assert_eq!(restored.completed_steps, original.completed_steps);
615        assert_eq!(restored.pending_steps, original.pending_steps);
616        assert_eq!(restored.state_data, original.state_data);
617        assert_eq!(restored.paused_at, original.paused_at);
618        assert_eq!(restored.pause_reason, original.pause_reason);
619    }
620
621    #[test]
622    fn test_save_load_delete() {
623        let mut store = PauseCheckpointStore::new();
624        let cp = PauseCheckpoint::new("wf-store", 0);
625
626        store.save(cp.clone());
627        assert_eq!(store.count(), 1);
628
629        let loaded = store.load("wf-store").expect("should find it");
630        assert_eq!(loaded.workflow_id, "wf-store");
631
632        assert!(store.delete("wf-store"));
633        assert_eq!(store.count(), 0);
634        assert!(store.load("wf-store").is_none());
635    }
636
637    #[test]
638    fn test_list_paused_returns_all() {
639        let mut store = PauseCheckpointStore::new();
640        store.save(PauseCheckpoint::new("wf-alpha", 0));
641        store.save(PauseCheckpoint::new("wf-beta", 1));
642        store.save(PauseCheckpoint::new("wf-gamma", 2));
643
644        let list = store.list_paused();
645        assert_eq!(list.len(), 3);
646    }
647
648    #[test]
649    fn test_remaining_steps() {
650        let cp = PauseCheckpoint::new("wf-rem", 1)
651            .with_pending("step-x")
652            .with_pending("step-y")
653            .with_pending("step-z");
654        assert_eq!(cp.remaining_steps(), 3);
655    }
656
657    #[test]
658    fn test_with_state() {
659        let out = std::env::temp_dir()
660            .join("oximedia-workflow-pause-out.mp4")
661            .to_string_lossy()
662            .into_owned();
663        let cp = PauseCheckpoint::new("wf-state", 0)
664            .with_state("output_path", &out)
665            .with_state("fps", "24");
666        assert_eq!(
667            cp.state_data.get("output_path").map(String::as_str),
668            Some(out.as_str())
669        );
670        assert_eq!(cp.state_data.get("fps").map(String::as_str), Some("24"));
671    }
672
673    #[test]
674    fn test_from_json_malformed_returns_err() {
675        let result = PauseCheckpoint::from_json("not json at all");
676        assert!(result.is_err());
677    }
678
679    #[test]
680    fn test_delete_nonexistent_returns_false() {
681        let mut store = PauseCheckpointStore::new();
682        assert!(!store.delete("ghost-workflow"));
683    }
684}
oximedia_workflow/pause_resume.rs

oximedia_workflow/
pause_resume.rs
