Background Tasks¶

use spikard::prelude::*;
use tokio::task;

app.route(post("/send-email"), |ctx: Context| async move {
    let email = ctx.body_as::<EmailRequest>().await?;

    // Spawn background task without waiting for completion
    task::spawn(async move {
        if let Err(e) = send_email_internal(&email).await {
            eprintln!("Email send failed: {}", e);
        }
    });

    Ok(Json(json!({
        "status": "queued",
        "message": "Email will be sent shortly"
    })))
})?;

async fn send_email_internal(email: &EmailRequest) -> Result<(), Box<dyn std::error::Error>> {
    // Simulate async email sending
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    println!("Email sent to: {}", email.to);
    Ok(())
}

#[derive(serde::Deserialize)]
struct EmailRequest {
    to: String,
    subject: String,
}

use spikard::prelude::*;
use tokio::time::{sleep, Duration};

async fn retry_with_backoff<F, T>(mut f: F, max_retries: u32) -> Result<T, Box<dyn std::error::Error>>
where
    F: FnMut() -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, Box<dyn std::error::Error>>>>>,
{
    let mut retries = 0;
    loop {
        match f().await {
            Ok(result) => return Ok(result),
            Err(e) => {
                retries += 1;
                if retries >= max_retries {
                    return Err(e);
                }
                // Exponential backoff: 1s, 2s, 4s, 8s...
                let backoff = Duration::from_secs(2_u64.pow(retries - 1));
                eprintln!("Retry attempt {} after {:?}: {}", retries, backoff, e);
                sleep(backoff).await;
            }
        }
    }
}

app.route(post("/process"), |ctx: Context| async move {
    let data = ctx.body_as::<ProcessRequest>().await?;

    tokio::spawn(async move {
        if let Err(e) = retry_with_backoff(
            || {
                Box::pin(async {
                    process_data_internal(&data).await
                })
            },
            3,
        )
        .await
        {
            eprintln!("Processing failed after retries: {}", e);
        }
    });

    Ok(Json(json!({"status": "processing"})))
})?;

async fn process_data_internal(data: &ProcessRequest) -> Result<(), Box<dyn std::error::Error>> {
    // Simulate processing that may fail
    Ok(())
}

#[derive(serde::Deserialize)]
struct ProcessRequest {
    id: String,
}

use spikard::prelude::*;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;

struct TaskMonitor {
    active_tasks: Arc<AtomicU32>,
    completed_tasks: Arc<AtomicU32>,
}

impl TaskMonitor {
    fn new() -> Self {
        Self {
            active_tasks: Arc::new(AtomicU32::new(0)),
            completed_tasks: Arc::new(AtomicU32::new(0)),
        }
    }

    fn record_start(&self) {
        self.active_tasks.fetch_add(1, Ordering::SeqCst);
    }

    fn record_complete(&self) {
        self.active_tasks.fetch_sub(1, Ordering::SeqCst);
        self.completed_tasks.fetch_add(1, Ordering::SeqCst);
    }
}

let monitor = Arc::new(TaskMonitor::new());

app.route(get("/task-stats"), {
    let monitor = Arc::clone(&monitor);
    |_ctx: Context| async move {
        Ok(Json(json!({
            "active_tasks": monitor.active_tasks.load(Ordering::SeqCst),
            "completed_tasks": monitor.completed_tasks.load(Ordering::SeqCst),
        })))
    }
})?;

app.route(post("/process"), {
    let monitor = Arc::clone(&monitor);
    |ctx: Context| async move {
        let data = ctx.body_as::<ProcessRequest>().await?;

        let monitor_clone = Arc::clone(&monitor);
        tokio::spawn(async move {
            monitor_clone.record_start();
            if let Err(e) = process_data(&data).await {
                eprintln!("Processing failed: {}", e);
            }
            monitor_clone.record_complete();
        });

        Ok(Json(json!({"status": "processing"})))
    }
})?;

async fn process_data(data: &ProcessRequest) -> Result<(), Box<dyn std::error::Error>> {
    Ok(())
}

#[derive(serde::Deserialize)]
struct ProcessRequest {
    id: String,
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::time::{sleep, timeout, Duration};

    #[tokio::test]
    async fn test_background_task_completion() {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<String>(1);

        tokio::spawn(async move {
            sleep(Duration::from_millis(100)).await;
            let _ = tx.send("task_complete".to_string()).await;
        });

        let result = timeout(Duration::from_secs(1), rx.recv())
            .await
            .expect("timeout")
            .expect("channel closed");

        assert_eq!(result, "task_complete");
    }

    #[tokio::test]
    async fn test_multiple_background_tasks() {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<u32>(10);

        for i in 0..5 {
            let tx_clone = tx.clone();
            tokio::spawn(async move {
                sleep(Duration::from_millis(50 * i)).await;
                let _ = tx_clone.send(i).await;
            });
        }
        drop(tx); // Close sender to allow recv to complete

        let mut results = Vec::new();
        while let Ok(value) = timeout(Duration::from_secs(1), rx.recv()).await {
            if let Some(v) = value {
                results.push(v);
            } else {
                break;
            }
        }

        assert_eq!(results.len(), 5);
    }

    #[tokio::test]
    async fn test_task_error_handling() {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<Result<String, String>>(1);

        tokio::spawn(async move {
            sleep(Duration::from_millis(50)).await;
            let _ = tx.send(Err("task_failed".to_string())).await;
        });

        let result = timeout(Duration::from_secs(1), rx.recv())
            .await
            .expect("timeout")
            .expect("channel closed");

        assert!(result.is_err());
        assert_eq!(result.err().unwrap(), "task_failed");
    }
}