runner_q 0.6.4

use crate::activity::activity::{
    Activity, ActivityFuture, ActivityHandler, ActivityHandlerRegistry, ActivityOption,
    ActivityPriority, OnDuplicate,
};
use crate::config::WorkerConfig;
use crate::queue::queue::{ActivityQueueTrait, ActivityResult, ResultState};
use crate::runner::error::WorkerError;
use crate::storage::{FailureKind, IdempotencyBehavior, QueuedActivity, Storage};
use crate::{ActivityContext, ActivityError};
use chrono::Utc;

#[cfg(feature = "postgres")]
use crate::observability::QueueInspector;
use futures::FutureExt;
use serde_json::json;
use std::panic::AssertUnwindSafe;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{watch, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};

/// Optional metrics sink to expose counters without coupling to a specific backend.
///
/// This trait allows you to collect metrics about activity processing, including
/// completion rates, retry counts, and execution times. Implement this trait
/// to integrate with your preferred metrics system (Prometheus, StatsD, etc.).
///
/// # Examples
///
/// ```rust,no_run
/// use runner_q::MetricsSink;
/// use std::time::Duration;
/// use std::sync::Arc;
///
/// // Prometheus metrics implementation
/// struct PrometheusMetrics {
///     // Contains pre-registered Prometheus metrics
/// }
///
/// impl MetricsSink for PrometheusMetrics {
///     fn inc_counter(&self, name: &str, value: u64) {
///         // Increment the appropriate counter based on name
///     }
///
///     fn observe_duration(&self, name: &str, duration: Duration) {
///         // Record the duration in the appropriate histogram
///     }
/// }
///
/// // Simple logging metrics implementation
/// struct LoggingMetrics;
///
/// impl MetricsSink for LoggingMetrics {
///     fn inc_counter(&self, name: &str, value: u64) {
///         println!("METRIC: {} += {}", name, value);
///     }
///
///     fn observe_duration(&self, name: &str, duration: Duration) {
///         println!("METRIC: {} = {:?}", name, duration);
///     }
/// }
/// ```
pub trait MetricsSink: Send + Sync + 'static {
    /// Increment a counter metric by the specified value.
    ///
    /// This is typically used for counting events like activity completions,
    /// retries, failures, etc.
    fn inc_counter(&self, name: &str, value: u64);

    /// Record a duration metric.
    ///
    /// This is typically used for measuring execution times, queue wait times, etc.
    /// The default implementation is a no-op, so you only need to implement this
    /// if you want to collect duration metrics.
    fn observe_duration(&self, _name: &str, _dur: Duration) {
        let _ = (_name, _dur);
    }
}

/// No-op metrics sink that discards all metrics.
///
/// This is the default metrics implementation that does nothing with the metrics.
/// Use this when you don't need metrics collection or as a fallback.
///
/// # Examples
///
/// ```rust
/// use runner_q::{NoopMetrics, MetricsSink};
/// use std::time::Duration;
///
/// let metrics = NoopMetrics;
///
/// // These calls do nothing
/// metrics.inc_counter("activities_completed", 1);
/// metrics.observe_duration("activity_execution", Duration::from_secs(5));
/// ```
pub struct NoopMetrics;

impl MetricsSink for NoopMetrics {
    fn inc_counter(&self, _name: &str, _value: u64) {}
}

/// Simple exponential backoff helper for idle polls
struct Backoff {
    current: Duration,
    base: Duration,
    max: Duration,
}
impl Backoff {
    fn new(base: Duration, max: Duration) -> Self {
        Self {
            current: base,
            base,
            max,
        }
    }
    fn reset(&mut self) {
        self.current = self.base;
    }
    fn next(&mut self) -> Duration {
        let next = self.current;
        self.current = (self.current.mul_f32(2.0)).min(self.max);
        next
    }
}

pub struct WorkerEngine {
    activity_queue: Arc<dyn ActivityQueueTrait>,
    backend: Arc<dyn Storage>,
    activity_handlers: ActivityHandlerRegistry, // kept as-is per request
    config: WorkerConfig,
    running: Arc<RwLock<bool>>, // retains external visibility
    shutdown_tx: watch::Sender<bool>,
    cancel_token: CancellationToken,
    metrics: Arc<dyn MetricsSink>,
}

impl WorkerEngine {
    /// Creates a new WorkerEngine with a custom backend implementation.
    ///
    /// This allows using alternative backends like Valkey, Kafka, or SQL-based implementations.
    ///
    /// # Parameters
    ///
    /// * `backend` - Custom backend implementing the [`Storage`] trait
    /// * `config` - Configuration settings for the worker engine
    pub fn new_with_backend(backend: Arc<dyn Storage>, config: WorkerConfig) -> Self {
        let (shutdown_tx, _shutdown_rx) = watch::channel(false);
        let adapter = Arc::new(BackendQueueAdapter::new(
            backend.clone(),
            config.activity_types.clone(),
        ));
        Self {
            activity_queue: adapter,
            backend,
            activity_handlers: ActivityHandlerRegistry::new(),
            config,
            running: Arc::new(RwLock::new(false)),
            shutdown_tx,
            cancel_token: CancellationToken::new(),
            metrics: Arc::new(NoopMetrics),
        }
    }

    pub fn with_metrics(&mut self, sink: Arc<dyn MetricsSink>) {
        self.metrics = sink;
    }

    /// Returns a `QueueInspector` for observability operations.
    ///
    /// The inspector uses the backend's inspection capabilities for
    /// reading queue state and activity information.
    #[cfg(feature = "postgres")]
    pub fn inspector(&self) -> QueueInspector {
        QueueInspector::new(self.backend.clone())
            .with_max_workers(self.config.max_concurrent_activities)
    }

