common/

long_operation.rs

// Generated by Qleany v1.11.0 from long_operation.tera
//! This module provides a framework for managing long-running operations with the ability to track
//! status, progress, and enable cancellation. It includes the infrastructure for defining, executing,
//! and monitoring such operations. For undoable operations, it is recommended to use the Undo/Redo framework.
//!
//! # Components:
//!
//! - **OperationStatus**: Enum representing the state of an operation.
//! - **OperationProgress**: Struct holding details about the progress of an operation.
//! - **LongOperation**: Trait that must be implemented by any long-running operation.
//! - **LongOperationManager**: Manager that orchestrates the execution, tracking, and cleanup of multiple operations.
//!
//! # Usage:
//!
//! 1. Implement the `LongOperation` trait for your task.
//! 2. Use `LongOperationManager` to start, track, and manage your operations.
//! 3. Access methods like:
//!     - `start_operation` to start new operations.
//!     - `get_operation_status`, `get_operation_progress` to query operation details.
//!     - `cancel_operation` to cancel operations.
//!     - `cleanup_finished_operations` to remove completed or cancelled operations.
//!
//! # Example:
//!
//! ```rust,ignore
//! // Define your long-running operation
//! use std::sync::Arc;
//! use std::sync::atomic::{AtomicBool, Ordering};
//! use std::thread;
//! use std::time::Duration;
//! use common::long_operation::{LongOperation, LongOperationManager, OperationProgress};
//!
//! pub struct MyOperation {
//!     pub total_steps: usize,
//! }
//!
//! impl LongOperation for MyOperation {
//!     fn execute(
//!         &self,
//!         progress_callback: Box<dyn Fn(OperationProgress) + Send>,
//!         cancel_flag: Arc<AtomicBool>,
//!     ) -> Result<(), String> {
//!         for i in 0..self.total_steps {
//!             if cancel_flag.load(Ordering::Relaxed) {
//!                 return Err("Operation cancelled".to_string());
//!             }
//!             thread::sleep(Duration::from_millis(500));
//!             progress_callback(OperationProgress::new(
//!                 (i as f32 / self.total_steps as f32) * 100.0,
//!                 Some(format!("Step {}/{}", i + 1, self.total_steps)),
//!             ));
//!         }
//!         Ok(())
//!     }
//! }
//!
//! let manager = LongOperationManager::new();
//! let my_operation = MyOperation { total_steps: 5 };
//! let operation_id = manager.start_operation(my_operation);
//!
//! while let Some(status) = manager.get_operation_status(&operation_id) {
//!     println!("{:?}", status);
//!     thread::sleep(Duration::from_millis(100));
//! }
//! ```
//!
//! # Notes:
//!
//! - Thread-safety is ensured through the use of `Arc<Mutex<T>>` and `AtomicBool`.
//! - Operations run in their own threads, ensuring non-blocking execution.
//! - Proper cleanup of finished operations is encouraged using `cleanup_finished_operations`.
//!
//! # Definitions:
//!
//! ## `OperationStatus`
//! Represents the state of an operation. Possible states are:
//! - `Running`: Operation is ongoing.
//! - `Completed`: Operation finished successfully.
//! - `Cancelled`: Operation was cancelled by the user.
//! - `Failed(String)`: Operation failed with an error message.
//!
//! ## `OperationProgress`
//! Describes the progress of an operation, including:
//! - `percentage` (0.0 to 100.0): Indicates completion progress.
//! - `message`: Optional user-defined progress description.
//!
//! ## `LongOperation` Trait
//! Any custom long-running operation must implement this trait:
//! - `execute`: Defines the operation logic, accepting a progress callback and cancellation flag.
//!
//! ## `LongOperationManager`
//! Provides APIs to manage operations, including:
//! - `start_operation`: Starts a new operation and returns its unique ID.
//! - `get_operation_status`: Queries the current status of an operation.
//! - `get_operation_progress`: Retrieves the progress of an operation.
//! - `cancel_operation`: Cancels an operation.
//! - `cleanup_finished_operations`: Removes completed or cancelled operations to free resources.
//!
//! ## Example Operation: FileProcessingOperation
//! Represents a long-running operation to process files. Demonstrates typical usage of the framework.
//!
//! - **Fields**:
//!     - `file_path`: Path of the file to process.
//!     - `total_files`: Number of files to process.
//! - **Behavior**:
//!   Simulates file processing with periodic progress updates. Supports cancellation.
//!
//!

