How to Add Background Jobs to an Actix Web App

[dependencies]
encore-dev = "0.1"
actix-web = "4"
serde = { version = "1", features = ["derive"] }

-- src/migrations/1_create_orders.up.sql
CREATE TABLE orders (
    id BIGSERIAL PRIMARY KEY,
    email TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE receipts_sent (
    order_id BIGINT PRIMARY KEY REFERENCES orders (id)
);

use std::time::Duration;

use actix_web::{post, web, App, HttpResponse, HttpServer, Responder};
use encore::cron::{Job, Schedule};
use encore::pubsub::{Message, Subscription, Topic};
use encore::storage::postgres::Database;
use serde::{Deserialize, Serialize};

static DB: Database = Database::new("orders").migrations("./migrations");

#[derive(Debug, Clone, Serialize, Deserialize)]
struct OrderPlaced {
    order_id: i64,
    email: String,
}

// The job queue: SNS/SQS on AWS, Pub/Sub on GCP, built in under encore run.
static ORDER_PLACED: Topic<OrderPlaced> = Topic::new("order-placed");

// The worker, started with the app by encore::initialize().
// A handler that returns an error is retried with backoff.
#[encore::pubsub::register]
static SEND_RECEIPT: Subscription<OrderPlaced> =
    Subscription::new(&ORDER_PLACED, "send-receipt", send_receipt)
        .min_backoff(Duration::from_secs(5))
        .max_retries(5);

// Delivery is at least once, so the handler checks receipts_sent first
// and a redelivered message does not send a second email.
async fn send_receipt(msg: Message<OrderPlaced>) -> Result<(), encore::Error> {
    let order = &msg.data;
    if DB
        .query_opt("SELECT 1 FROM receipts_sent WHERE order_id = $1", &[&order.order_id])
        .await?
        .is_some()
    {
        return Ok(());
    }
    send_email(&order.email, order.order_id).await?;
    DB.execute(
        "INSERT INTO receipts_sent (order_id) VALUES ($1) ON CONFLICT DO NOTHING",
        &[&order.order_id],
    )
    .await?;
    Ok(())
}

// Placeholder for your email provider.
async fn send_email(to: &str, order_id: i64) -> Result<(), encore::Error> {
    println!("receipt for order {order_id} sent to {to}");
    Ok(())
}

#[derive(Deserialize)]
struct NewOrder {
    email: String,
}

#[post("/orders")]
async fn place_order(body: web::Json<NewOrder>) -> impl Responder {
    let row = match DB
        .query_one("INSERT INTO orders (email) VALUES ($1) RETURNING id", &[&body.email])
        .await
    {
        Ok(row) => row,
        Err(_) => return HttpResponse::InternalServerError().finish(),
    };
    let order_id: i64 = row.get("id");
    let job = OrderPlaced { order_id, email: body.email.clone() };
    match ORDER_PLACED.publish(&job).await {
        Ok(_) => HttpResponse::Accepted().body(format!("order {order_id} placed\n")),
        Err(_) => HttpResponse::InternalServerError().finish(),
    }
}

// Cancels orders left pending for a day, every hour in deployed environments.
static EXPIRE_ORDERS: Job =
    Job::new("expire-orders", "/internal/expire-orders", Schedule::Every("1h"))
        .title("Expire stale pending orders");

async fn expire_orders() -> impl Responder {
    match DB
        .execute(
            "UPDATE orders SET status = 'expired' \
             WHERE status = 'pending' AND created_at < now() - interval '1 day'",
            &[],
        )
        .await
    {
        Ok(n) => HttpResponse::Ok().body(format!("expired {n} orders\n")),
        Err(_) => HttpResponse::InternalServerError().finish(),
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    encore::initialize();
    let addr = std::env::var("ENCORE_LISTEN_ADDR").unwrap_or("0.0.0.0:8080".into());
    HttpServer::new(|| {
        App::new()
            .service(place_order)
            .route(EXPIRE_ORDERS.path(), web::post().to(expire_orders))
    })
    .bind(addr)?
    .run()
    .await
}

use actix_web::{post, web, App, HttpResponse, HttpServer, Responder};
use apalis::prelude::*;
use apalis_redis::RedisStorage;
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
struct OrderPlaced {
    order_id: i64,
    email: String,
}

async fn send_receipt(job: OrderPlaced) -> Result<(), Error> {
    // Check a receipts table first: a retried job must not email twice.
    println!("receipt for order {} sent to {}", job.order_id, job.email);
    Ok(())
}

#[post("/orders")]
async fn place_order(storage: web::Data<RedisStorage<OrderPlaced>>) -> impl Responder {
    let mut storage = storage.get_ref().clone();
    let job = OrderPlaced { order_id: 1, email: "ada@example.com".into() };
    match storage.push(job).await {
        Ok(_) => HttpResponse::Accepted().finish(),
        Err(_) => HttpResponse::InternalServerError().finish(),
    }
}

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let conn = apalis_redis::connect("redis://127.0.0.1:6379").await.expect("redis");
    let storage: RedisStorage<OrderPlaced> = RedisStorage::new(conn);

    let worker = WorkerBuilder::new("send-receipt")
        .backend(storage.clone())
        .build_fn(send_receipt);
    let monitor = Monitor::new().register(worker);

    let data = web::Data::new(storage);
    let http = HttpServer::new(move || App::new().app_data(data.clone()).service(place_order))
        .bind("0.0.0.0:8080")?
        .run();

    tokio::select! {
        res = http => res,
        res = monitor.run() => res,
    }
}

use tokio_cron_scheduler::{Job, JobScheduler, JobSchedulerError};

#[tokio::main]
async fn main() -> Result<(), JobSchedulerError> {
    let scheduler = JobScheduler::new().await?;
    // Six fields: seconds first. Runs at minute 0 of every hour.
    scheduler
        .add(Job::new_async("0 0 * * * *", |_id, _scheduler| {
            Box::pin(async move {
                println!("expiring stale orders");
            })
        })?)
        .await?;
    scheduler.start().await?;
    tokio::signal::ctrl_c().await.ok();
    Ok(())
}