//! The webhook delivery worker: claims due deliveries, sends them with bounded concurrency, and
//! records each result. A failed attempt is retried on the backoff schedule until the sixth
//! attempt, after which the delivery is `gave_up`.
//!
//! The signature is computed at send time from the decrypted secret, so a rotated secret applies
//! to retries too. The headers queued at enqueue time never carry a signature or a delivery GUID;
//! both are added here. After each send, the headers actually sent replace the stored ones, so the
//! delivery log shows what the receiver got.

use std::sync::Arc;
use std::time::{Duration, Instant};

use forge_core::Error;
use forge_core::time::{now_stamp, stamp_after, stamp_ago};
use forge_core::webhooks::{AttemptOutcome, Delivery};
use futures_util::StreamExt;
use serde_json::{Map, Value};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tokio::sync::Notify;

use super::client::{HookClient, SendOutcome, WebhookPolicy, sign, validate_url};
use crate::state::AppState;

/// Handle that wakes the worker. `emit` calls `notify_one` after queueing deliveries.
pub type WebhookWaker = Arc<Notify>;

/// Deliveries claimed per pass.
const BATCH: u32 = 16;
/// Deliveries sent at once.
const CONCURRENCY: usize = 4;
/// The shortest sleep, so a due time that has already passed does not spin the loop.
const MIN_SLEEP: Duration = Duration::from_secs(1);
/// The longest sleep. It only guards against a missed wake, it is not polling.
const BACKSTOP: Duration = Duration::from_secs(15 * 60);
/// Deliveries older than this are purged.
const RETENTION: Duration = Duration::from_secs(14 * 24 * 60 * 60);
/// How often the purge runs.
const PURGE_EVERY: Duration = Duration::from_secs(24 * 60 * 60);
/// Delay before the next attempt, indexed by the attempt that just failed (attempt 1 waits 1m).
/// After the sixth attempt there is no retry.
const BACKOFF: [Duration; 5] = [
    Duration::from_secs(60),
    Duration::from_secs(5 * 60),
    Duration::from_secs(30 * 60),
    Duration::from_secs(2 * 60 * 60),
    Duration::from_secs(6 * 60 * 60),
];
/// Headers the worker owns. Anything stored under these names is replaced at send time.
const WORKER_HEADERS: [&str; 3] = ["Content-Type", "X-Forge-Delivery", "X-Forge-Signature-256"];

/// Starts the worker task and returns the handle that wakes it. The task runs until the process
/// exits; it never panics on a bad delivery.
pub fn spawn(state: AppState) -> WebhookWaker {
    let waker = Arc::clone(&state.webhook_waker);
    let task_waker = Arc::clone(&waker);
    tokio::spawn(run(state, task_waker));
    waker
}

async fn run(state: AppState, waker: WebhookWaker) {
    let policy = match WebhookPolicy::from_config(&state.config.webhooks) {
        Ok(policy) => policy,
        Err(e) => {
            // Fall back to the strictest policy: no internal hosts and no plain HTTP.
            tracing::error!(error = %e, "webhooks: invalid [webhooks] config, using the default policy");
            WebhookPolicy {
                allow_http: false,
                allowed_hosts: Vec::new(),
            }
        }
    };
    let client = match HookClient::new(policy.clone()) {
        Ok(client) => client,
        Err(e) => {
            tracing::error!(error = %e, "webhooks: cannot build the HTTP client, deliveries stop");
            return;
        }
    };

    let mut last_purge: Option<Instant> = None;
    loop {
        purge_if_due(&state, &mut last_purge).await;
        let claimed = match process_due(&state, &client, &policy, &now_stamp()).await {
            Ok(claimed) => claimed,
            Err(e) => {
                tracing::error!(error = %e, "webhooks: claiming deliveries failed");
                0
            }
        };
        // A full batch may mean more is due, so go straight back for it.
        if claimed < usize::try_from(BATCH).unwrap_or(usize::MAX) {
            // Wake when the earliest pending delivery is due (a claimed one wakes at its lease
            // expiry), or at the next purge. With nothing pending, sleep up to `BACKSTOP`.
            let next = match state.store.next_delivery_due().await {
                Ok(next) => next,
                Err(e) => {
                    tracing::error!(error = %e, "webhooks: could not read the next due time");
                    None
                }
            };
            let nap = next
                .as_deref()
                .map_or(BACKSTOP, |due| sleep_for(due, OffsetDateTime::now_utc()));
            let purge_nap = last_purge.map_or(Duration::ZERO, |at| {
                PURGE_EVERY.saturating_sub(at.elapsed())
            });
            tokio::select! {
                () = waker.notified() => {}
                () = tokio::time::sleep(nap) => {}
                () = tokio::time::sleep(purge_nap) => {}
            }
        }
    }
}

