qrush 3.0.1

//! Regression test for cron job execution wiring.
//!
//! A cron job is stored type-erased (only its serialized payload survives), so
//! when it fires the scheduler rebuilds the queue entry by hand. That entry MUST
//! carry a `job_name` field — the worker dispatches to a handler by it, and a
//! missing name makes every cron fire fail with "No handler registered for ''".
//! This test drives `run_now` (same enqueue path as a scheduled fire) and asserts
//! the resulting job hash is dispatchable.
//!
//! Requires a reachable Redis; set `REDIS_URL` or run one on localhost. When no
//! Redis is available the test soft-skips so it never breaks a Redis-free CI run.
//!
//! Run with: `cargo test --test cron_enqueue`

use std::collections::HashMap;

use async_trait::async_trait;
use redis::AsyncCommands;
use serde::{Deserialize, Serialize};

use qrush::config::set_redis_url;
use qrush::cron::cron_job::CronJob;
use qrush::cron::cron_scheduler::CronScheduler;
use qrush::job::Job;
use qrush::utils::rdconfig::get_redis_connection;

#[derive(Clone, Serialize, Deserialize)]
struct CronTestJob {
    message: String,
}

#[async_trait]
impl Job for CronTestJob {
    async fn perform(&self) -> anyhow::Result<()> {
        Ok(())
    }
    fn name(&self) -> &'static str {
        "CronTestJob"
    }
    fn queue(&self) -> &'static str {
        "cron_test_queue"
    }
}

#[async_trait]
impl CronJob for CronTestJob {
    fn cron_expression(&self) -> &'static str {
        "0 0 * * * *"
    }
    fn cron_id(&self) -> &'static str {
        "cron_enqueue_test_job"
    }
}

#[tokio::test]
async fn run_now_enqueues_a_dispatchable_job() {
    let redis_url =
        std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
    set_redis_url(redis_url).expect("failed to set Redis URL");

    // Soft-skip when Redis is unreachable so a Redis-free CI run stays green.
    let mut conn = match get_redis_connection().await {
        Ok(c) => c,
        Err(e) => {
            eprintln!("skipping cron_enqueue test: Redis unavailable ({e})");
            return;
        }
    };

    let cron_id = CronTestJob {
        message: String::new(),
    }
    .cron_id();

    // Start from a clean slate in case a prior run left the schedule behind.
    CronScheduler::delete_cron_job(cron_id)
        .await
        .expect("cleanup delete failed");

    let job = CronTestJob {
        message: "hello from cron".into(),
    };
    CronScheduler::register_cron_job(job)
        .await
        .expect("register_cron_job failed");

    // `run_now` shares the enqueue path with a scheduled fire.
    let enqueued_id = CronScheduler::run_now(cron_id)
        .await
        .expect("run_now failed");

    // The enqueued job hash must be dispatchable by the worker.
    let job_key = format!("snm:job:{enqueued_id}");
    let job_hash: HashMap<String, String> =
        conn.hgetall(&job_key).await.expect("hgetall failed");

    assert_eq!(
        job_hash.get("job_name").map(String::as_str),
        Some("CronTestJob"),
        "enqueued cron job is missing the job_name the worker dispatches on"
    );
    assert_eq!(
        job_hash.get("queue").map(String::as_str),
        Some("cron_test_queue"),
        "enqueued cron job landed on the wrong queue"
    );
    // Payload is the serialized job, so a registered handler can rebuild it.
    let payload = job_hash.get("payload").expect("missing payload");
    let rebuilt: CronTestJob =
        serde_json::from_str(payload).expect("payload is not a valid CronTestJob");
    assert_eq!(rebuilt.message, "hello from cron");

    // Cleanup: the schedule entry and the enqueued job hash.
    CronScheduler::delete_cron_job(cron_id)
        .await
        .expect("cleanup delete failed");
    let _: () = conn.del(&job_key).await.unwrap_or_default();
}