what_core/jobs/

mod.rs

//! Background job queue for non-blocking operations.
//!
//! A simple in-process async task queue built on Tokio channels.
//! Jobs run as Tokio tasks. Failed jobs log errors (no retry by default).
//!
//! The framework uses this internally — users don't interact with it directly.

use std::sync::Arc;
use tokio::sync::mpsc;

use crate::config::EmailConfig;
use crate::email::EmailMessage;
use crate::sessions::SessionBackend;

/// A background job to be executed asynchronously.
#[derive(Debug)]
pub enum Job {
    /// Purge expired sessions from the SQLite session store.
    SessionCleanup,

    /// Send an email in the background.
    SendEmail { message: EmailMessage },

    /// A custom closure job (for future extensibility).
    /// The string is a label for logging.
    Custom {
        label: String,
        #[allow(dead_code)]
        payload: serde_json::Value,
    },
}

/// Handle for enqueuing background jobs.
///
/// Cheap to clone — wraps an `mpsc::Sender`.
#[derive(Clone)]
pub struct JobQueue {
    tx: mpsc::Sender<Job>,
}

impl JobQueue {
    /// Enqueue a job for background execution.
    /// Returns `true` if the job was enqueued, `false` if the queue is full or closed.
    pub async fn enqueue(&self, job: Job) -> bool {
        match self.tx.try_send(job) {
            Ok(()) => true,
            Err(mpsc::error::TrySendError::Full(job)) => {
                tracing::warn!("Job queue full, dropping job: {:?}", job);
                false
            }
            Err(mpsc::error::TrySendError::Closed(job)) => {
                tracing::warn!("Job queue closed, dropping job: {:?}", job);
                false
            }
        }
    }
}

/// Context available to the job worker for executing jobs.
struct WorkerContext {
    sessions: Option<SessionBackend>,
    email_config: Option<EmailConfig>,
}

/// Start the background job system.
///
/// Returns a `JobQueue` handle for enqueuing jobs.
/// If called from within a Tokio runtime, also spawns the worker task and
/// periodic session cleanup timer. If no runtime is available (e.g. in sync tests),
/// creates the queue without spawning background tasks.
pub fn start(sessions: Option<SessionBackend>, email_config: Option<EmailConfig>) -> JobQueue {
    let (tx, rx) = mpsc::channel::<Job>(256);
    let queue = JobQueue { tx };

    // Only spawn background tasks if a Tokio runtime is available
    if let Ok(_handle) = tokio::runtime::Handle::try_current() {
        let ctx = Arc::new(WorkerContext {
            sessions,
            email_config,
        });

        // Spawn the worker that processes jobs from the channel
        let worker_ctx = Arc::clone(&ctx);
        tokio::spawn(async move {
            run_worker(rx, worker_ctx).await;
        });

        // Spawn periodic session cleanup (every hour)
        let cleanup_queue = queue.clone();
        tokio::spawn(async move {
            run_session_cleanup_timer(cleanup_queue).await;
        });

        tracing::info!("Background job queue started (capacity: 256)");
    } else {
        tracing::debug!("No Tokio runtime — job queue created without background worker");
    }

    queue
}

/// The main worker loop — receives jobs and executes them.
async fn run_worker(mut rx: mpsc::Receiver<Job>, ctx: Arc<WorkerContext>) {
    while let Some(job) = rx.recv().await {
        let ctx = Arc::clone(&ctx);
        // Spawn each job as its own task so one slow job doesn't block others
        tokio::spawn(async move {
            execute_job(job, &ctx).await;
        });
    }
    tracing::debug!("Job worker shutting down — channel closed");
}

/// Execute a single job.
async fn execute_job(job: Job, ctx: &WorkerContext) {
    match job {
        Job::SessionCleanup => {
            if let Some(ref sessions) = ctx.sessions {
                match sessions.cleanup_expired().await {
                    Ok(count) if count > 0 => {
                        tracing::info!("Session cleanup: purged {} expired sessions", count);
                    }
                    Ok(_) => {
                        tracing::debug!("Session cleanup: no expired sessions");
                    }
                    Err(e) => {
                        tracing::warn!("Session cleanup failed: {}", e);
                    }
                }
            }
        }
        Job::SendEmail { message } => {
            if let Some(ref email_config) = ctx.email_config {
                match crate::email::send_email(email_config, &message).await {
                    Ok(()) => {
                        tracing::info!("Email sent to {}", message.to);
                    }
                    Err(e) => {
                        tracing::error!("Failed to send email to {}: {}", message.to, e);
                    }
                }
            } else {
                tracing::warn!(
                    "SendEmail job received but no [email] config — dropping email to {}",
                    message.to
                );
            }
        }
        Job::Custom { label, .. } => {
            tracing::debug!("Custom job executed: {}", label);
        }
    }
}

