
use crate::{registry::get_job_handler, config::is_shutting_down};
use crate::utils::rdconfig::{get_redis_connection, get_dedicated_connection};
use tokio::time::{sleep, Duration};
use redis::AsyncCommands;
use chrono::Utc;
use crate::utils::constants::{DELAYED_JOBS_KEY, MAX_RETRIES};

/// How long a worker blocks on BLMOVE before looping to re-check shutdown.
const BLOCK_TIMEOUT_SECS: f64 = 5.0;



/// Requeue any job ids left in the per-queue in-flight list by a previous run that
/// crashed mid-processing, so they are retried instead of lost.
///
/// Note: this assumes a single engine process per queue. If you run multiple engine
/// processes sharing one Redis, a restart of one may requeue jobs another is actively
/// processing (at-least-once semantics — handlers should be idempotent).
async fn recover_inflight_jobs(queue: &str) {
    let inflight_key = format!("snm:inflight:{}", queue);
    let queue_key = format!("snm:queue:{}", queue);
    let mut conn = match get_redis_connection().await {
        Ok(c) => c,
        Err(_) => return,
    };

    let mut recovered = 0u64;
    loop {
        // Move oldest in-flight id back to the head of the queue.
        let moved: Option<String> = conn
            .lmove(&inflight_key, &queue_key, redis::Direction::Right, redis::Direction::Left)
            .await
            .unwrap_or(None);
        if moved.is_none() {
            break;
        }
        recovered += 1;
    }

    if recovered > 0 {
        tracing::warn!("Recovered {} orphaned in-flight job(s) for queue '{}'", recovered, queue);
    }
}

