//! The mail worker. It leases due outbox rows and sends them. A transient failure is retried on
//! the schedule in `forge_core::mail::retry_delay`; a permanent one, or the last failed try, gives
//! up. It sleeps until the next due time, or until `AppState::mail_waker` is notified (a message
//! was queued, or the settings changed). While mail is off it sends nothing and keeps the
//! messages queued.

use std::time::Duration;

use forge_core::mail::{MailError, OutboxEntry, OutgoingMail, retry_delay};
use forge_core::time::{now_stamp, stamp_after};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;

use crate::state::AppState;

/// The longest the worker sleeps when nothing wakes it.
const POLL: Duration = Duration::from_secs(60);
/// The shortest sleep, so a due time that has already passed does not spin the loop.
const MIN_SLEEP: Duration = Duration::from_secs(1);
/// Messages leased per pass.
const BATCH: u32 = 20;
/// How long a leased row stays with this pass before another may take it.
const LEASE: Duration = Duration::from_secs(5 * 60);

/// Starts the worker task. It runs until the process exits.
pub fn spawn(state: AppState) {
    tokio::spawn(run(state));
}

async fn run(state: AppState) {
    loop {
        let nap = match tick(&state).await {
            Ok(nap) => nap,
            Err(e) => {
                tracing::error!(error = %e, "mail: could not run a pass");
                POLL
            }
        };
        tokio::select! {
            _ = tokio::time::sleep(nap) => {}
            _ = state.mail_waker.notified() => {}
        }
    }
}

/// One pass: sends what is due when mail is on. Returns how long to sleep before the next one.
async fn tick(state: &AppState) -> forge_core::Result<Duration> {
    let settings = state.store.mail_settings().await?;
    if !settings.enabled {
        return Ok(POLL);
    }
    let now = now_stamp();
    let lease_until = stamp_after(LEASE);
    for entry in state.store.lease_mail(&now, &lease_until, BATCH).await? {
        deliver(state, &entry).await;
    }
    let next = state.store.next_mail_due().await?;
    Ok(sleep_for(next.as_deref(), OffsetDateTime::now_utc()))
}

/// Sends one leased message and records the outcome.
async fn deliver(state: &AppState, entry: &OutboxEntry) {
    let mail = OutgoingMail {
        to: entry.to_addr.clone(),
        subject: entry.subject.clone(),
        body: entry.body.clone(),
        message_id: entry.message_id.clone(),
        in_reply_to: entry.in_reply_to.clone(),
    };
    let recorded = match state.mailer.send(&mail).await {
        Ok(()) => state.store.mark_mail_sent(&entry.id, &now_stamp()).await,
        Err(MailError::Permanent(error)) => {
            tracing::warn!(id = %entry.id, error = %error, "mail: permanent failure, giving up");
            state.store.mark_mail_failed(&entry.id, &error, None).await
        }
        Err(MailError::Transient(error)) => {
            // `attempts` counts the tries before this one; this try makes one more.
            let next = retry_delay(entry.attempts + 1).map(stamp_after);
            if next.is_none() {
                tracing::warn!(id = %entry.id, error = %error, "mail: last try failed, giving up");
            } else {
                tracing::info!(id = %entry.id, error = %error, "mail: try failed, will retry");
            }
            state
                .store
                .mark_mail_failed(&entry.id, &error, next.as_deref())
                .await
        }
    };
    if let Err(e) = recorded {
        tracing::error!(id = %entry.id, error = %e, "mail: could not record the outcome");
    }
}