/// How long the worker sleeps until `next_due` (an RFC 3339 stamp, the earliest pending
/// delivery), clamped to `MIN_SLEEP..=BACKSTOP`. An unparsable stamp gives `MIN_SLEEP`, so it is
/// retried promptly rather than stalling.
fn sleep_for(next_due: &str, now: OffsetDateTime) -> Duration {
    let Ok(due) = OffsetDateTime::parse(next_due, &Rfc3339) else {
        return MIN_SLEEP;
    };
    let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO);
    until.clamp(MIN_SLEEP, BACKSTOP)
}

/// Claims up to [`BATCH`] due deliveries at `now`, sends them, and records each result. Returns
/// how many were claimed.
pub(crate) async fn process_due(
    state: &AppState,
    client: &HookClient,
    policy: &WebhookPolicy,
    now: &str,
) -> forge_core::Result<usize> {
    let claimed = state.store.claim_due_deliveries(now, BATCH).await?;
    let count = claimed.len();
    futures_util::stream::iter(claimed)
        .for_each_concurrent(CONCURRENCY, |delivery| async move {
            let outcome = attempt(state, client, policy, &delivery).await;
            if let Err(e) = state
                .store
                .record_delivery_result(&delivery.id, &outcome)
                .await
            {
                tracing::error!(error = %e, delivery_id = %delivery.id, "webhooks: recording the result failed");
            }
        })
        .await;
    Ok(count)
}

/// One send attempt for a claimed delivery. Never fails: every problem becomes an outcome.
async fn attempt(
    state: &AppState,
    client: &HookClient,
    policy: &WebhookPolicy,
    delivery: &Delivery,
) -> AttemptOutcome {
    let hook = match state.store.webhook(&delivery.hook_id).await {
        Ok(Some(hook)) => hook,
        Ok(None) => return give_up("webhook was deleted"),
        Err(e) => return retry(delivery, &format!("loading the webhook failed: {e}")),
    };
    let secret = match state.store.webhook_secret(&hook.id, &state.key).await {
        Ok(secret) => secret,
        Err(Error::NotFound) => return give_up("webhook was deleted"),
        Err(Error::Crypto(e)) => return give_up(&format!("cannot open the webhook secret: {e}")),
        Err(e) => return retry(delivery, &format!("loading the webhook secret failed: {e}")),
    };
    // Policy can change between save and send, so the URL is checked again here.
    if let Err(e) = validate_url(&hook.url, policy) {
        return retry(delivery, &e.to_string());
    }
    let body = delivery.request_body.clone().into_bytes();
    let headers = request_headers(delivery, &secret, &body);
    let sent = client.send(&hook.url, &headers, body).await;
    let mut result = outcome(delivery, sent);
    result.request_headers = Some(headers_json(&headers));
    result
}

/// The headers as a JSON object, the form the store keeps. Stored after each send so the delivery
/// log shows the GUID and signature the receiver got.
fn headers_json(headers: &[(String, String)]) -> String {
    let map: Map<String, Value> = headers
        .iter()
        .map(|(name, value)| (name.clone(), Value::String(value.clone())))
        .collect();
    Value::Object(map).to_string()
}