    /// Creates a new WorkerEngineBuilder for fluent configuration.
    ///
    /// This provides a more ergonomic API for configuring the WorkerEngine with method chaining.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use runner_q::WorkerEngine;
    /// use std::time::Duration;
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let engine = WorkerEngine::builder()
    ///     .redis_url("redis://localhost:6379")
    ///     .queue_name("my_app")
    ///     .max_workers(8)
    ///     .schedule_poll_interval(Duration::from_secs(30))
    ///     .build()
    ///     .await?;
    /// # Ok(())
    /// # }
    ///
    /// // With custom metrics
    /// # async fn example_with_metrics() -> Result<(), Box<dyn std::error::Error>> {
    /// # use runner_q::{WorkerEngine, MetricsSink, storage::PostgresBackend};
    /// # use std::sync::Arc;
    /// struct PrometheusMetrics;
    /// impl MetricsSink for PrometheusMetrics {
    ///     fn inc_counter(&self, name: &str, value: u64) { let _ = (name, value); }
    ///     fn observe_duration(&self, name: &str, duration: Duration) { let _ = (name, duration); }
    /// }
    /// let backend = PostgresBackend::new("postgres://localhost/mydb", "my_app").await?;
    /// let engine = WorkerEngine::builder()
    ///     .backend(Arc::new(backend))
    ///     .queue_name("my_app")
    ///     .max_workers(8)
    ///     .metrics(Arc::new(PrometheusMetrics))
    ///     .build()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn builder() -> WorkerEngineBuilder {
        WorkerEngineBuilder::new()
    }

    /// Starts the worker engine and manages its full execution lifecycle.
    ///
    /// This method performs the following steps:
    /// - Verifies that the engine is not already running, returning `WorkerError::AlreadyRunning` if it is.
    /// - Marks the engine as active.
    /// - Spawns both:
    ///   - a background task that periodically processes scheduled activities, and
    ///   - a pool of worker loops (one per available concurrency slot) that dequeue and execute activities.
    /// - Awaits either:
    ///   - a shutdown signal (Ctrl+C or SIGTERM), or
    ///   - the completion or failure of any worker loop.
    /// - Once a shutdown signal is received, transitions the engine into graceful stop mode,
    ///   halting new activity execution and allowing in-flight work to complete before cleanup.
    ///
    /// # Behavior
    ///
    /// - The method runs until the engine is explicitly stopped or a shutdown signal is received.
    /// - When it returns, the engine is fully stopped and resources have been released.
    /// - If the engine was already running, it immediately returns `Err(WorkerError::AlreadyRunning)`.
    ///
    /// # Returns
    ///
    /// - `Ok(())` — the engine shut down cleanly.
    /// - `Err(WorkerError::AlreadyRunning)` — start was attempted while another instance was active.
    /// - Other `WorkerError` variants — internal initialization or runtime failures.
    ///
    /// # Examples
    ///
    /// ```
    /// # async fn example() -> Result<(), WorkerError> {
    /// // Initialize the engine (pseudo-code)
    /// let engine = WorkerEngine::new(redis_pool, config);
    ///
    /// // Start processing activities — this call will block until shutdown
    /// engine.start().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn start(&self) -> Result<(), WorkerError> {
        {
            let mut running = self.running.write().await;
            if *running {
                return Err(WorkerError::AlreadyRunning);
            }
            *running = true;
        }

        if let Some(ref types) = self.config.activity_types {
            let missing: Vec<_> = types
                .iter()
                .filter(|t| !self.activity_handlers.contains_key(t.as_str()))
                .collect();
            if !missing.is_empty() {
                panic!(
                    "activity_types filter contains types with no registered handler: {:?}",
                    missing
                );
            }
        }

        info!(
            max_concurrent_activities = self.config.max_concurrent_activities,
            "Starting worker engine"
        );

        let mut join_handles = Vec::new();

        // Scheduled activities processor (skipped if backend handles it in dequeue)
        if !self.activity_queue.schedules_natively() {
            let scheduled_handle = self.start_scheduled_activities_processor().await;
            join_handles.push(scheduled_handle);
        }

        // Reaper processor for re-queueing expired processing items
        let reaper_handle = self.start_reaper_processor().await;
        join_handles.push(reaper_handle);

        // Worker loops — one dedicated loop per worker, no semaphore needed
        for worker_id in 0..self.config.max_concurrent_activities {
            let handle = self.start_worker_loop(worker_id).await;
            join_handles.push(handle);
        }

        // Wait for shutdown signal or all workers finishing.
        tokio::select! {
            _ = self.wait_for_shutdown() => {
                info!("Shutdown signal received, stopping worker engine");
            }
            result = futures::future::try_join_all(join_handles) => {
                match result {
                    Ok(_) => info!("All worker loops completed"),
                    Err(e) => error!(error = %e, "A worker task failed"),
                }
            }
        }

        self.stop().await;
        info!("Worker engine stopped");
        Ok(())
    }

    /// Signal a graceful stop of the worker engine.
    ///
    /// This method initiates a graceful shutdown process:
    /// - Stops accepting new activities
    /// - Allows currently running activities to complete
    /// - Cancels any pending operations
    /// - Releases resources
    ///
    /// The shutdown is cooperative - activities should check the cancellation token
    /// and exit cleanly when requested.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use runner_q::{WorkerEngine, WorkerConfig};
    /// use std::sync::Arc;
    /// use tokio::time::{sleep, Duration};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut engine = WorkerEngine::new(redis_pool, config);
    ///
    /// // Start the engine in a background task
    /// let engine_handle = tokio::spawn(async move {
    ///     engine.start().await
    /// });
    ///
    /// // Let it run for a while
    /// sleep(Duration::from_secs(10)).await;
    ///
    /// // Gracefully stop the engine
    /// engine.stop().await;
    ///
    /// // Wait for the engine to finish
    /// engine_handle.await??;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn stop(&self) {
        info!("Stopping worker engine");
        let mut running = self.running.write().await;
        *running = false;
        // broadcast shutdown
        let _ = self.shutdown_tx.send(true);
        self.cancel_token.cancel();
    }

    /// Spawns a background worker loop that continuously dequeues and executes activities.
    ///
    /// Each worker operates independently and performs the following cycle while the engine is running:
    /// - Attempts to dequeue an activity from the queue (waiting briefly if none are available).
    /// - Looks up the registered handler for the dequeued activity type.
    /// - Executes the handler with a per-activity timeout, passing in an `ActivityContext` containing
    ///   metadata and a reference back to the engine for nested execution.
    /// - Records the outcome through the activity queue:
    ///   - **Success:** Marks the activity as completed and stores the result with `ResultState::Ok`.
    ///   - **Retryable failure (`ActivityError::Retry`)**: Marks the activity as failed and eligible for retry.
    ///   - **Non-retryable failure (`ActivityError::NonRetry`)**: Marks the activity as permanently failed and
    ///     stores a structured JSON error (`{"error": <reason>, "type": "non_retryable", "failed_at": <RFC3339>}`).
    ///   - **Timeout:** If execution exceeds its allowed duration, the activity is marked as failed and retried.
    ///
    /// The worker loop terminates when:
    /// - The engine’s running flag is cleared (via `stop()`), or
    /// - A shutdown signal is received through cooperative cancellation.
    ///
    /// Upon termination, any ongoing work completes its current iteration before the worker exits.
    ///
    /// # Returns
    ///
    /// Returns a [`tokio::task::JoinHandle`] wrapping the background worker task.
    /// The handle can be awaited to monitor completion or detached to run in the background.
    ///
    /// # Notes
    ///
    /// - This function does not block; it spawns a background task and immediately returns.
    /// - Concurrency is bounded by the number of worker loops spawned
    ///   (`max_concurrent_activities`), so no additional semaphore is needed.
    async fn start_worker_loop(
        &self,
        worker_id: usize,
    ) -> tokio::task::JoinHandle<Result<(), WorkerError>> {
        let running = self.running.clone();
        let activity_queue = self.activity_queue.clone();
        let activity_handlers = self.activity_handlers.clone();
        let activity_queue_for_context = self.activity_queue.clone();
        let mut shutdown_rx = self.shutdown_tx.subscribe();
        let cancel_token = self.cancel_token.clone();
        let metrics = self.metrics.clone();

        tokio::spawn(async move {
            debug!(%worker_id, "Starting worker loop");
            let worker_label = format!("worker-{}", worker_id);
            let mut backoff = Backoff::new(Duration::from_millis(100), Duration::from_secs(5));

            while *running.read().await {
                // Honor fast shutdown
                if *shutdown_rx.borrow() {
                    break;
                }

                // Dequeue with cooperative shutdown
                let dequeue_fut = activity_queue.dequeue(Duration::from_secs(1), &worker_label);
                let activity_opt = tokio::select! {
                    _ = shutdown_rx.changed() => { break; }
                    res = dequeue_fut => res
                };

                let activity = match activity_opt {
                    Ok(Some(a)) => {
                        backoff.reset();
                        a
                    }
                    Ok(None) => {
                        let sleep_for = backoff.next();
                        tokio::select! {
                            _ = tokio::time::sleep(sleep_for) => {},
                            _ = shutdown_rx.changed() => break,
                        }
                        continue;
                    }
                    Err(e) => {
                        error!(%worker_id, error = %e, "Failed to dequeue activity");
                        tokio::select! {
                            _ = tokio::time::sleep(Duration::from_secs(1)) => {},
                            _ = shutdown_rx.changed() => break,
                        }
                        continue;
                    }
                };

                // Resolve handler
                let activity_id = activity.id;
                let activity_type = activity.activity_type.clone();

                debug!(%worker_id, activity_id = %activity_id, activity_type = ?activity_type, "Worker processing activity");

                let handler = match activity_handlers.get(&activity.activity_type) {
                    Some(h) => h.clone(),
                    None => {
                        error!(%worker_id, activity_id = %activity_id, activity_type = ?activity_type, "No handler found for activity type");
                        if let Err(e) = activity_queue
                            .mark_failed(
                                activity,
                                "handler_not_found".to_string(),
                                false,
                                worker_label.as_str(),
                            )
                            .await
                        {
                            error!(%worker_id, activity_id = %activity_id, error = %e, "Failed to mark activity as failed");
                        }
                        continue;
                    }
                };

                // Prepare context
                let context = ActivityContext {
                    activity_id,
                    activity_type: activity_type.clone(),
                    retry_count: activity.retry_count,
                    metadata: activity.metadata.clone(),
                    cancel_token: cancel_token.child_token(),
                    activity_executor: Arc::new(WorkerEngineWrapper::new(
                        activity_queue_for_context.clone(),
                    )),
                };

                // Save payload for potential dead letter callback
                let payload_for_dead_letter = activity.payload.clone();

                let activity_timeout = Duration::from_secs(activity.timeout_seconds);
                let handle_fut = handler.handle(activity.payload.clone(), context);

                // Execute with timeout and cooperative shutdown
                let timed = tokio::select! {
                    _ = shutdown_rx.changed() => {
                        break;
                    }
                    res = tokio::time::timeout(activity_timeout, async {
                        // Use catch_unwind to catch panics from handle_fut.await
                        match AssertUnwindSafe(handle_fut).catch_unwind().await {
                            Ok(value) => value,
                            Err(e) => {
                                let err_msg = e.downcast_ref::<String>()
                                    .map(|s| format!("panic: {}", s))
                                    .or_else(|| e.downcast_ref::<&str>().map(|s| format!("panic: {}", s)))
                                    .unwrap_or_else(|| "panic (unknown)".to_string());
                                Err(ActivityError::Retry(err_msg))
                            },
                        }
                    }) => res,
                };

                match timed {
                    Ok(Ok(value)) => {
                        metrics.inc_counter("activity_completed", 1);
                        if let Err(e) = activity_queue
                            .mark_completed(&activity, worker_label.as_str())
                            .await
                        {
                            error!(%worker_id, activity_id = %activity_id, error = %e, "Failed to mark activity as completed");
                        }
                        info!(%worker_id, activity_id = %activity_id, activity_type = ?activity_type, "Activity completed successfully");

                        // Store result (fire-and-forget to avoid blocking the worker on slow I/O)
                        let aq = activity_queue.clone();
                        let result_to_store = ActivityResult {
                            data: value,
                            state: ResultState::Ok,
                        };
                        tokio::spawn(async move {
                            if let Err(e) = aq.store_result(activity_id, result_to_store).await {
                                error!(activity_id = %activity_id, error = %e, "Failed to store activity result");
                            }
                        });
                    }
                    Ok(Err(ActivityError::Retry(reason))) => {
                        metrics.inc_counter("activity_retry", 1);
                        warn!(%worker_id, activity_id = %activity_id, activity_type = ?activity_type, reason = %reason, "Activity requesting retry");
                        match activity_queue
                            .mark_failed(activity, reason.clone(), true, worker_label.as_str())
                            .await
                        {
                            Ok(true) => {
                                // Activity was dead-lettered, call the callback
                                let dead_letter_context = ActivityContext {
                                    activity_id,
                                    activity_type: activity_type.clone(),
                                    retry_count: 0, // Not relevant for dead letter callback
                                    metadata: Default::default(),
                                    cancel_token: cancel_token.child_token(),
                                    activity_executor: Arc::new(WorkerEngineWrapper::new(
                                        activity_queue_for_context.clone(),
                                    )),
                                };
                                handler
                                    .on_dead_letter(
                                        payload_for_dead_letter.clone(),
                                        dead_letter_context,
                                        reason,
                                    )
                                    .await;
                            }
                            Ok(false) => {
                                // Activity was retried or failed, no callback needed
                            }
                            Err(e) => {
                                error!(%worker_id, activity_id = %activity_id, error = %e, "Failed to mark activity for retry");
                            }
                        }
                    }
                    Ok(Err(ActivityError::NonRetry(reason))) => {
                        metrics.inc_counter("activity_failed_non_retry", 1);
                        error!(%worker_id, activity_id = %activity_id, activity_type = ?activity_type, reason = %reason, "Activity failed");
                        if let Err(e) = activity_queue
                            .mark_failed(activity, reason.clone(), false, worker_label.as_str())
                            .await
                        {
                            error!(%worker_id, activity_id = %activity_id, error = %e, "Failed to mark activity as failed");
                        }
                        let aq = activity_queue.clone();
                        tokio::spawn(async move {
                            let activity_result = ActivityResult {
                                data: Some(json!({
                                    "error": reason,
                                    "type": "non_retryable",
                                    "failed_at": Utc::now().to_rfc3339()
                                })),
                                state: ResultState::Err,
                            };
                            if let Err(e) = aq.store_result(activity_id, activity_result).await {
                                error!(activity_id = %activity_id, error = %e, "Failed to store activity result");
                            }
                        });
                    }
                    Err(_elapsed) => {
                        metrics.inc_counter("activity_timeout", 1);
                        let error_msg = "Activity execution timed out".to_string();
                        error!(%worker_id, activity_id = %activity_id, activity_type = ?activity_type, timeout = ?activity_timeout, "Activity timed out");
                        match activity_queue
                            .mark_failed(activity, error_msg.clone(), true, worker_label.as_str())
                            .await
                        {
                            Ok(true) => {
                                // Activity was dead-lettered, call the callback
                                let dead_letter_context = ActivityContext {
                                    activity_id,
                                    activity_type: activity_type.clone(),
                                    retry_count: 0,
                                    metadata: Default::default(),
                                    cancel_token: cancel_token.child_token(),
                                    activity_executor: Arc::new(WorkerEngineWrapper::new(
                                        activity_queue_for_context.clone(),
                                    )),
                                };
                                handler
                                    .on_dead_letter(
                                        payload_for_dead_letter.clone(),
                                        dead_letter_context,
                                        error_msg,
                                    )
                                    .await;
                            }
                            Ok(false) => {
                                // Activity was retried, no callback needed
                            }
                            Err(e) => {
                                error!(%worker_id, activity_id = %activity_id, error = %e, "Failed to mark activity as failed");
                            }
                        }
                    }
                }
            }

            debug!(%worker_id, "Worker loop stopped");
            Ok(())
        })
    }

    /// Spawns a background processor that periodically executes scheduled activities.
    ///
    /// This task runs continuously while the engine remains active, performing the following loop:
    /// - Calls [`process_scheduled_activities()`] on the engine’s activity queue to identify and enqueue
    ///   any activities whose scheduled execution time has arrived.
    /// - Logs any errors encountered during processing but continues operation.
    /// - Waits for a fixed interval (default: 30 seconds, configurable via `config.schedule_poll_interval_seconds`)
    ///   before repeating the cycle.
    ///
    /// The processor automatically stops when:
    /// - The engine’s running flag is cleared (via [`stop()`]), or
    /// - A shutdown signal is received through cooperative cancellation.
    ///
    /// This processor is **only started** for backends where
    /// [`schedules_natively()`](crate::storage::QueueStorage::schedules_natively)
    /// returns `false` (e.g. Redis). Backends that handle scheduled activities
    /// directly in `dequeue()` (e.g. PostgreSQL) skip this loop entirely.
    ///
    /// # Returns
    ///
    /// Returns a [`tokio::task::JoinHandle`] representing the spawned background processor.
    /// The handle can be:
    /// - **awaited**, to wait for graceful completion; or
    /// - **aborted**, to terminate the processor immediately.
    ///
    /// # Notes
    ///
    /// - Failures during scheduled-activity processing are logged and do not halt the engine.
    /// - The polling interval can be tuned through the worker configuration to balance responsiveness and load.
    ///
    /// # Examples
    ///
    /// ```
    /// // Assuming `engine` is an initialized WorkerEngine instance:
    /// let handle = engine.start_scheduled_activities_processor().await;
    ///
    /// // The processor runs in the background until the engine stops.
    /// // You may choose to abort or await it as needed:
    /// handle.abort();
    /// // or
    /// handle.await.ok();
    /// ```
    async fn start_scheduled_activities_processor(
        &self,
    ) -> tokio::task::JoinHandle<Result<(), WorkerError>> {
        let activity_queue = self.activity_queue.clone();
        let running = self.running.clone();
        let mut shutdown_rx = self.shutdown_tx.subscribe();

        // Make poll interval configurable; default to 30s if config lacks it.
        let poll_interval = {
            let secs = self
                .config
                .schedule_poll_interval_seconds
                .unwrap_or(5)
                .max(1);
            Duration::from_secs(secs)
        };

        tokio::spawn(async move {
            debug!("Starting scheduled activities processor");
            while *running.read().await {
                if *shutdown_rx.borrow() {
                    break;
                }

                if let Err(e) = activity_queue.process_scheduled_activities().await {
                    error!(error = %e, "Failed to process scheduled activities");
                }

                tokio::select! {
                    _ = tokio::time::sleep(poll_interval) => {},
                    _ = shutdown_rx.changed() => break,
                }
            }
            debug!("Scheduled activities processor stopped");
            Ok(())
        })
    }

    /// Spawns a background reaper that moves expired leased items back to the main queue.
    async fn start_reaper_processor(&self) -> tokio::task::JoinHandle<Result<(), WorkerError>> {
        let activity_queue = self.activity_queue.clone();
        let running = self.running.clone();
        let mut shutdown_rx = self.shutdown_tx.subscribe();

        let interval = Duration::from_secs(self.config.reaper_interval_seconds.unwrap_or(5).max(1));
        let batch_size = self.config.reaper_batch_size.unwrap_or(100);

        tokio::spawn(async move {
            debug!("Starting reaper processor");
            while *running.read().await {
                if *shutdown_rx.borrow() {
                    break;
                }

                if let Err(e) = activity_queue.requeue_expired(batch_size).await {
                    error!(error = %e, "Reaper failed to requeue expired items");
                }

                tokio::select! {
                    _ = tokio::time::sleep(interval) => {},
                    _ = shutdown_rx.changed() => break,
                }
            }
            debug!("Reaper processor stopped");
            Ok(())
        })
    }

    async fn wait_for_shutdown(&self) {
        let ctrl_c = async {
            tokio::signal::ctrl_c()
                .await
                .expect("failed to install Ctrl+C handler");
        };

        #[cfg(unix)]
        let terminate = async {
            use tokio::signal::unix::{signal, SignalKind};
            let mut sigterm =
                signal(SignalKind::terminate()).expect("failed to install SIGTERM handler");
            sigterm.recv().await;
        };

        #[cfg(not(unix))]
        let terminate = std::future::pending::<()>();

        tokio::select! {
            _ = ctrl_c => { info!("Received Ctrl+C signal"); },
            _ = terminate => { info!("Received SIGTERM signal"); },
        }

        self.stop().await;
    }
}