/// How long the worker sleeps: until `next_due` (an RFC 3339 stamp), clamped to
/// `MIN_SLEEP..=POLL`. `None` or an unparsable stamp gives `POLL`.
pub fn sleep_for(next_due: Option<&str>, now: OffsetDateTime) -> Duration {
    let Some(due) = next_due.and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok()) else {
        return POLL;
    };
    let until = Duration::try_from(due - now).unwrap_or(Duration::ZERO);
    until.clamp(MIN_SLEEP, POLL)
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use forge_core::mail::{MailSettingsUpdate, OutboxStatus, Security};
    use forge_core::time::{now_stamp, stamp_after};

    use super::*;
    use crate::mail::{FakeMailer, enqueue};
    use crate::test_support::test_state;

    async fn enabled_state(dir: &std::path::Path) -> (AppState, Arc<FakeMailer>) {
        let mut state = test_state(dir).await;
        let fake = Arc::new(FakeMailer::default());
        state.mailer = fake.clone();
        state
            .store
            .update_mail_settings(&MailSettingsUpdate {
                host: "smtp.example",
                port: 587,
                security: Security::StartTls,
                username: "",
                from_addr: "forge@example.com",
                from_name: "forge",
                enabled: true,
                password: None,
            })
            .await
            .expect("settings");
        (state, fake)
    }

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

        let now = datetime!(2026-10-15 12:00:00 UTC);
        let stamp = |t: OffsetDateTime| t.format(&Rfc3339).expect("stamp");
        assert_eq!(sleep_for(None, now), POLL);
        let in_10s = stamp(now + time::Duration::seconds(10));
        assert_eq!(sleep_for(Some(&in_10s), now), Duration::from_secs(10));
        let in_1h = stamp(now + time::Duration::hours(1));
        assert_eq!(sleep_for(Some(&in_1h), now), POLL);
        let past = stamp(now - time::Duration::seconds(5));
        assert_eq!(sleep_for(Some(&past), now), MIN_SLEEP);
        assert_eq!(sleep_for(Some("soon"), now), POLL);
    }

    #[tokio::test]
    async fn a_queued_message_is_sent_once() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (state, fake) = enabled_state(dir.path()).await;
        assert!(
            enqueue(&state, "reset:1", "a@example.com", "Hi", "body")
                .await
                .expect("queued")
        );
        // A second enqueue with the same key is a no-op.
        assert!(
            !enqueue(&state, "reset:1", "a@example.com", "Hi", "body")
                .await
                .expect("again")
        );

        tick(&state).await.expect("pass");
        tick(&state).await.expect("second pass");
        let sent = fake.sent();
        assert_eq!(sent.len(), 1);
        assert_eq!(sent[0].to, "a@example.com");
        let row = state.store.recent_mail(10).await.expect("rows");
        assert_eq!(row.len(), 1);
        assert_eq!(row[0].status(), OutboxStatus::Sent);
    }

    #[tokio::test]
    async fn nothing_is_sent_while_mail_is_off() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (state, fake) = enabled_state(dir.path()).await;
        state
            .store
            .update_mail_settings(&MailSettingsUpdate {
                host: "smtp.example",
                port: 587,
                security: Security::StartTls,
                username: "",
                from_addr: "forge@example.com",
                from_name: "forge",
                enabled: false,
                password: None,
            })
            .await
            .expect("off");
        enqueue(&state, "k", "a@example.com", "Hi", "body")
            .await
            .expect("queued");
        assert_eq!(tick(&state).await.expect("pass"), POLL);
        assert!(fake.sent().is_empty());
        let row = state.store.recent_mail(10).await.expect("rows");
        assert_eq!(row[0].status(), OutboxStatus::Queued);
        assert_eq!(row[0].attempts, 0, "waiting costs no attempts");
    }

    #[tokio::test]
    async fn a_transient_failure_is_retried_later_and_a_permanent_one_gives_up() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (state, fake) = enabled_state(dir.path()).await;
        enqueue(&state, "retry", "a@example.com", "Hi", "body")
            .await
            .expect("queued");
        enqueue(&state, "bounce", "b@example.com", "Hi", "body")
            .await
            .expect("queued");
        fake.fail_next(MailError::Transient("timeout".into()));
        fake.fail_next(MailError::Permanent("550 no such user".into()));

        tick(&state).await.expect("pass");
        let rows = state.store.recent_mail(10).await.expect("rows");
        let by_key = |to: &str| rows.iter().find(|r| r.to_addr == to).expect("row").clone();

        let retried = by_key("a@example.com");
        assert_eq!(retried.status(), OutboxStatus::Queued);
        assert_eq!(retried.attempts, 1);
        assert_eq!(retried.last_error.as_deref(), Some("timeout"));
        let due = retried.next_attempt_at.expect("scheduled");
        assert!(due > now_stamp(), "the retry waits about a minute");
        assert!(due <= stamp_after(Duration::from_secs(120)));

        let bounced = by_key("b@example.com");
        assert_eq!(bounced.status(), OutboxStatus::GaveUp);
        assert_eq!(bounced.attempts, 1);
        assert_eq!(bounced.last_error.as_deref(), Some("550 no such user"));
        assert_eq!(fake.sent().len(), 0, "nothing reached the transport");
    }

    #[tokio::test]
    async fn the_last_try_gives_up_and_an_earlier_one_reschedules() {
        let dir = tempfile::tempdir().expect("tempdir");
        let (state, fake) = enabled_state(dir.path()).await;
        enqueue(&state, "stubborn", "a@example.com", "Hi", "body")
            .await
            .expect("queued");
        let row = state.store.recent_mail(10).await.expect("rows").remove(0);
        fake.fail_next(MailError::Transient("timeout".into()));
        fake.fail_next(MailError::Transient("timeout".into()));

        // The row has failed MAX_ATTEMPTS - 1 times already: this try is the last one.
        let last = OutboxEntry {
            attempts: forge_core::mail::MAX_ATTEMPTS - 1,
            ..row.clone()
        };
        deliver(&state, &last).await;
        let after = state.store.recent_mail(10).await.expect("rows").remove(0);
        assert_eq!(after.status(), OutboxStatus::GaveUp);
        assert_eq!(after.last_error.as_deref(), Some("timeout"));

        // The same failure on an early try is rescheduled instead.
        let early = OutboxEntry { attempts: 1, ..row };
        deliver(&state, &early).await;
        let after = state.store.recent_mail(10).await.expect("rows").remove(0);
        assert_eq!(after.status(), OutboxStatus::Queued);
    }
}