//! The worker side of the durable work queue (oxygenhub#276,
//! docs/roles.md): every worker process takes due items, a batch at a
//! time, and does them. Work that fails where it starts (a notification's
//! email, the Git mirror after a push) is queued here instead of being
//! lost, and retried with backoff.

use crate::db::{self, queue::Item};
use crate::Shared;
use std::time::Duration;

/// How often an idle worker looks for due items.
pub const POLL: Duration = Duration::from_secs(2);
/// How long a claim lasts before another worker may take the item.
const LEASE_MS: i64 = 5 * 60 * 1000;
/// Items taken at once.
const BATCH: i64 = 10;

/// An email to send again (`mail`).
pub async fn queue_mail(
    app: &Shared,
    to_name: &str,
    to_addr: &str,
    subject: &str,
    body: &str,
    action: Option<(&str, &str)>,
    unsubscribe: Option<&str>,
) {
    let payload = serde_json::json!({
        "to_name": to_name, "to_addr": to_addr, "subject": subject, "body": body,
        "action_label": action.map(|a| a.0), "action_url": action.map(|a| a.1), "unsubscribe": unsubscribe,
    });
    if let Err(e) = db::queue::enqueue(&app.db, "mail", &payload, "", 30_000).await {
        tracing::warn!("queueing an email to {to_addr}: {e}");
    }
}

/// A repository's Git mirror to refresh again (`mirror.refresh`), once.
pub async fn queue_mirror(app: &Shared, repo_id: i64) {
    let payload = serde_json::json!({ "repo_id": repo_id });
    if let Err(e) = db::queue::enqueue(&app.db, "mirror.refresh", &payload, &repo_id.to_string(), 30_000).await {
        tracing::warn!("queueing a Git mirror refresh: {e}");
    }
}

/// This process's name as a claimant: its host and pid.
fn worker_name() -> String {
    let host = std::fs::read_to_string("/etc/hostname").map(|h| h.trim().to_string()).unwrap_or_default();
    format!("{host}:{}", std::process::id())
}

/// Take and do due items until the process stops.
pub async fn worker(app: Shared) {
    let me = worker_name();
    loop {
        let items = match db::queue::claim(&app.db, &me, BATCH, LEASE_MS).await {
            Ok(items) => items,
            Err(e) => {
                tracing::warn!("work queue: claiming: {e}");
                Vec::new()
            }
        };
        let idle = items.is_empty();
        for item in items {
            run(&app, item).await;
        }
        if idle {
            tokio::time::sleep(POLL).await;
        }
    }
}

/// Do one item and record how it went.
pub async fn run(app: &Shared, item: Item) {
    let outcome = match item.kind.as_str() {
        "mail" => mail(app, &item).await,
        "mirror.refresh" => mirror(app, &item).await,
        other => Err(format!("no worker knows {other:?}")),
    };
    let recorded = match outcome {
        Ok(()) => db::queue::done(&app.db, item.id).await,
        Err(e) => {
            tracing::warn!(kind = %item.kind, attempt = item.attempts, "work queue: {e}");
            // An item nobody can do is dead at once.
            let item = if item.kind != "mail" && item.kind != "mirror.refresh" {
                Item { attempts: db::queue::MAX_ATTEMPTS, ..item }
            } else {
                item
            };
            db::queue::failed(&app.db, &item, &e).await
        }
    };
    if let Err(e) = recorded {
        tracing::warn!("work queue: recording an outcome: {e}");
    }
}

async fn mail(app: &Shared, item: &Item) -> Result<(), String> {
    let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(|e| e.to_string())?;
    let s = |k: &str| p[k].as_str().unwrap_or_default().to_string();
    let (label, url) = (s("action_label"), s("action_url"));
    let action = (!url.is_empty()).then_some(crate::mail::Action { label: &label, url: &url });
    let unsubscribe = s("unsubscribe");
    crate::mail::send_letter(
        app,
        &s("to_name"),
        &s("to_addr"),
        &s("subject"),
        &s("body"),
        action,
        (!unsubscribe.is_empty()).then_some(unsubscribe.as_str()),
    )
    .await
}

async fn mirror(app: &Shared, item: &Item) -> Result<(), String> {
    let p: serde_json::Value = serde_json::from_str(&item.payload).map_err(|e| e.to_string())?;
    let id = p["repo_id"].as_i64().ok_or("no repo_id")?;
    // Gone, or its mirror turned off meanwhile: nothing to do.
    let Some(row) = db::repos::get_by_id(&app.db, id).await.map_err(|e| e.to_string())? else { return Ok(()) };
    crate::mirror::refresh(app, &row).await
}