impl WorkerEngine {
    /// Register an activity handler for a specific activity type.
    ///
    /// This method associates an activity type string with a handler implementation.
    /// Activities of the registered type will be processed by the provided handler.
    ///
    /// # Parameters
    ///
    /// * `activity_type` - String identifier for the activity type
    /// * `activity` - Handler implementation for processing activities of this type
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use runner_q::{WorkerEngine, ActivityHandler, ActivityContext, ActivityHandlerResult};
    /// use async_trait::async_trait;
    /// use serde_json::Value;
    /// use std::sync::Arc;
    ///
    /// // Define a custom activity handler
    /// pub struct EmailHandler;
    ///
    /// #[async_trait]
    /// impl ActivityHandler for EmailHandler {
    ///     async fn handle(&self, payload: Value, _context: ActivityContext) -> ActivityHandlerResult {
    ///         println!("Sending email: {:?}", payload);
    ///         Ok(Some(serde_json::json!({"status": "sent"})))
    ///     }
    ///
    ///     fn activity_type(&self) -> String {
    ///         "send_email".to_string()
    ///     }
    /// }
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let mut engine = WorkerEngine::new(redis_pool, config);
    ///
    /// // Register the email handler
    /// engine.register_activity(
    ///     "send_email".to_string(),
    ///     Arc::new(EmailHandler)
    /// );
    ///
    /// // Now activities of type "send_email" will be processed by EmailHandler
    /// # Ok(())
    /// # }
    /// ```
    pub fn register_activity(&mut self, activity_type: String, activity: Arc<dyn ActivityHandler>) {
        self.activity_handlers.insert(activity_type, activity);
    }