pub async fn start_worker_pool(queue: &str, concurrency: usize) {
    recover_inflight_jobs(queue).await;
    for _ in 0..concurrency {
        let queue = queue.to_string();

        tokio::spawn(async move {
            let queue_key = format!("snm:queue:{}", queue);
            let inflight_key = format!("snm:inflight:{}", queue);

            // Dedicated connection for the blocking BLMOVE fetch, so it never stalls
            // the shared connection used for enqueue/metrics/result writes.
            let mut fetch_conn = match get_dedicated_connection().await {
                Ok(c) => c,
                Err(_) => return,
            };

            loop {
                if is_shutting_down() {
                    break;
                }

                // Reliable fetch: atomically move the job id from the queue head into a
                // per-queue in-flight list, so a crash mid-processing doesn't lose the job.
                // BLMOVE blocks up to BLOCK_TIMEOUT_SECS instead of busy-polling; a short
                // timeout lets the loop periodically re-check the shutdown flag.
                let job_id: Option<String> = match fetch_conn
                    .blmove(
                        &queue_key,
                        &inflight_key,
                        redis::Direction::Left,
                        redis::Direction::Right,
                        BLOCK_TIMEOUT_SECS,
                    )
                    .await
                {
                    Ok(v) => v,
                    Err(_) => {
                        // Connection hiccup: back off briefly and reconnect.
                        sleep(Duration::from_secs(1)).await;
                        if let Ok(c) = get_dedicated_connection().await {
                            fetch_conn = c;
                        }
                        continue;
                    }
                };

                let Some(job_id) = job_id else { continue };

                let mut conn = match get_redis_connection().await {
                    Ok(c) => c,
                    Err(_) => {
                        sleep(Duration::from_secs(1)).await;
                        continue;
                    }
                };

                {
                    // Recompute the date each job so a long-lived worker rolls over to
                    // the new day's stats keys instead of writing to its start date forever.
                    let today = Utc::now().date_naive().format("%Y-%m-%d").to_string();
                    let job_key = format!("snm:job:{}", job_id);
                    let job_payload: String = conn.hget(&job_key, "payload").await.unwrap_or_default();
                    let job_name: String = conn.hget(&job_key, "job_name").await.unwrap_or_default();

                    let mut handled = false;
                    let mut error_message: Option<String> = None;

                    // Dispatch to the specific handler registered under this job's name,
                    // instead of guessing by trying every registered handler.
                    match get_job_handler(&job_name) {
                        Some(handler) => match handler(job_payload.clone()).await {
                            Ok(job) => {
                                if job.before().await.is_err() {
                                    // Skipped is a terminal, non-failure state.
                                    let _: () = conn.hset_multiple(&job_key, &[
                                        ("status", "skipped"),
                                        ("skipped_at", &Utc::now().to_rfc3339()),
                                    ]).await.unwrap_or_default();
                                } else {
                                    match job.perform().await {
                                        Ok(_) => {
                                            let _ = job.after().await;
                                            let _: () = conn.hset_multiple(&job_key, &[
                                                ("status", "success"),
                                                ("completed_at", &Utc::now().to_rfc3339()),
                                            ]).await.unwrap_or_default();
                                            let _: () = conn.incr("snm:qrush:success", 1).await.unwrap_or_default();
                                            // Track success jobs
                                            let _: () = conn.rpush(format!("snm:success:{}", queue), &job_id).await.unwrap_or_default();

                                            let key = format!("snm:stats:jobs:{}", today);
                                            // increment daily success counter
                                            let _: () = conn.incr(&key, 1).await.unwrap_or_default();
                                            let _: () = conn.incr("snm:qrush:total_jobs", 1).await.unwrap_or_default();
                                        }
                                        Err(err) => {
                                            let _ = job.on_error(&err).await;
                                            let retries: i64 = conn.hincr(&job_key, "retries", 1).await.unwrap_or(1);
                                            let _: () = conn.hset(&job_key, "error", err.to_string()).await.unwrap_or_default();

                                            if retries <= MAX_RETRIES as i64 {
                                                // Exponential backoff (10s * 2^retries) plus jitter, to avoid
                                                // thundering-herd retries. Jitter is derived from the current
                                                // sub-second clock, so no extra RNG dependency is needed.
                                                let base = 10_i64 * (1_i64 << retries.min(16));
                                                let jitter = (Utc::now().timestamp_subsec_nanos() as i64) % base.max(1);
                                                let run_at = Utc::now().timestamp() + base + jitter;
                                                let _: () = conn.hset_multiple(&job_key, &[
                                                    ("status", "retrying".to_string()),
                                                    ("run_at", run_at.to_string()),
                                                ]).await.unwrap_or_default();
                                                let _: () = conn.rpush(format!("snm:retry:{}", queue), &job_id).await.unwrap_or_default();
                                                let _: () = conn.zadd(DELAYED_JOBS_KEY, &job_id, run_at).await.unwrap_or_default();
                                            } else {
                                                // Retries exhausted: move to the dead-letter queue instead of
                                                // silently dropping the job.
                                                let _: () = conn.hset_multiple(&job_key, &[
                                                    ("status", "dead".to_string()),
                                                    ("dead_at", Utc::now().to_rfc3339()),
                                                ]).await.unwrap_or_default();
                                                let _: () = conn.rpush("snm:dead_jobs", &job_id).await.unwrap_or_default();
                                                let _: () = conn.rpush(format!("snm:dead:{}", queue), &job_id).await.unwrap_or_default();
                                                let _: () = conn.incr("snm:qrush:dead", 1).await.unwrap_or_default();
                                            }
                                        }
                                    }
                                    let _ = job.always().await;
                                }
                                handled = true;
                            }
                            Err(e) => {
                                error_message = Some(e.to_string());
                            }
                        },
                        None => {
                            error_message = Some(format!("No handler registered for job '{}'", job_name));
                        }
                    }

                    if !handled {
                        let mut hset_data = vec![
                            ("status", "failed".to_string()),
                            ("failed_at", Utc::now().to_rfc3339()),
                            ("queue", queue.clone()),
                            ("failed_at", Utc::now().to_rfc3339()),
                        ];

                        if let Some(ref emsg) = error_message {
                            hset_data.push(("error", emsg.clone()));
                        }

                        let fail_key = format!("snm:stats:jobs:{}:failed", today);
                        // increment daily failed counter
                        let _: () = conn.incr(&fail_key, 1).await.unwrap_or_default();


                        let _: () = conn.hset_multiple(&job_key, &hset_data).await.unwrap_or_default();
                        
                        let _: Result<(), _> = conn.lpush(
                            format!("snm:logs:{}", queue),
                            format!("[{}] ❌ Job {} failed", Utc::now(), job_id),
                        ).await;
                        let _: Result<(), _> = conn.ltrim(format!("snm:logs:{}", queue), 0, 99).await;
                        let _: () = conn.rpush("snm:failed_jobs", &job_id).await.unwrap_or_default();
                        // Track failed jobs
                        let _: () = conn.rpush(format!("snm:failed:{}", queue), &job_id).await.unwrap_or_default();
                        let _: () = conn.incr("snm:qrush:failed", 1).await.unwrap_or_default();
                    }

                    // The job has reached a terminal state (success / skipped / failed) or
                    // has been scheduled for retry, so it is no longer in flight.
                    let _: () = conn.lrem(&inflight_key, 1, &job_id).await.unwrap_or_default();
                }
            }
        });
    }
}


pub async fn start_delayed_worker_pool() {
    tokio::spawn(async move {
        loop {
            if is_shutting_down() {
                break;
            }
            let now = chrono::Utc::now().timestamp();
            let mut conn = match get_redis_connection().await {
                Ok(c) => c,
                Err(_) => {
                    tokio::time::sleep(Duration::from_secs(1)).await;
                    continue;
                }
            };

            let jobs: Vec<String> = conn.zrangebyscore(DELAYED_JOBS_KEY, 0, now).await.unwrap_or_default();
            for job_id in jobs {
                // Atomic claim: ZREM removes the member exactly once, so only the worker
                // whose ZREM returns 1 enqueues it — no duplicates across processes.
                let claimed: i64 = conn.zrem(DELAYED_JOBS_KEY, &job_id).await.unwrap_or(0);
                if claimed == 0 {
                    continue;
                }

                // Route back to the job's original queue instead of hardcoding "default".
                let job_key = format!("snm:job:{}", job_id);
                let queue: String = conn.hget(&job_key, "queue").await.unwrap_or_default();
                let queue = if queue.is_empty() { "default".to_string() } else { queue };

                let _: () = conn.hset(&job_key, "status", "pending").await.unwrap_or_default();
                let _: () = conn.rpush(format!("snm:queue:{}", queue), &job_id).await.unwrap_or_default();
            }
            tokio::time::sleep(Duration::from_secs(5)).await;
        }
    });
}