/// The stored headers, minus the ones the worker owns, plus the fresh GUID, content type and
/// signature.
fn request_headers(delivery: &Delivery, secret: &str, body: &[u8]) -> Vec<(String, String)> {
    let mut headers: Vec<(String, String)> = match serde_json::from_str::<Map<String, Value>>(
        &delivery.request_headers,
    ) {
        Ok(map) => map
            .into_iter()
            .filter_map(|(name, value)| match value {
                Value::String(value) => Some((name, value)),
                _ => None,
            })
            .collect(),
        Err(e) => {
            tracing::error!(error = %e, delivery_id = %delivery.id, "webhooks: stored headers are not a JSON object");
            Vec::new()
        }
    };
    headers.retain(|(name, _)| {
        !WORKER_HEADERS
            .iter()
            .any(|owned| owned.eq_ignore_ascii_case(name))
    });
    headers.push(("Content-Type".to_owned(), "application/json".to_owned()));
    headers.push(("X-Forge-Delivery".to_owned(), delivery.guid.clone()));
    headers.push((
        "X-Forge-Signature-256".to_owned(),
        sign(secret.as_bytes(), body),
    ));
    headers
}

/// Turns the send result into the outcome to record. Only a 2xx status counts as success.
fn outcome(delivery: &Delivery, sent: SendOutcome) -> AttemptOutcome {
    let ok = matches!(sent.status, Some(200..=299));
    let mut response_map = Map::new();
    for (name, value) in sent.headers {
        response_map.insert(name, Value::String(value));
    }
    AttemptOutcome {
        ok,
        status: sent.status,
        response_headers: Some(Value::Object(response_map).to_string()),
        response_body: Some(sent.body),
        error: sent.error,
        duration_ms: Some(sent.duration_ms),
        next_attempt_at: if ok {
            None
        } else {
            next_attempt_at(delivery.attempt)
        },
        request_headers: None,
    }
}

/// A failed attempt that will be retried, when the schedule has a slot for it.
fn retry(delivery: &Delivery, error: &str) -> AttemptOutcome {
    AttemptOutcome {
        error: Some(error.to_owned()),
        next_attempt_at: next_attempt_at(delivery.attempt),
        ..AttemptOutcome::default()
    }
}

/// A failure that will not be retried.
fn give_up(error: &str) -> AttemptOutcome {
    AttemptOutcome {
        error: Some(error.to_owned()),
        ..AttemptOutcome::default()
    }
}

/// When the attempt after `attempt` is due, or `None` once the schedule is used up.
fn next_attempt_at(attempt: u32) -> Option<String> {
    let index = usize::try_from(attempt.checked_sub(1)?).ok()?;
    let delay = BACKOFF.get(index)?;
    Some(stamp_after(*delay))
}

/// Deletes deliveries older than [`RETENTION`], at most once per [`PURGE_EVERY`].
async fn purge_if_due(state: &AppState, last: &mut Option<Instant>) {
    if last.is_some_and(|at| at.elapsed() < PURGE_EVERY) {
        return;
    }
    *last = Some(Instant::now());
    let cutoff = stamp_ago(RETENTION);
    match state.store.purge_deliveries_before(&cutoff).await {
        Ok(0) => {}
        Ok(removed) => tracing::info!(removed, "webhooks: purged old deliveries"),
        Err(e) => tracing::error!(error = %e, "webhooks: purging old deliveries failed"),
    }
}

#[cfg(test)]
mod tests {
    use forge_core::Visibility;
    use forge_core::webhooks::{Delivery, DeliveryState, Webhook, WebhookEvent};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;
    use tokio::sync::mpsc;

    use super::*;
    use crate::test_support::{KEY, test_state};

    /// A request as the local test server saw it.
    struct Seen {
        headers: Vec<(String, String)>,
        body: Vec<u8>,
    }