// Generated by Qleany. Edit at your own risk! Be careful when regenerating this file
// as changes will be lost.

use crate::event::{Event, EventHub, LongOperationEvent, Origin};
use anyhow::Result;
use std::collections::{HashMap, HashSet};
use std::sync::{
    Arc, Condvar, Mutex,
    atomic::{AtomicBool, Ordering},
};
use std::thread;
use std::time::Duration;

// Status of a long operation
#[derive(Debug, Clone, PartialEq)]
pub enum OperationStatus {
    Running,
    Completed,
    Cancelled,
    Failed(String),
}

// Progress information
#[derive(Debug, Clone)]
pub struct OperationProgress {
    pub percentage: f32, // 0.0 to 100.0
    pub message: Option<String>,
}

impl OperationProgress {
    pub fn new(percentage: f32, message: Option<String>) -> Self {
        Self {
            percentage: percentage.clamp(0.0, 100.0),
            message,
        }
    }
}

// Trait that long operations must implement
pub trait LongOperation: Send + 'static {
    type Output: Send + Sync + 'static + serde::Serialize;

    fn execute(
        &self,
        progress_callback: Box<dyn Fn(OperationProgress) + Send>,
        cancel_flag: Arc<AtomicBool>,
    ) -> Result<Self::Output>;
}

// Trait for operation handles (type-erased)
trait OperationHandleTrait: Send {
    fn get_status(&self) -> OperationStatus;
    fn get_progress(&self) -> OperationProgress;
    fn cancel(&self);
    fn is_finished(&self) -> bool;
}

/// Lock a mutex, recovering from poisoning by returning the inner value — and
/// **clearing** the poison, so recovery is permanent rather than per-call.
///
/// If a thread panicked while holding the lock the data may be in an
/// inconsistent state, but for the transaction handles, event buffers and
/// status/progress trackers this guards, that is strictly preferable to
/// propagating the panic: the release profile unwinds, so a panic inside a long
/// operation is caught and reported as `Failed` — and every *later* use of the
/// same mutex would otherwise panic again, turning one reported failure into a
/// dead process.
///
/// `clear_poison` matches the `RwLock` counterparts in
/// `database::hashmap_store` (`read_or_recover` / `write_or_recover`).
/// `PoisonError::into_inner` alone leaves the flag set, so a plain
/// `.lock().unwrap()` elsewhere in the process would still trip on a lock this
/// function had already recovered.
pub fn lock_or_recover<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    match mutex.lock() {
        Ok(guard) => guard,
        Err(poisoned) => {
            let guard = poisoned.into_inner();
            mutex.clear_poison();
            guard
        }
    }
}

/// The completion signal every operation publishes to when it ends.
///
/// This is what lets a caller **block until** an operation finishes instead of
/// re-checking it on a timer. A worker marks its id finished here as the very
/// last thing it does — after the result is stored, the event emitted, and the
/// final status set — so a waiter woken by it is guaranteed to observe all
/// three. In particular, `is_finished(id) == true` implies
/// [`LongOperationManager::get_operation_result`] already has the result (for a
/// `Completed` operation), which is what makes an event-free blocking wait
/// correct.
///
/// Obtain one via [`LongOperationManager::completion_signal`] and block on it
/// **after** releasing any lock guarding the manager — see that method.
pub struct OperationCompletion {
    /// Ids that reached a final status. Deliberately never pruned: it mirrors
    /// the manager's `results` map, so a wait on an already-finished (even
    /// cleaned-up) operation returns at once instead of blocking on a signal
    /// that can no longer come.
    finished: Mutex<HashSet<String>>,
    condvar: Condvar,
}

impl OperationCompletion {
    fn new() -> Self {
        Self {
            finished: Mutex::new(HashSet::new()),
            condvar: Condvar::new(),
        }
    }

    /// Publish `id` as finished and wake every waiter.
    ///
    /// `notify_all`, not `notify_one`: waiters block on one shared condvar but
    /// each waits for its *own* id, so waking a single arbitrary thread could
    /// wake one waiting on a different operation and leave the right one asleep.
    fn mark_finished(&self, id: &str) {
        let mut finished = lock_or_recover(&self.finished);
        finished.insert(id.to_string());
        drop(finished);
        self.condvar.notify_all();
    }

    /// Has `id` already reached a final status?
    pub fn is_finished(&self, id: &str) -> bool {
        lock_or_recover(&self.finished).contains(id)
    }