    /// Get an activity executor for orchestrating activities from within handlers.
    ///
    /// This method returns an `ActivityExecutor` that can be used by activity handlers
    /// to execute other activities, enabling complex workflows and activity orchestration.
    ///
    /// The returned executor is thread-safe and can be shared across multiple handlers.
    ///
    /// # Returns
    ///
    /// Returns an `Arc<dyn ActivityExecutor>` that can be used to execute activities.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use runner_q::{WorkerEngine, ActivityHandler, ActivityContext, ActivityHandlerResult, ActivityOption, ActivityPriority};
    /// use async_trait::async_trait;
    /// use serde_json::Value;
    /// use std::sync::Arc;
    ///
    /// pub struct OrderProcessingHandler;
    ///
    /// #[async_trait]
    /// impl ActivityHandler for OrderProcessingHandler {
    ///     async fn handle(&self, payload: Value, context: ActivityContext) -> ActivityHandlerResult {
    ///         let order_id = payload["order_id"].as_str().unwrap();
    ///
    ///         // Execute payment processing activity
    ///         let payment_future = context.activity_executor.execute_activity(
    ///             "process_payment".to_string(),
    ///             serde_json::json!({"order_id": order_id, "amount": payload["amount"]}),
    ///             Some(ActivityOption {
    ///                 priority: Some(ActivityPriority::High),
    ///                 max_retries: 3,
    ///                 timeout_seconds: 300,
    ///                 delay_seconds: None,
    ///             })
    ///         ).await?;
    ///
    ///         // Execute inventory update activity
    ///         let inventory_future = context.activity_executor.execute_activity(
    ///             "update_inventory".to_string(),
    ///             serde_json::json!({"order_id": order_id, "items": payload["items"]}),
    ///             None // Use default options
    ///         ).await?;
    ///
    ///         Ok(Some(serde_json::json!({
    ///             "order_id": order_id,
    ///             "status": "processing",
    ///             "sub_activities": ["payment", "inventory"]
    ///         })))
    ///     }
    ///
    ///     fn activity_type(&self) -> String {
    ///         "process_order".to_string()
    ///     }
    /// }
    /// ```
    pub fn get_activity_executor(&self) -> Arc<dyn ActivityExecutor> {
        Arc::new(WorkerEngineWrapper::new(self.activity_queue.clone()))
    }
}