    /// Serves one request per connection, reporting each request on the channel, and answers
    /// with `status`.
    async fn server(status: u16) -> (String, mpsc::UnboundedReceiver<Seen>) {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("addr");
        let (tx, rx) = mpsc::unbounded_channel();
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let tx = tx.clone();
                tokio::spawn(async move {
                    let mut buf = Vec::new();
                    let mut chunk = [0u8; 1024];
                    let head_end = loop {
                        if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
                            break pos + 4;
                        }
                        match socket.read(&mut chunk).await {
                            Ok(0) | Err(_) => return,
                            Ok(n) => buf.extend_from_slice(&chunk[..n]),
                        }
                    };
                    let head = String::from_utf8_lossy(&buf[..head_end]).into_owned();
                    let headers: Vec<(String, String)> = head
                        .lines()
                        .skip(1)
                        .filter_map(|line| {
                            line.split_once(':')
                                .map(|(k, v)| (k.trim().to_owned(), v.trim().to_owned()))
                        })
                        .collect();
                    let length: usize = headers
                        .iter()
                        .find(|(k, _)| k.eq_ignore_ascii_case("content-length"))
                        .and_then(|(_, v)| v.parse().ok())
                        .unwrap_or(0);
                    while buf.len() < head_end + length {
                        match socket.read(&mut chunk).await {
                            Ok(0) | Err(_) => break,
                            Ok(n) => buf.extend_from_slice(&chunk[..n]),
                        }
                    }
                    let body = buf[head_end..].to_vec();
                    let _ = tx.send(Seen { headers, body });
                    let reply = format!(
                        "HTTP/1.1 {status} X\r\nContent-Length: 2\r\nConnection: close\r\n\r\nno"
                    );
                    let _ = socket.write_all(reply.as_bytes()).await;
                    let _ = socket.shutdown().await;
                });
            }
        });
        (format!("http://127.0.0.1:{}/hook", addr.port()), rx)
    }

    fn local_policy() -> WebhookPolicy {
        WebhookPolicy::from_config(&forge_core::config::WebhookConfig {
            allow_http: true,
            allowed_hosts: vec!["127.0.0.1/32".into()],
        })
        .expect("policy")
    }

    /// A repo with one hook pointing at `url`, subscribed to push, and one queued delivery.
    async fn queued(state: &AppState, url: &str) -> (Webhook, Delivery, tempfile::TempDir) {
        let tmp = tempfile::tempdir().expect("tempdir");
        let owner = state
            .store
            .create_user("alice", None, "unused", false)
            .await
            .expect("user");
        let repo = state
            .store
            .create_repo(&owner.id, "proj", "", Visibility::Public, "main")
            .await
            .expect("repo");
        let sealed = forge_core::secret::seal(&KEY, b"s3cret").expect("seal");
        let hook = state
            .store
            .create_webhook(
                &repo.id,
                &owner.id,
                url,
                sealed.as_bytes(),
                &[WebhookEvent::Push],
                true,
            )
            .await
            .expect("hook");
        let headers = r#"{"Content-Type":"application/json","X-Forge-Event":"push","X-Forge-Delivery":"stale","X-Forge-Signature-256":"sha256=stale"}"#;
        let delivery = state
            .store
            .enqueue_delivery(
                &hook.id,
                WebhookEvent::Push,
                "",
                headers,
                r#"{"ref":"refs/heads/main"}"#,
            )
            .await
            .expect("enqueue");
        (hook, delivery, tmp)
    }

    #[tokio::test]
    async fn delivers_with_fresh_signature_and_headers() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let state = test_state(tmp.path()).await;
        let (url, mut seen) = server(204).await;
        let (hook, delivery, _dir) = queued(&state, &url).await;
        let client = HookClient::new(local_policy()).expect("client");

        let claimed = process_due(&state, &client, &local_policy(), &now_stamp())
            .await
            .expect("process");
        assert_eq!(claimed, 1);

        let request = seen.recv().await.expect("request arrives");
        let header = |name: &str| {
            request
                .headers
                .iter()
                .find(|(k, _)| k.eq_ignore_ascii_case(name))
                .map(|(_, v)| v.clone())
        };
        assert_eq!(request.body, br#"{"ref":"refs/heads/main"}"#);
        assert_eq!(
            header("x-forge-signature-256"),
            Some(sign(b"s3cret", &request.body))
        );
        assert_eq!(header("x-forge-delivery"), Some(delivery.guid.clone()));
        assert_eq!(header("x-forge-event"), Some("push".into()));
        assert_eq!(header("content-type"), Some("application/json".into()));

        let done = state
            .store
            .delivery(&delivery.id)
            .await
            .expect("lookup")
            .expect("row");
        assert_eq!(done.state, DeliveryState::Ok);
        assert_eq!(done.response_status, Some(204));
        assert!(done.next_attempt_at.is_none());
        assert_eq!(done.hook_id, hook.id);

        // The stored request headers are what was sent, including the GUID and signature.
        let stored: Value = serde_json::from_str(&done.request_headers).expect("json");
        assert_eq!(stored["X-Forge-Delivery"], delivery.guid.as_str());
        assert_eq!(
            stored["X-Forge-Signature-256"],
            sign(b"s3cret", &request.body).as_str()
        );
        assert_eq!(stored["X-Forge-Event"], "push");
    }

    #[tokio::test]
    async fn failure_schedules_a_retry_and_six_attempts_give_up() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let state = test_state(tmp.path()).await;
        let (url, _seen) = server(500).await;
        let (_hook, delivery, _dir) = queued(&state, &url).await;
        let client = HookClient::new(local_policy()).expect("client");

        process_due(&state, &client, &local_policy(), &now_stamp())
            .await
            .expect("process");
        let failed = state
            .store
            .delivery(&delivery.id)
            .await
            .expect("lookup")
            .expect("row");
        assert_eq!(failed.state, DeliveryState::Failed);
        assert_eq!(failed.response_status, Some(500));
        let due = failed.next_attempt_at.clone().expect("retry scheduled");
        let expected = stamp_after(Duration::from_secs(60));
        // Both stamps are RFC 3339 UTC, so a text comparison orders them; allow a few seconds.
        let parse = |s: &str| s.parse::<jiff::Timestamp>().expect("stamp").as_second();
        let drift = parse(&due) - parse(&expected);
        assert!(
            drift.abs() <= 5,
            "retry at {due}, expected about {expected}"
        );

        // Each later attempt is claimed at its due time. Attempts 2 through 6 all fail.
        let far_future = stamp_after(Duration::from_secs(10 * 24 * 60 * 60));
        let mut last = failed;
        for attempt in 2..=6u32 {
            process_due(&state, &client, &local_policy(), &far_future)
                .await
                .expect("process");
            last = state
                .store
                .delivery(&delivery.id)
                .await
                .expect("lookup")
                .expect("row");
            assert_eq!(last.attempt, attempt);
        }
        assert_eq!(last.state, DeliveryState::GaveUp);
        assert!(last.next_attempt_at.is_none());
    }

    #[tokio::test]
    async fn invalid_url_is_a_failed_attempt_not_a_send() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let state = test_state(tmp.path()).await;
        // Loopback is not in the allow list of this policy, so validation refuses it.
        let strict = WebhookPolicy::from_config(&forge_core::config::WebhookConfig::default())
            .expect("policy");
        let (_hook, delivery, _dir) = queued(&state, "https://127.0.0.1/hook").await;
        let client = HookClient::new(strict.clone()).expect("client");
        assert!(validate_url("https://127.0.0.1/hook", &strict).is_err());

        process_due(&state, &client, &strict, &now_stamp())
            .await
            .expect("process");
        let row = state
            .store
            .delivery(&delivery.id)
            .await
            .expect("lookup")
            .expect("row");
        assert_eq!(row.state, DeliveryState::Failed);
        assert!(row.error.is_some());
        assert!(row.next_attempt_at.is_some());
    }

    #[test]
    fn sleep_stops_at_the_next_due_time_and_never_spins() {
        use time::macros::datetime;

        let now = datetime!(2026-10-15 12:00:00 UTC);
        let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp");
        // Due in 10 minutes: sleep until then.
        let in_10m = stamp(now + time::Duration::minutes(10));
        assert_eq!(sleep_for(&in_10m, now), Duration::from_secs(600));
        // Due later than the backstop: wake at the backstop.
        let in_1h = stamp(now + time::Duration::hours(1));
        assert_eq!(sleep_for(&in_1h, now), BACKSTOP);
        // Already due, or a lease that has just passed: wait the minimum.
        let past = stamp(now - time::Duration::seconds(5));
        assert_eq!(sleep_for(&past, now), MIN_SLEEP);
        // A stamp that does not parse also waits the minimum.
        assert_eq!(sleep_for("soon", now), MIN_SLEEP);
    }

    #[test]
    fn backoff_schedule_matches_the_contract() {
        assert_eq!(next_attempt_at(0), None);
        assert!(next_attempt_at(1).is_some());
        assert!(next_attempt_at(5).is_some());
        assert_eq!(next_attempt_at(6), None);
        assert_eq!(BACKOFF[0], Duration::from_secs(60));
        assert_eq!(BACKOFF[4], Duration::from_secs(6 * 60 * 60));
    }
}