    /// Block until `id` reaches a final status; `true` once it has.
    ///
    /// There is no lost-wakeup window: the finished-set is inspected while
    /// holding the very mutex the condvar is paired with, and a worker must take
    /// that same mutex to publish — so an operation that ends between a caller's
    /// check and its wait is already recorded and returns immediately.
    ///
    /// `timeout` bounds one wait, not the whole call, and exists only as a
    /// backstop for a signal that can never arrive (e.g. a worker killed before
    /// publishing). `None` waits indefinitely.
    pub fn wait_for(&self, id: &str, timeout: Option<Duration>) -> bool {
        let mut finished = lock_or_recover(&self.finished);
        loop {
            if finished.contains(id) {
                return true;
            }
            match timeout {
                Some(t) => {
                    let (guard, outcome) = self
                        .condvar
                        .wait_timeout(finished, t)
                        .unwrap_or_else(|poisoned| poisoned.into_inner());
                    finished = guard;
                    if outcome.timed_out() {
                        return finished.contains(id);
                    }
                }
                None => {
                    finished = self
                        .condvar
                        .wait(finished)
                        .unwrap_or_else(|poisoned| poisoned.into_inner());
                }
            }
        }
    }
}

impl std::fmt::Debug for OperationCompletion {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("OperationCompletion")
            .field("finished_len", &lock_or_recover(&self.finished).len())
            .finish()
    }
}

// Concrete handle implementation
struct OperationHandle {
    status: Arc<Mutex<OperationStatus>>,
    progress: Arc<Mutex<OperationProgress>>,
    cancel_flag: Arc<AtomicBool>,
    _join_handle: thread::JoinHandle<()>,
}

impl OperationHandleTrait for OperationHandle {
    fn get_status(&self) -> OperationStatus {
        lock_or_recover(&self.status).clone()
    }

    fn get_progress(&self) -> OperationProgress {
        lock_or_recover(&self.progress).clone()
    }

    fn cancel(&self) {
        self.cancel_flag.store(true, Ordering::Relaxed);
        let mut status = lock_or_recover(&self.status);
        if matches!(*status, OperationStatus::Running) {
            *status = OperationStatus::Cancelled;
        }
    }

    fn is_finished(&self) -> bool {
        matches!(
            self.get_status(),
            OperationStatus::Completed | OperationStatus::Cancelled | OperationStatus::Failed(_)
        )
    }
}

// Manager for long operations
pub struct LongOperationManager {
    operations: Arc<Mutex<HashMap<String, Box<dyn OperationHandleTrait>>>>,
    next_id: Arc<Mutex<u64>>,
    results: Arc<Mutex<HashMap<String, String>>>, // Store serialized results
    event_hub: Option<Arc<EventHub>>,
    /// Signalled by each worker as it ends; lets callers block until an
    /// operation finishes rather than polling. See [`OperationCompletion`].
    completion: Arc<OperationCompletion>,
}

impl std::fmt::Debug for LongOperationManager {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let operations_len = lock_or_recover(&self.operations).len();
        let next_id = *lock_or_recover(&self.next_id);
        let results_len = lock_or_recover(&self.results).len();

        f.debug_struct("LongOperationManager")
            .field("operations_len", &operations_len)
            .field("next_id", &next_id)
            .field("results_len", &results_len)
            .field("event_hub_set", &self.event_hub.is_some())
            .finish()
    }
}

impl LongOperationManager {
    pub fn new() -> Self {
        Self {
            operations: Arc::new(Mutex::new(HashMap::new())),
            next_id: Arc::new(Mutex::new(0)),
            results: Arc::new(Mutex::new(HashMap::new())),
            event_hub: None,
            completion: Arc::new(OperationCompletion::new()),
        }
    }

    /// Inject the event hub to allow sending long operation related events
    pub fn set_event_hub(&mut self, event_hub: &Arc<EventHub>) {
        self.event_hub = Some(Arc::clone(event_hub));
    }