/// Builder for creating WorkerEngine instances with fluent configuration.
///
/// This builder provides a more ergonomic API for configuring the WorkerEngine
/// with method chaining instead of constructing a WorkerConfig struct directly.
///
/// # Examples
///
/// ```rust,no_run
/// use runner_q::{WorkerEngine, storage::PostgresBackend};
/// use std::sync::Arc;
/// use std::time::Duration;
///
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// let backend = PostgresBackend::new("postgres://localhost/mydb", "my_app").await?;
/// let engine = WorkerEngine::builder()
///     .backend(Arc::new(backend))
///     .queue_name("my_app")
///     .max_workers(8)
///     .schedule_poll_interval(Duration::from_secs(30))
///     .build()
///     .await?;
/// # Ok(())
/// # }
/// ```
pub struct WorkerEngineBuilder {
    queue_name: Option<String>,
    max_workers: Option<usize>,
    schedule_poll_interval: Option<Duration>,
    metrics: Option<Arc<dyn MetricsSink>>,
    backend: Option<Arc<dyn Storage>>,
    activity_types: Option<Vec<String>>,
}

impl WorkerEngineBuilder {
    /// Creates a new WorkerEngineBuilder with default values.
    pub fn new() -> Self {
        Self {
            queue_name: None,
            max_workers: None,
            schedule_poll_interval: None,
            metrics: None,
            backend: None,
            activity_types: None,
        }
    }

    /// Sets the queue name for the worker engine.
    ///
    /// # Parameters
    ///
    /// * `name` - Queue name used as key prefix
    pub fn queue_name(mut self, name: &str) -> Self {
        self.queue_name = Some(name.to_string());
        self
    }

    /// Sets the maximum number of concurrent workers.
    ///
    /// # Parameters
    ///
    /// * `max` - Maximum number of concurrent activities
    pub fn max_workers(mut self, max: usize) -> Self {
        self.max_workers = Some(max);
        self
    }

    /// Sets the schedule poll interval for processing scheduled activities.
    ///
    /// Only takes effect for backends that do not handle scheduling natively
    /// in `dequeue()`. The PostgreSQL backend skips the polling loop entirely.
    ///
    /// # Parameters
    ///
    /// * `interval` - Duration between scheduled activity polls
    pub fn schedule_poll_interval(mut self, interval: Duration) -> Self {
        self.schedule_poll_interval = Some(interval);
        self
    }

