hammerwork 2.5.0

//! Comprehensive tests for job result storage and retrieval functionality.

mod test_utils;

use hammerwork::{
    Job,
    job::{ResultConfig, ResultStorage},
    worker::JobResult,
};
use serde_json::json;
use std::time::Duration;

#[cfg(any(feature = "postgres", feature = "mysql"))]
use hammerwork::{
    JobStatus, Worker, WorkerPool, queue::DatabaseQueue, worker::JobHandlerWithResult,
};
#[cfg(any(feature = "postgres", feature = "mysql"))]
use std::sync::Arc;

/// Poll until the job reaches `Completed`, returning whether it did within `timeout`.
#[cfg(any(feature = "postgres", feature = "mysql"))]
async fn wait_for_completion<Q: DatabaseQueue>(
    queue: &Q,
    job_id: hammerwork::JobId,
    timeout: Duration,
) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;
    while tokio::time::Instant::now() < deadline {
        if let Some(job) = queue.get_job(job_id).await.unwrap()
            && job.status == JobStatus::Completed
        {
            return true;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
    false
}

/// Store a result on a dequeued job, then complete it: the result must still be
/// retrievable afterwards.
#[cfg(any(feature = "postgres", feature = "mysql"))]
async fn assert_result_survives_job_completion<Q: DatabaseQueue>(
    queue: &Q,
    cleanup: fn(&str) -> test_utils::QueueCleanup,
) {
    let queue_name = format!("completion_test_{}", uuid::Uuid::new_v4());
    let _cleanup = cleanup(&queue_name);
    let job = Job::new(queue_name.clone(), json!({"test": "completion"}))
        .with_result_storage(ResultStorage::Database);
    let job_id = queue.enqueue(job).await.unwrap();

    let dequeued = queue.dequeue(&queue_name).await.unwrap().unwrap();
    assert_eq!(dequeued.id, job_id);
    assert_eq!(dequeued.result_config.storage, ResultStorage::Database);

    let result_data = json!({"completed": true});
    queue
        .store_job_result(job_id, result_data.clone(), None)
        .await
        .unwrap();
    queue.complete_job(job_id).await.unwrap();

    let job = queue.get_job(job_id).await.unwrap().unwrap();
    assert_eq!(job.status, JobStatus::Completed);
    assert_eq!(
        queue.get_job_result(job_id).await.unwrap(),
        Some(result_data)
    );
}

/// A worker stores a `Memory` job's result in the queue's in-memory store: the queue
/// reads it back, it expires at its TTL, it can be deleted, and it is never written to
/// the database (a queue with its own store, like another process, doesn't see it).
#[cfg(any(feature = "postgres", feature = "mysql"))]
async fn assert_worker_stores_memory_result<DB>(
    queue: Arc<hammerwork::JobQueue<DB>>,
    cleanup: fn(&str) -> test_utils::QueueCleanup,
) where
    DB: sqlx::Database + Send + Sync + 'static,
    hammerwork::JobQueue<DB>: DatabaseQueue<Database = DB> + Send + Sync + 'static,
{
    let queue_name = test_utils::unique_queue("memory_result");
    let _cleanup = cleanup(&queue_name);
    let handler: JobHandlerWithResult = Arc::new(|job| {
        Box::pin(async move { Ok(JobResult::with_data(json!({ "echo": job.payload["n"] }))) })
    });

    let ttl = Duration::from_secs(2);
    let expiring = Job::new(queue_name.clone(), json!({"n": 1}))
        .with_result_storage(ResultStorage::Memory)
        .with_result_ttl(ttl);
    let lasting =
        Job::new(queue_name.clone(), json!({"n": 2})).with_result_storage(ResultStorage::Memory);
    let expiring_id = queue.enqueue(expiring).await.unwrap();
    let lasting_id = queue.enqueue(lasting).await.unwrap();

    let worker = Worker::new_with_result_handler(queue.clone(), queue_name.clone(), handler)
        .with_poll_interval(Duration::from_millis(10));
    let mut pool = WorkerPool::new();
    pool.add_worker(worker);
    // The results are stored after this, so they expire no earlier than `started + ttl`.
    let started = tokio::time::Instant::now();
    let pool_task = tokio::spawn(async move { pool.start().await });

    for id in [expiring_id, lasting_id] {
        assert!(
            wait_for_completion(queue.as_ref(), id, Duration::from_secs(20)).await,
            "job {id} did not complete"
        );
    }
    // Measured after completion, so the results expire no later than `completed_at + ttl`.
    let completed_at = tokio::time::Instant::now();
    pool_task.abort();

    assert_eq!(
        queue.get_job_result(lasting_id).await.unwrap(),
        Some(json!({"echo": 2}))
    );
    if started.elapsed() < ttl {
        assert_eq!(
            queue.get_job_result(expiring_id).await.unwrap(),
            Some(json!({"echo": 1}))
        );
    }

    // Nothing was written to the result columns.
    for id in [expiring_id, lasting_id] {
        let job = queue.get_job(id).await.unwrap().unwrap();
        assert_eq!(job.status, JobStatus::Completed);
        assert_eq!(
            job.result_data, None,
            "job {id} result written to the database"
        );
        assert_eq!(job.result_stored_at, None);
    }
    let other_process = hammerwork::JobQueue::<DB>::new(queue.get_pool().clone());
    assert_eq!(
        other_process.get_job_result(lasting_id).await.unwrap(),
        None
    );

    // The result with a TTL expires; the one without stays until it is deleted.
    tokio::time::sleep_until(completed_at + ttl + Duration::from_millis(200)).await;
    assert_eq!(queue.get_job_result(expiring_id).await.unwrap(), None);
    assert!(queue.get_job_result(lasting_id).await.unwrap().is_some());
    queue.delete_job_result(lasting_id).await.unwrap();
    assert_eq!(queue.get_job_result(lasting_id).await.unwrap(), None);

    for id in [expiring_id, lasting_id] {
        queue.delete_job(id).await.unwrap();
    }
}

#[cfg(feature = "postgres")]
mod postgres_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_postgres_result_survives_job_completion() {
        let queue = test_utils::setup_postgres_queue().await;
        assert_result_survives_job_completion(queue.as_ref(), test_utils::QueueCleanup::postgres)
            .await;
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_postgres_worker_stores_memory_result() {
        assert_worker_stores_memory_result(
            test_utils::setup_postgres_queue().await,
            test_utils::QueueCleanup::postgres,
        )
        .await;
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_job_result_storage_and_retrieval() {
        let queue = test_utils::setup_postgres_queue().await;

        let queue_name = test_utils::unique_queue("result_storage");
        let _cleanup = test_utils::QueueCleanup::postgres(&queue_name);

        // Create a job with result storage enabled
        let job = Job::new(queue_name, json!({"task": "process_data"}))
            .with_result_storage(ResultStorage::Database)
            .with_result_ttl(Duration::from_secs(3600)); // 1 hour TTL

        let job_id = queue.enqueue(job).await.unwrap();

        // Store some result data
        let result_data = json!({
            "processed_items": 42,
            "total_time_ms": 1500,
            "status": "completed",
            "output_file": "/tmp/result.json"
        });

        queue
            .store_job_result(job_id, result_data.clone(), None)
            .await
            .unwrap();

        // Retrieve the result
        let retrieved_result = queue.get_job_result(job_id).await.unwrap();
        assert!(retrieved_result.is_some());
        assert_eq!(retrieved_result.unwrap(), result_data);

        // Test deletion
        queue.delete_job_result(job_id).await.unwrap();
        let deleted_result = queue.get_job_result(job_id).await.unwrap();
        assert!(deleted_result.is_none());
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_result_expiration() {
        let queue = test_utils::setup_postgres_queue().await;

        let queue_name = test_utils::unique_queue("result_expiration");
        let _cleanup = test_utils::QueueCleanup::postgres(&queue_name);
        let job = Job::new(queue_name, json!({"task": "short_lived"}));
        let job_id = queue.enqueue(job).await.unwrap();

        let result_data = json!({"status": "completed"});
        let expires_at = chrono::Utc::now() + chrono::Duration::seconds(1);

        queue
            .store_job_result(job_id, result_data, Some(expires_at))
            .await
            .unwrap();

        // Should be retrievable immediately
        let result = queue.get_job_result(job_id).await.unwrap();
        assert!(result.is_some());

        // Wait for expiration
        tokio::time::sleep(Duration::from_secs(2)).await;

        // Should be expired now
        let expired_result = queue.get_job_result(job_id).await.unwrap();
        assert!(expired_result.is_none());
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_cleanup_expired_results() {
        let queue = test_utils::setup_postgres_queue().await;

        // Clear any existing expired results to ensure clean test state
        let _ = queue.cleanup_expired_results().await;

        // Create multiple jobs with different expiration times
        let mut job_ids = Vec::new();
        let test_queue = test_utils::unique_queue("test_cleanup");
        let _cleanup = test_utils::QueueCleanup::postgres(&test_queue);
        for i in 0..5 {
            let job = Job::new(test_queue.clone(), json!({"task": i}));
            let job_id = queue.enqueue(job).await.unwrap();
            job_ids.push(job_id);

            let result_data = json!({"task_id": i});
            let expires_at = if i < 3 {
                // First 3 expire immediately
                Some(chrono::Utc::now() - chrono::Duration::seconds(1))
            } else {
                // Last 2 expire in the future
                Some(chrono::Utc::now() + chrono::Duration::hours(1))
            };

            queue
                .store_job_result(job_id, result_data, expires_at)
                .await
                .unwrap();
        }

        // Clean up expired results. Worker pools in other tests clean expired results on
        // their own schedule too, so the count returned here is not ours alone; check the
        // stored rows instead. (Expired results are hidden by `get_job_result` either way,
        // so only the row shows whether cleanup removed them.)
        queue.cleanup_expired_results().await.unwrap();

        for (i, job_id) in job_ids.iter().enumerate() {
            let stored = queue.get_job(*job_id).await.unwrap().unwrap().result_data;
            let result = queue.get_job_result(*job_id).await.unwrap();
            if i < 3 {
                assert!(
                    stored.is_none(),
                    "job {i}: the expired result is still stored"
                );
                assert!(
                    result.is_none(),
                    "job {i}: the expired result is still readable"
                );
            } else {
                assert!(stored.is_some(), "job {i}: an unexpired result was removed");
                assert!(
                    result.is_some(),
                    "job {i}: an unexpired result is not readable"
                );
            }
        }
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_worker_automatic_result_storage() {
        let queue = test_utils::setup_postgres_queue().await;
        let unique_queue = format!("auto_result_test_{}", uuid::Uuid::new_v4());
        let _cleanup = test_utils::QueueCleanup::postgres(&unique_queue);

        // Create a handler that returns result data
        let handler: JobHandlerWithResult = Arc::new(|job| {
            Box::pin(async move {
                let task_id: u64 = job.payload["task_id"].as_u64().unwrap();
                let result_data = json!({
                    "task_id": task_id,
                    "processed_at": chrono::Utc::now(),
                    "result": format!("Processed task {}", task_id)
                });

                Ok(JobResult::with_data(result_data))
            })
        });

        // Create worker with result handler
        let worker = Worker::new_with_result_handler(queue.clone(), unique_queue.clone(), handler);

        // Create a job with result storage enabled
        let job = Job::new(unique_queue, json!({"task_id": 123}))
            .with_result_storage(ResultStorage::Database)
            .with_result_ttl(Duration::from_secs(3600));

        let job_id = queue.enqueue(job).await.unwrap();

        // Create and start worker pool to process the job
        let mut worker_pool = WorkerPool::new();
        worker_pool.add_worker(worker);

        // Start the worker pool for a short time to process the job
        let worker_handle = tokio::spawn(async move { worker_pool.start().await });

        // Wait for the job to be processed
        assert!(
            wait_for_completion(queue.as_ref(), job_id, Duration::from_secs(20)).await,
            "Job should have completed within timeout"
        );

        // Verify result was automatically stored
        let stored_result = queue.get_job_result(job_id).await.unwrap();
        assert!(
            stored_result.is_some(),
            "Job result should be stored after completion"
        );

        let result = stored_result.unwrap();
        assert_eq!(result["task_id"], 123);
        assert!(result["result"].as_str().unwrap().contains("123"));

        // Stop the worker pool
        worker_handle.abort();
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_worker_legacy_handler_compatibility() {
        let queue = test_utils::setup_postgres_queue().await;
        let unique_queue = format!("legacy_test_{}", uuid::Uuid::new_v4());
        let _cleanup = test_utils::QueueCleanup::postgres(&unique_queue);

        // Create a legacy handler (returns ())
        let legacy_handler: hammerwork::worker::JobHandler = Arc::new(|_job| {
            Box::pin(async move {
                // Just complete successfully without result data
                Ok(())
            })
        });

        // Create worker with legacy handler
        let worker = Worker::new(queue.clone(), unique_queue.clone(), legacy_handler);

        // Create a job (even with result storage enabled)
        let job = Job::new(unique_queue, json!({"task": "legacy"}))
            .with_result_storage(ResultStorage::Database);

        let job_id = queue.enqueue(job).await.unwrap();

        // Process the job using worker pool
        let mut worker_pool = WorkerPool::new();
        worker_pool.add_worker(worker);

        let worker_handle = tokio::spawn(async move { worker_pool.start().await });

        // Wait for the job to be processed
        assert!(
            wait_for_completion(queue.as_ref(), job_id, Duration::from_secs(10)).await,
            "Job should have completed within timeout"
        );

        // Verify no result was stored (legacy handler returns no data)
        let stored_result = queue.get_job_result(job_id).await.unwrap();
        assert!(
            stored_result.is_none(),
            "Legacy handler should not store result data"
        );

        worker_handle.abort();
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_result_storage_none_configuration() {
        let queue = test_utils::setup_postgres_queue().await;

        // Create a handler that returns result data
        let handler: JobHandlerWithResult = Arc::new(|_job| {
            Box::pin(async move {
                let result_data = json!({"status": "completed"});
                Ok(JobResult::with_data(result_data))
            })
        });

        let unique_queue = format!("no_storage_test_{}", uuid::Uuid::new_v4());
        let _cleanup = test_utils::QueueCleanup::postgres(&unique_queue);
        let worker = Worker::new_with_result_handler(queue.clone(), unique_queue.clone(), handler);

        // Create a job with result storage disabled
        let job = Job::new(unique_queue, json!({"task": "no_storage"}))
            .with_result_storage(ResultStorage::None);

        let job_id = queue.enqueue(job).await.unwrap();

        // Process the job using worker pool
        let mut worker_pool = WorkerPool::new();
        worker_pool.add_worker(worker);

        let worker_handle = tokio::spawn(async move { worker_pool.start().await });

        // Wait for job processing
        assert!(
            wait_for_completion(queue.as_ref(), job_id, Duration::from_secs(10)).await,
            "Job should have completed within timeout"
        );

        // Verify no result was stored despite handler returning data
        let stored_result = queue.get_job_result(job_id).await.unwrap();
        assert!(stored_result.is_none());

        worker_handle.abort();
    }
}

#[cfg(feature = "mysql")]
mod mysql_tests {
    use super::*;

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_mysql_result_survives_job_completion() {
        let queue = test_utils::setup_mysql_queue().await;
        assert_result_survives_job_completion(queue.as_ref(), test_utils::QueueCleanup::mysql)
            .await;
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_mysql_worker_stores_memory_result() {
        assert_worker_stores_memory_result(
            test_utils::setup_mysql_queue().await,
            test_utils::QueueCleanup::mysql,
        )
        .await;
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_mysql_result_storage() {
        let queue = test_utils::setup_mysql_queue().await;

        let queue_name = test_utils::unique_queue("mysql_result_storage");
        let _cleanup = test_utils::QueueCleanup::mysql(&queue_name);

        // Test basic result storage with MySQL
        let job = Job::new(queue_name, json!({"task": "mysql_test"}))
            .with_result_storage(ResultStorage::Database);

        let job_id = queue.enqueue(job).await.unwrap();

        let result_data = json!({
            "database": "mysql",
            "processed_items": 100,
            "status": "success"
        });

        queue
            .store_job_result(job_id, result_data.clone(), None)
            .await
            .unwrap();

        let retrieved_result = queue.get_job_result(job_id).await.unwrap();
        assert!(retrieved_result.is_some());
        assert_eq!(retrieved_result.unwrap(), result_data);
    }

    #[tokio::test]
    #[ignore] // Requires database connection
    async fn test_mysql_worker_integration() {
        let queue = test_utils::setup_mysql_queue().await;

        // Clear any existing test data
        let _ = queue.cleanup_expired_results().await;

        let handler: JobHandlerWithResult = Arc::new(|job| {
            Box::pin(async move {
                let result_data = json!({
                    "mysql_worker": true,
                    "payload": job.payload
                });
                Ok(JobResult::with_data(result_data))
            })
        });

        let test_queue = test_utils::unique_queue("mysql_worker");
        let _cleanup = test_utils::QueueCleanup::mysql(&test_queue);
        let worker = Worker::new_with_result_handler(queue.clone(), test_queue.clone(), handler);

        let job = Job::new(test_queue.clone(), json!({"test": "mysql"}))
            .with_result_storage(ResultStorage::Database);

        let job_id = queue.enqueue(job).await.unwrap();

        // Process job using worker pool
        let mut worker_pool = WorkerPool::new();
        worker_pool.add_worker(worker);

        let worker_handle = tokio::spawn(async move { worker_pool.start().await });

        // Wait for job processing
        assert!(
            wait_for_completion(queue.as_ref(), job_id, Duration::from_secs(10)).await,
            "Job should have completed within timeout"
        );

        // Verify result storage
        let result = queue.get_job_result(job_id).await.unwrap();
        assert!(result.is_some());
        assert_eq!(result.unwrap()["mysql_worker"], true);

        worker_handle.abort();
    }
}

// Tests that work regardless of database backend
#[tokio::test]
async fn test_result_config_builder_methods() {
    // Test the job builder methods for result configuration
    let job1 = Job::new("test".to_string(), json!({})).with_result_storage(ResultStorage::Database);
    assert_eq!(job1.result_config.storage, ResultStorage::Database);

    let job2 = Job::new("test".to_string(), json!({})).with_result_ttl(Duration::from_secs(7200));
    assert_eq!(job2.result_config.ttl, Some(Duration::from_secs(7200)));

    let config = ResultConfig {
        storage: ResultStorage::Database,
        ttl: Some(Duration::from_secs(3600)),
        max_size_bytes: Some(1024 * 1024), // 1MB
    };
    let job3 = Job::new("test".to_string(), json!({})).with_result_config(config.clone());
    assert_eq!(job3.result_config.storage, config.storage);
    assert_eq!(job3.result_config.ttl, config.ttl);
    assert_eq!(job3.result_config.max_size_bytes, config.max_size_bytes);
}

#[tokio::test]
async fn test_result_types() {
    // Test JobResult creation methods
    let success_result = JobResult::success();
    assert!(success_result.data.is_none());

    let data_result = JobResult::with_data(json!({"key": "value"}));
    assert!(data_result.data.is_some());
    assert_eq!(data_result.data.unwrap()["key"], "value");

    // Test default
    let default_result = JobResult::default();
    assert!(default_result.data.is_none());
}

#[tokio::test]
async fn test_result_config_defaults() {
    let config = ResultConfig::default();
    assert_eq!(config.storage, ResultStorage::None);
    assert!(config.ttl.is_none());
    assert!(config.max_size_bytes.is_none());
}