    /// Start a new long operation and return its ID
    pub fn start_operation<Op: LongOperation>(&self, operation: Op) -> String {
        let id = {
            let mut next_id = lock_or_recover(&self.next_id);
            *next_id += 1;
            format!("op_{}", *next_id)
        };

        // Emit started event
        if let Some(event_hub) = &self.event_hub {
            event_hub.send_event(Event {
                origin: Origin::LongOperation(LongOperationEvent::Started),
                ids: vec![],
                data: Some(id.clone()),
            });
        }

        let status = Arc::new(Mutex::new(OperationStatus::Running));
        let progress = Arc::new(Mutex::new(OperationProgress::new(0.0, None)));
        let cancel_flag = Arc::new(AtomicBool::new(false));

        let status_clone = status.clone();
        let progress_clone = progress.clone();
        let cancel_flag_clone = cancel_flag.clone();
        let results_clone = self.results.clone();
        let id_clone = id.clone();
        let event_hub_opt = self.event_hub.clone();
        let completion_clone = self.completion.clone();

        let join_handle = thread::spawn(move || {
            let progress_callback = {
                let progress = progress_clone.clone();
                let event_hub_opt = event_hub_opt.clone();
                let id_for_cb = id_clone.clone();
                Box::new(move |prog: OperationProgress| {
                    *lock_or_recover(&progress) = prog.clone();
                    if let Some(event_hub) = &event_hub_opt {
                        let payload = serde_json::json!({
                            "id": id_for_cb,
                            "percentage": prog.percentage,
                            "message": prog.message,
                        })
                        .to_string();
                        event_hub.send_event(Event {
                            origin: Origin::LongOperation(LongOperationEvent::Progress),
                            ids: vec![],
                            data: Some(payload),
                        });
                    }
                }) as Box<dyn Fn(OperationProgress) + Send>
            };

            // Catch a panic in the operation body so a background failure is
            // reported as `Failed` (with the panic message) instead of leaving
            // the operation stuck at `Running` and the thread silently dead.
            let operation_result =
                match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    operation.execute(progress_callback, cancel_flag_clone.clone())
                })) {
                    Ok(result) => result,
                    Err(payload) => {
                        let msg = payload
                            .downcast_ref::<&str>()
                            .map(|s| s.to_string())
                            .or_else(|| payload.downcast_ref::<String>().cloned())
                            .unwrap_or_else(|| "operation panicked".to_string());
                        Err(anyhow::anyhow!("operation panicked: {msg}"))
                    }
                };

            let final_status = if cancel_flag_clone.load(Ordering::Relaxed) {
                OperationStatus::Cancelled
            } else {
                match &operation_result {
                    Ok(result) => {
                        // Store the result
                        if let Ok(serialized) = serde_json::to_string(result) {
                            let mut results = lock_or_recover(&results_clone);
                            results.insert(id_clone.clone(), serialized);
                        }
                        OperationStatus::Completed
                    }
                    Err(e) => OperationStatus::Failed(e.to_string()),
                }
            };

            // Emit final status event
            if let Some(event_hub) = &event_hub_opt {
                let (event, data) = match &final_status {
                    OperationStatus::Completed => (
                        LongOperationEvent::Completed,
                        serde_json::json!({"id": id_clone}).to_string(),
                    ),
                    OperationStatus::Cancelled => (
                        LongOperationEvent::Cancelled,
                        serde_json::json!({"id": id_clone}).to_string(),
                    ),
                    OperationStatus::Failed(err) => (
                        LongOperationEvent::Failed,
                        serde_json::json!({"id": id_clone, "error": err}).to_string(),
                    ),
                    OperationStatus::Running => (
                        LongOperationEvent::Progress,
                        serde_json::json!({"id": id_clone}).to_string(),
                    ),
                };
                event_hub.send_event(Event {
                    origin: Origin::LongOperation(event),
                    ids: vec![],
                    data: Some(data),
                });
            }

            *lock_or_recover(&status_clone) = final_status;

