//! 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

}