/// Periodically enqueue session cleanup jobs (every hour).
async fn run_session_cleanup_timer(queue: JobQueue) {
    let mut interval = tokio::time::interval(std::time::Duration::from_secs(3600));
    // Skip the first immediate tick — startup cleanup already runs in AppState::with_dev_mode
    interval.tick().await;

    loop {
        interval.tick().await;
        let _ = queue.enqueue(Job::SessionCleanup).await;
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_job_queue_enqueue() {
        let (tx, mut rx) = mpsc::channel(16);
        let queue = JobQueue { tx };

        let ok = queue.enqueue(Job::SessionCleanup).await;
        assert!(ok);

        let job = rx.try_recv().unwrap();
        assert!(matches!(job, Job::SessionCleanup));
    }

    #[tokio::test]
    async fn test_job_queue_full() {
        let (tx, _rx) = mpsc::channel(1);
        let queue = JobQueue { tx };

        // Fill the channel
        assert!(queue.enqueue(Job::SessionCleanup).await);
        // Next enqueue should fail (channel full, no receiver draining)
        let ok = queue.enqueue(Job::SessionCleanup).await;
        assert!(!ok);
    }

    #[tokio::test]
    async fn test_job_queue_closed() {
        let (tx, rx) = mpsc::channel(16);
        let queue = JobQueue { tx };

        drop(rx); // Close the receiver
        let ok = queue.enqueue(Job::SessionCleanup).await;
        assert!(!ok);
    }

    #[tokio::test]
    async fn test_start_returns_queue() {
        let queue = start(None, None);

        // Can enqueue a job
        let ok = queue
            .enqueue(Job::Custom {
                label: "test".to_string(),
                payload: serde_json::Value::Null,
            })
            .await;
        assert!(ok);

        // Give the worker a moment to process
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }

    #[tokio::test]
    async fn test_session_cleanup_without_sessions() {
        // With no session backend, cleanup should be a no-op
        let ctx = Arc::new(WorkerContext {
            sessions: None,
            email_config: None,
        });
        execute_job(Job::SessionCleanup, &ctx).await;
        // Should not panic — just silently skip
    }

    #[tokio::test]
    async fn test_custom_job_execution() {
        let ctx = Arc::new(WorkerContext {
            sessions: None,
            email_config: None,
        });
        execute_job(
            Job::Custom {
                label: "test-job".to_string(),
                payload: serde_json::json!({"key": "value"}),
            },
            &ctx,
        )
        .await;
        // Should not panic
    }

    #[tokio::test]
    async fn test_send_email_job_without_config() {
        // SendEmail job with no email config should not panic — just logs a warning
        let ctx = Arc::new(WorkerContext {
            sessions: None,
            email_config: None,
        });
        execute_job(
            Job::SendEmail {
                message: EmailMessage {
                    to: "user@example.com".into(),
                    subject: "Test".into(),
                    html_body: "<p>Hi</p>".into(),
                    text_body: None,
                },
            },
            &ctx,
        )
        .await;
    }

    #[tokio::test]
    async fn test_send_email_job_enqueue() {
        let (tx, mut rx) = mpsc::channel(16);
        let queue = JobQueue { tx };

        let msg = EmailMessage {
            to: "user@example.com".into(),
            subject: "Welcome".into(),
            html_body: "<h1>Hi</h1>".into(),
            text_body: None,
        };
        let ok = queue.enqueue(Job::SendEmail { message: msg }).await;
        assert!(ok);

        let job = rx.try_recv().unwrap();
        assert!(matches!(job, Job::SendEmail { .. }));
    }

    #[test]
    fn test_job_debug_format() {
        let job = Job::SessionCleanup;
        let debug = format!("{:?}", job);
        assert!(debug.contains("SessionCleanup"));

        let job = Job::SendEmail {
            message: EmailMessage {
                to: "a@b.com".into(),
                subject: "Hi".into(),
                html_body: "<p>Hi</p>".into(),
                text_body: None,
            },
        };
        let debug = format!("{:?}", job);
        assert!(debug.contains("SendEmail"));

        let job = Job::Custom {
            label: "hello".to_string(),
            payload: serde_json::Value::Null,
        };
        let debug = format!("{:?}", job);
        assert!(debug.contains("Custom"));
        assert!(debug.contains("hello"));
    }

    #[test]
    fn test_job_queue_is_clone() {
        // Verify JobQueue implements Clone (compile-time check)
        fn assert_clone<T: Clone>() {}
        assert_clone::<JobQueue>();
    }
}