            // Publish completion **last** — after the result is stored, the
            // event emitted and the final status written — so any thread woken
            // by this observes a fully-settled operation rather than racing the
            // bookkeeping that follows it.
            completion_clone.mark_finished(&id_clone);
        });

        let handle = OperationHandle {
            status,
            progress,
            cancel_flag,
            _join_handle: join_handle,
        };

        lock_or_recover(&self.operations).insert(id.clone(), Box::new(handle));

        id
    }

    /// Get the status of an operation
    pub fn get_operation_status(&self, id: &str) -> Option<OperationStatus> {
        let operations = lock_or_recover(&self.operations);
        operations.get(id).map(|handle| handle.get_status())
    }

    /// Get the progress of an operation
    pub fn get_operation_progress(&self, id: &str) -> Option<OperationProgress> {
        let operations = lock_or_recover(&self.operations);
        operations.get(id).map(|handle| handle.get_progress())
    }

    /// Cancel an operation
    pub fn cancel_operation(&self, id: &str) -> bool {
        let operations = lock_or_recover(&self.operations);
        if let Some(handle) = operations.get(id) {
            handle.cancel();
            // Emit cancelled event immediately
            if let Some(event_hub) = &self.event_hub {
                let payload = serde_json::json!({"id": id}).to_string();
                event_hub.send_event(Event {
                    origin: Origin::LongOperation(LongOperationEvent::Cancelled),
                    ids: vec![],
                    data: Some(payload),
                });
            }
            true
        } else {
            false
        }
    }

    /// Check if an operation is finished
    pub fn is_operation_finished(&self, id: &str) -> Option<bool> {
        let operations = lock_or_recover(&self.operations);
        operations.get(id).map(|handle| handle.is_finished())
    }

    /// Remove finished operations from memory
    pub fn cleanup_finished_operations(&self) {
        let mut operations = lock_or_recover(&self.operations);
        operations.retain(|_, handle| !handle.is_finished());
    }

    /// Get list of all operation IDs
    pub fn list_operations(&self) -> Vec<String> {
        let operations = lock_or_recover(&self.operations);
        operations.keys().cloned().collect()
    }

    /// Get summary of all operations
    pub fn get_operations_summary(&self) -> Vec<(String, OperationStatus, OperationProgress)> {
        let operations = lock_or_recover(&self.operations);
        operations
            .iter()
            .map(|(id, handle)| (id.clone(), handle.get_status(), handle.get_progress()))
            .collect()
    }

    /// Store an operation result
    pub fn store_operation_result<T: serde::Serialize>(&self, id: &str, result: T) -> Result<()> {
        let serialized = serde_json::to_string(&result)?;
        let mut results = lock_or_recover(&self.results);
        results.insert(id.to_string(), serialized);
        Ok(())
    }

    /// Get an operation result
    pub fn get_operation_result(&self, id: &str) -> Option<String> {
        let results = lock_or_recover(&self.results);
        results.get(id).cloned()
    }

    /// A cloneable handle to the completion signal — the supported way to
    /// **block** until an operation finishes instead of re-checking it on a
    /// timer.
    ///
    /// Handed out as an `Arc` by design. A manager is normally reached through a
    /// shared lock, and blocking while holding that lock would stall every other
    /// operation query for as long as the wait lasts. So take the handle, drop
    /// the manager lock, *then* wait:
    ///
    /// ```rust,ignore
    /// let completion = ctx.long_operation_manager.lock().completion_signal();
    /// // manager lock released here
    /// completion.wait_for(&op_id, Some(Duration::from_secs(30)));
    /// let result = ctx.long_operation_manager.lock().get_operation_result(&op_id);
    /// ```
    pub fn completion_signal(&self) -> Arc<OperationCompletion> {
        Arc::clone(&self.completion)
    }

    /// Block until `id` finishes; `true` once it has.
    ///
    /// Convenience for an owner holding the manager directly. **Do not call this
    /// through a shared lock** — it blocks for the operation's whole duration and
    /// would hold that lock the entire time; take
    /// [`completion_signal`](Self::completion_signal) and wait on that instead.
    pub fn wait_for_operation(&self, id: &str, timeout: Option<Duration>) -> bool {
        self.completion.wait_for(id, timeout)
    }
}

impl Default for LongOperationManager {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::anyhow;
    use std::time::Duration;

    // Example implementation of a long operation
    pub struct FileProcessingOperation {
        pub _file_path: String,
        pub total_files: usize,
    }

    impl LongOperation for FileProcessingOperation {
        type Output = ();

        fn execute(
            &self,
            progress_callback: Box<dyn Fn(OperationProgress) + Send>,
            cancel_flag: Arc<AtomicBool>,
        ) -> Result<Self::Output> {
            for i in 0..self.total_files {
                // Check if operation was cancelled
                if cancel_flag.load(Ordering::Relaxed) {
                    return Err(anyhow!("Operation was cancelled".to_string()));
                }

                // Simulate work
                thread::sleep(Duration::from_millis(500));

                // Report progress
                let percentage = (i as f32 / self.total_files as f32) * 100.0;
                progress_callback(OperationProgress::new(
                    percentage,
                    Some(format!("Processing file {} of {}", i + 1, self.total_files)),
                ));
            }

            // Final progress
            progress_callback(OperationProgress::new(100.0, Some("Completed".to_string())));
            Ok(())
        }
    }