    /// Restricts this engine to only dequeue the specified activity types.
    ///
    /// When set, workers will only claim activities whose `activity_type`
    /// matches one of the listed values. When not set (the default),
    /// workers dequeue all activity types.
    ///
    /// At startup, the engine will panic if any listed type has no
    /// registered handler.
    pub fn activity_types(mut self, types: &[&str]) -> Self {
        self.activity_types = Some(types.iter().map(|s| s.to_string()).collect());
        self
    }

    /// Sets the metrics sink for monitoring activity processing.
    ///
    /// # Parameters
    ///
    /// * `sink` - Metrics implementation for collecting statistics
    pub fn metrics(mut self, sink: Arc<dyn MetricsSink>) -> Self {
        self.metrics = Some(sink);
        self
    }

    /// Sets the storage backend. Required—you must call this or `build()` will fail.
    ///
    /// Use `PostgresBackend::new(...)` for PostgreSQL, or the `runner_q_redis` crate
    /// for Redis.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// use runner_q::{WorkerEngine, storage::PostgresBackend};
    /// use std::sync::Arc;
    ///
    /// let backend = PostgresBackend::new("postgres://localhost/mydb", "my_app").await?;
    /// let engine = WorkerEngine::builder()
    ///     .backend(Arc::new(backend))
    ///     .max_workers(8)
    ///     .build()
    ///     .await?;
    /// ```
    pub fn backend(mut self, backend: Arc<dyn Storage>) -> Self {
        self.backend = Some(backend);
        self
    }

    /// Builds the WorkerEngine with the configured settings.
    ///
    /// # Returns
    ///
    /// Returns a `Result<WorkerEngine, WorkerError>` containing the configured engine
    /// or an error if connection fails.
    ///
    /// # Errors
    ///
    /// Returns `WorkerError` if connection cannot be established.
    pub async fn build(self) -> Result<WorkerEngine, WorkerError> {
        let max_concurrent_activities = self.max_workers.unwrap_or(10);
        let schedule_poll_interval_seconds = self.schedule_poll_interval.map_or(5, |d| d.as_secs());

        if let Some(backend) = self.backend {
            let queue_name = self.queue_name.unwrap_or_else(|| "default".to_string());
            let config = WorkerConfig {
                queue_name,
                max_concurrent_activities,
                schedule_poll_interval_seconds: Some(schedule_poll_interval_seconds),
                lease_ms: Some(60_000),
                reaper_interval_seconds: Some(5),
                reaper_batch_size: Some(100),
                activity_types: self.activity_types.clone(),
            };

            let mut worker_engine = WorkerEngine::new_with_backend(backend, config);

            if let Some(metrics) = self.metrics {
                worker_engine.with_metrics(metrics);
            }

            return Ok(worker_engine);
        }

        Err(WorkerError::Configuration(
            "No backend configured. Call .backend(Arc::new(your_backend)) before .build(). \
             Use PostgresBackend::new(...) for PostgreSQL, or the runner_q_redis crate for Redis.".to_string()
        ))
    }
}

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

/// Builder for creating and executing activities with fluent configuration.
///
/// This builder provides a more ergonomic API for activity execution with method chaining
/// instead of constructing an ActivityOption struct directly.
///
/// # Examples
///
/// ```rust,no_run
/// use runner_q::{WorkerEngine, ActivityPriority};
/// use serde_json::json;
/// use std::time::Duration;
///
/// # async fn example(engine: &WorkerEngine) -> Result<(), Box<dyn std::error::Error>> {
/// let future = engine
///     .activity("send_email")
///     .payload(json!({"to": "user@example.com", "subject": "Hello"}))
///     .priority(ActivityPriority::High)
///     .max_retries(5)
///     .timeout(Duration::from_secs(600))
///     .execute()
///     .await?;
/// # Ok(())
/// # }
/// ```
pub struct ActivityBuilder<'a> {
    engine: &'a WorkerEngineWrapper,
    activity_type: String,
    payload: Option<serde_json::Value>,
    priority: Option<ActivityPriority>,
    max_retries: Option<u32>,
    timeout: Option<Duration>,
    delay: Option<Duration>,
    idempotency_key: Option<(String, OnDuplicate)>,
}

impl<'a> ActivityBuilder<'a> {
    /// Creates a new ActivityBuilder for the given engine and activity type.
    pub fn new(engine: &'a WorkerEngineWrapper, activity_type: String) -> Self {
        Self {
            engine,
            activity_type,
            payload: None,
            priority: None,
            max_retries: None,
            timeout: None,
            delay: None,
            idempotency_key: None,
        }
    }

    /// Sets the payload for the activity.
    ///
    /// # Parameters
    ///
    /// * `payload` - JSON payload containing the activity data
    pub fn payload(mut self, payload: serde_json::Value) -> Self {
        self.payload = Some(payload);
        self
    }

    /// Sets the priority for the activity.
    ///
    /// # Parameters
    ///
    /// * `priority` - Activity priority level
    pub fn priority(mut self, priority: ActivityPriority) -> Self {
        self.priority = Some(priority);
        self
    }

    /// Sets the maximum number of retries for the activity.
    ///
    /// # Parameters
    ///
    /// * `retries` - Maximum number of retry attempts (0 for unlimited)
    pub fn max_retries(mut self, retries: u32) -> Self {
        self.max_retries = Some(retries);
        self
    }

    /// Sets the timeout for the activity execution.
    ///
    /// # Parameters
    ///
    /// * `timeout` - Maximum execution time before timeout
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = Some(timeout);
        self
    }

    /// Sets the delay before the activity should be executed.
    ///
    /// # Parameters
    ///
    /// * `delay` - Delay before execution
    pub fn delay(mut self, delay: Duration) -> Self {
        self.delay = Some(delay);
        self
    }

    /// Sets the idempotency key and behavior for duplicate detection.
    ///
    /// When an activity with the same idempotency key already exists, the behavior
    /// determines how the system handles the duplicate:
    /// - `AllowReuse`: Always create new activity, updating idempotency record
    /// - `ReturnExisting`: Return existing ActivityFuture if key exists
    /// - `AllowReuseOnFailure`: Only allow new activity if previous one failed
    /// - `NoReuse`: Return error if key exists
    ///
    /// The idempotency record TTL (1 day by default).
    ///
    /// # Parameters
    ///
    /// * `key` - Idempotency key string
    /// * `behavior` - Behavior when duplicate is detected
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use runner_q::{OnDuplicate, ActivityPriority};
    ///
    /// // Return existing ActivityFuture if key exists
    /// executor
    ///     .activity("process_payment")
    ///     .payload(payload)
    ///     .idempotency_key("payment-order-123", OnDuplicate::ReturnExisting)
    ///     .execute()
    ///     .await?;
    /// ```
    pub fn idempotency_key(mut self, key: impl Into<String>, behavior: OnDuplicate) -> Self {
        self.idempotency_key = Some((key.into(), behavior));
        self
    }

    /// Executes the activity with the configured settings.
    ///
    /// # Returns
    ///
    /// Returns a `Result<ActivityFuture, WorkerError>` containing the activity future
    /// or an error if the activity cannot be enqueued.
    ///
    /// # Errors
    ///
    /// Returns `WorkerError` if the activity cannot be enqueued or if there are Redis connection issues.
    pub async fn execute(self) -> Result<ActivityFuture, WorkerError> {
        let payload = self
            .payload
            .ok_or_else(|| WorkerError::QueueError("Activity payload is required".to_string()))?;

        let option = if self.priority.is_some()
            || self.max_retries.is_some()
            || self.timeout.is_some()
            || self.delay.is_some()
            || self.idempotency_key.is_some()
        {
            let idempotency_key = self
                .idempotency_key
                .map(|(key, behavior)| (format!("{}-{}", key, self.activity_type), behavior));
            Some(ActivityOption {
                priority: self.priority,
                max_retries: self.max_retries.unwrap_or(3),
                timeout_seconds: self.timeout.map(|d| d.as_secs()).unwrap_or(300),
                delay_seconds: self.delay.map(|d| d.as_secs()),
                idempotency_key,
            })
        } else {
            None
        };

        self.engine
            .execute_activity(self.activity_type, payload, option)
            .await
    }
}