    /// An operation that finishes at once and returns a value — the shape of a
    /// small import, and the case a blocking wait must add no latency to.
    pub struct InstantOperation;

    impl LongOperation for InstantOperation {
        type Output = u32;

        fn execute(
            &self,
            _progress_callback: Box<dyn Fn(OperationProgress) + Send>,
            _cancel_flag: Arc<AtomicBool>,
        ) -> Result<Self::Output> {
            Ok(42)
        }
    }

    /// The whole point of the completion signal: a caller is woken the moment the
    /// worker publishes, with no polling interval and no fixed sleep floor. A
    /// trivial operation must therefore settle in far less than a timer tick —
    /// this is what stops N sequential small operations costing N × a poll period.
    #[test]
    fn wait_for_returns_as_soon_as_the_operation_finishes() {
        let manager = LongOperationManager::new();
        let completion = manager.completion_signal();
        let started = std::time::Instant::now();
        let op_id = manager.start_operation(InstantOperation);

        assert!(completion.wait_for(&op_id, Some(Duration::from_secs(5))));
        let elapsed = started.elapsed();

        // The result is stored *before* completion is published, so a woken
        // waiter can always read it straight back.
        assert_eq!(
            manager.get_operation_result(&op_id).as_deref(),
            Some("42"),
            "completion must imply the result is already retrievable"
        );
        assert!(
            elapsed < Duration::from_millis(50),
            "a trivial operation must not wait out a polling tick (took {elapsed:?})"
        );
    }

    /// Waiting on an operation that has already finished returns at once, instead
    /// of blocking for a signal that was sent before the caller arrived.
    #[test]
    fn wait_for_an_already_finished_operation_returns_immediately() {
        let manager = LongOperationManager::new();
        let completion = manager.completion_signal();
        let op_id = manager.start_operation(InstantOperation);
        assert!(completion.wait_for(&op_id, Some(Duration::from_secs(5))));

        // The finished record persists, so a second wait is instant.
        let started = std::time::Instant::now();
        assert!(completion.wait_for(&op_id, Some(Duration::from_secs(5))));
        assert!(started.elapsed() < Duration::from_millis(50));
        assert!(completion.is_finished(&op_id));
    }

    /// The timeout bounds one wait; it is not a deadline for the operation. A
    /// still-running operation reports `false` and leaves the caller free to wait
    /// again.
    #[test]
    fn wait_for_times_out_while_the_operation_is_still_running() {
        let manager = LongOperationManager::new();
        let completion = manager.completion_signal();
        let op_id = manager.start_operation(FileProcessingOperation {
            _file_path: "/tmp/test".to_string(),
            total_files: 5,
        });

        assert!(
            !completion.wait_for(&op_id, Some(Duration::from_millis(50))),
            "a running operation has published no completion"
        );
        assert!(!completion.is_finished(&op_id));
        manager.cancel_operation(&op_id);
    }

    /// A cancelled operation must still publish completion — otherwise a caller
    /// blocked on it would be stranded forever.
    #[test]
    fn a_cancelled_operation_still_publishes_completion() {
        let manager = LongOperationManager::new();
        let completion = manager.completion_signal();
        let op_id = manager.start_operation(FileProcessingOperation {
            _file_path: "/tmp/test".to_string(),
            total_files: 5,
        });
        manager.cancel_operation(&op_id);

        assert!(
            completion.wait_for(&op_id, Some(Duration::from_secs(5))),
            "cancellation must wake a blocked waiter, not strand it"
        );
    }

    #[test]
    fn test_operation_manager() {
        let manager = LongOperationManager::new();

        let operation = FileProcessingOperation {
            _file_path: "/tmp/test".to_string(),
            total_files: 5,
        };

        let op_id = manager.start_operation(operation);

        // Check initial status
        assert_eq!(
            manager.get_operation_status(&op_id),
            Some(OperationStatus::Running)
        );

        // Wait a bit and check progress
        thread::sleep(Duration::from_millis(100));
        let progress = manager.get_operation_progress(&op_id);
        assert!(progress.is_some());

        // Test cancellation
        assert!(manager.cancel_operation(&op_id));
        thread::sleep(Duration::from_millis(100));
        assert_eq!(
            manager.get_operation_status(&op_id),
            Some(OperationStatus::Cancelled)
        );
    }
}