/// Trait for executing activities, enabling activity orchestration.
///
/// This trait allows activity handlers to execute other activities, creating
/// complex workflows and orchestration patterns. It's provided to handlers
/// through the `ActivityContext` to enable nested activity execution.
///
/// # Examples
///
/// ```rust,no_run
/// use runner_q::{ActivityExecutor, ActivityOption, ActivityPriority};
/// use serde_json::json;
/// use std::sync::Arc;
///
/// # async fn example(executor: Arc<dyn ActivityExecutor>) -> Result<(), Box<dyn std::error::Error>> {
/// // Execute a simple activity
/// let future = executor.execute_activity(
///     "send_notification".to_string(),
///     json!({"user_id": 123, "message": "Hello"}),
///     None
/// ).await?;
///
/// // Execute a high-priority activity with custom options
/// let priority_future = executor.execute_activity(
///     "process_payment".to_string(),
///     json!({"amount": 100.0}),
///     Some(ActivityOption {
///         priority: Some(ActivityPriority::High),
///         max_retries: 5,
///         timeout_seconds: 600,
///         delay_seconds: None,
///     })
/// ).await?;
///
/// // Wait for completion
/// if let Ok(Some(result)) = future.get_result().await {
///     println!("Activity completed: {:?}", result);
/// }
/// # Ok(())
/// # }
/// ```
#[async_trait::async_trait]
pub trait ActivityExecutor: Send + Sync {
    /// Creates a fluent activity builder for executing activities with method chaining.
    ///
    /// This provides a more ergonomic API for activity execution with fluent configuration.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use runner_q::{WorkerEngine, ActivityPriority};
    /// use serde_json::json;
    /// use std::time::Duration;
    ///
    /// # async fn example(engine: &WorkerEngine) -> Result<(), Box<dyn std::error::Error>> {
    /// let future = engine
    ///     .activity("send_email")
    ///     .payload(json!({"to": "user@example.com", "subject": "Hello"}))
    ///     .priority(ActivityPriority::High)
    ///     .max_retries(5)
    ///     .timeout(Duration::from_secs(600))
    ///     .execute()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    fn activity(&self, activity_type: &str) -> ActivityBuilder<'_>;
}

/// Wrapper that provides activity execution capabilities to handlers.
///
/// This struct implements `ActivityExecutor` and is used internally to provide
/// activity execution capabilities to activity handlers through the `ActivityContext`.
/// It wraps the underlying activity queue to enable orchestration.
#[derive(Clone)]
pub struct WorkerEngineWrapper {
    activity_queue: Arc<dyn ActivityQueueTrait>,
}

impl WorkerEngineWrapper {
    pub(crate) fn new(activity_queue: Arc<dyn ActivityQueueTrait>) -> Self {
        Self { activity_queue }
    }
    /// Execute an activity and return a future for tracking its completion.
    ///
    /// This method enqueues an activity for processing and returns an `ActivityFuture`
    /// that can be used to wait for the activity's completion and retrieve its result.
    ///
    /// # Parameters
    ///
    /// * `activity_type` - String identifier for the activity type
    /// * `payload` - JSON payload containing the activity data
    /// * `option` - Optional configuration for the activity
    ///
    /// # Returns
    ///
    /// Returns an `ActivityFuture` that can be awaited to get the activity result.
    async fn execute_activity(
        &self,
        activity_type: String,
        payload: serde_json::Value,
        option: Option<ActivityOption>,
    ) -> Result<ActivityFuture, WorkerError> {
        let activity = Activity::new(activity_type, payload, option);
        let activity_id = activity.id;

        // Evaluate idempotency rules early, before enqueueing
        if let Some(existing_id) = self
            .activity_queue
            .evaluate_idempotency_rule(&activity)
            .await?
        {
            // ReturnExisting behavior: return the existing ActivityFuture
            return Ok(ActivityFuture::new(
                self.activity_queue.clone(),
                existing_id,
            ));
        }

        // Proceed with enqueueing the new activity
        match activity.scheduled_at {
            None => self.activity_queue.enqueue(activity).await?,
            Some(_) => self.activity_queue.schedule_activity(activity).await?,
        }
        Ok(ActivityFuture::new(
            self.activity_queue.clone(),
            activity_id,
        ))
    }
}

#[async_trait::async_trait]
impl ActivityExecutor for WorkerEngineWrapper {
    /// Creates a fluent activity builder for executing activities with method chaining.
    ///
    /// This provides a more ergonomic API for activity execution with fluent configuration.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use runner_q::{WorkerEngine, ActivityPriority};
    /// use serde_json::json;
    /// use std::time::Duration;
    ///
    /// # async fn example(engine: &WorkerEngine) -> Result<(), Box<dyn std::error::Error>> {
    /// let future = engine
    ///     .activity("send_email")
    ///     .payload(json!({"to": "user@example.com", "subject": "Hello"}))
    ///     .priority(ActivityPriority::High)
    ///     .max_retries(5)
    ///     .timeout(Duration::from_secs(600))
    ///     .execute()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    fn activity(&self, activity_type: &str) -> ActivityBuilder<'_> {
        ActivityBuilder::new(self, activity_type.to_string())
    }
}

// ============================================================================
// BackendQueueAdapter - Bridges Backend trait to ActivityQueueTrait
// ============================================================================

/// Adapter that wraps a [`Storage`] implementation to provide [`ActivityQueueTrait`] interface.
///
/// This allows the existing `WorkerEngine` internals to work with custom backends
/// without requiring changes to the core processing logic.
struct BackendQueueAdapter {
    backend: Arc<dyn Storage>,
    activity_types: Option<Vec<String>>,
}

impl BackendQueueAdapter {
    fn new(backend: Arc<dyn Storage>, activity_types: Option<Vec<String>>) -> Self {
        Self {
            backend,
            activity_types,
        }
    }

    fn activity_to_queued(activity: &Activity) -> QueuedActivity {
        QueuedActivity {
            id: activity.id,
            activity_type: activity.activity_type.clone(),
            payload: activity.payload.clone(),
            priority: activity.priority.clone(),
            max_retries: activity.max_retries,
            retry_count: activity.retry_count,
            timeout_seconds: activity.timeout_seconds,
            retry_delay_seconds: activity.retry_delay_seconds,
            scheduled_at: activity.scheduled_at,
            metadata: activity.metadata.clone(),
            idempotency_key: activity.idempotency_key.as_ref().map(|(k, b)| {
                let behavior = match b {
                    OnDuplicate::AllowReuse => IdempotencyBehavior::AllowReuse,
                    OnDuplicate::ReturnExisting => IdempotencyBehavior::ReturnExisting,
                    OnDuplicate::AllowReuseOnFailure => IdempotencyBehavior::AllowReuseOnFailure,
                    OnDuplicate::NoReuse => IdempotencyBehavior::NoReuse,
                };
                (k.clone(), behavior)
            }),
            created_at: activity.created_at,
        }
    }

    fn queued_to_activity(queued: &QueuedActivity) -> Activity {
        Activity {
            id: queued.id,
            activity_type: queued.activity_type.clone(),
            payload: queued.payload.clone(),
            priority: queued.priority.clone(),
            status: crate::ActivityStatus::Pending,
            created_at: queued.created_at,
            scheduled_at: queued.scheduled_at,
            retry_count: queued.retry_count,
            max_retries: queued.max_retries,
            timeout_seconds: queued.timeout_seconds,
            retry_delay_seconds: queued.retry_delay_seconds,
            metadata: queued.metadata.clone(),
            idempotency_key: queued.idempotency_key.as_ref().map(|(k, b)| {
                let behavior = match b {
                    IdempotencyBehavior::AllowReuse => OnDuplicate::AllowReuse,
                    IdempotencyBehavior::ReturnExisting => OnDuplicate::ReturnExisting,
                    IdempotencyBehavior::AllowReuseOnFailure => OnDuplicate::AllowReuseOnFailure,
                    IdempotencyBehavior::NoReuse => OnDuplicate::NoReuse,
                };
                (k.clone(), behavior)
            }),
        }
    }
}

#[async_trait::async_trait]
impl ActivityQueueTrait for BackendQueueAdapter {
    async fn enqueue(&self, activity: Activity) -> Result<(), WorkerError> {
        let queued = Self::activity_to_queued(&activity);
        self.backend.enqueue(queued).await.map_err(Into::into)
    }

    async fn dequeue(
        &self,
        timeout: Duration,
        worker_id: &str,
    ) -> Result<Option<Activity>, WorkerError> {
        let types_ref = self.activity_types.as_deref();
        match self.backend.dequeue(worker_id, timeout, types_ref).await? {
            Some(activity) => Ok(Some(Self::queued_to_activity(&activity))),
            None => Ok(None),
        }
    }

    async fn mark_completed(
        &self,
        activity: &Activity,
        worker_id: &str,
    ) -> Result<(), WorkerError> {
        self.backend
            .ack_success(activity.id, None, worker_id)
            .await
            .map_err(Into::into)
    }

    async fn mark_failed(
        &self,
        activity: Activity,
        error_message: String,
        retryable: bool,
        worker_id: &str,
    ) -> Result<bool, WorkerError> {
        let failure = if retryable {
            FailureKind::Retryable {
                reason: error_message,
            }
        } else {
            FailureKind::NonRetryable {
                reason: error_message,
            }
        };
        self.backend
            .ack_failure(activity.id, failure, worker_id)
            .await
            .map_err(Into::into)
    }

    async fn schedule_activity(&self, activity: Activity) -> Result<(), WorkerError> {
        let mut queued = Self::activity_to_queued(&activity);
        if queued.scheduled_at.is_none() {
            queued.scheduled_at = Some(Utc::now());
        }
        self.backend.enqueue(queued).await.map_err(Into::into)
    }

    async fn process_scheduled_activities(&self) -> Result<Vec<Activity>, WorkerError> {
        let _count = self.backend.process_scheduled().await?;
        // Return empty vec - activities are moved to main queue internally
        Ok(vec![])
    }

    async fn requeue_expired(&self, max_to_process: usize) -> Result<u64, WorkerError> {
        self.backend
            .requeue_expired(max_to_process)
            .await
            .map_err(Into::into)
    }

    async fn evaluate_idempotency_rule(
        &self,
        activity: &Activity,
    ) -> Result<Option<uuid::Uuid>, WorkerError> {
        let queued = Self::activity_to_queued(activity);
        self.backend
            .check_idempotency(&queued)
            .await
            .map_err(Into::into)
    }

    async fn extend_lease(
        &self,
        activity_id: uuid::Uuid,
        extend_by: Duration,
    ) -> Result<bool, WorkerError> {
        self.backend
            .extend_lease(activity_id, extend_by)
            .await
            .map_err(Into::into)
    }

    async fn store_result(
        &self,
        activity_id: uuid::Uuid,
        result: ActivityResult,
    ) -> Result<(), WorkerError> {
        let backend_result = crate::storage::ActivityResult {
            data: result.data,
            state: match result.state {
                ResultState::Ok => crate::storage::ResultState::Ok,
                ResultState::Err => crate::storage::ResultState::Err,
            },
        };
        self.backend
            .store_result(activity_id, backend_result)
            .await
            .map_err(Into::into)
    }

    async fn get_result(
        &self,
        activity_id: uuid::Uuid,
    ) -> Result<Option<ActivityResult>, WorkerError> {
        match self.backend.get_result(activity_id).await? {
            Some(backend_result) => Ok(Some(ActivityResult {
                data: backend_result.data,
                state: match backend_result.state {
                    crate::storage::ResultState::Ok => ResultState::Ok,
                    crate::storage::ResultState::Err => ResultState::Err,
                },
            })),
            None => Ok(None),
        }
    }

    fn schedules_natively(&self) -> bool {
        self.backend.schedules_natively()
    }
}