//! Broadcast email to followers.

use axum::{
    Form,
    extract::State,
    response::{Html, IntoResponse, Response},
};
use serde::Deserialize;

use sqlx::PgPool;

use crate::{
    auth::AuthUser,
    config::Config,
    constants, db,
    email::EmailClient,
    error::{AppError, Result},
    templates::FormStatusTemplate,
};

/// Form input for broadcasting to followers.
#[derive(Debug, Deserialize)]
pub(crate) struct BroadcastForm {
    pub subject: String,
    pub body: String,
}

/// Send a plain-text broadcast email to all followers.
#[tracing::instrument(skip_all, name = "users::broadcast_send")]
pub(in crate::routes::api) async fn broadcast_send(
    State(db): State<PgPool>,
    State(config): State<Config>,
    State(email): State<EmailClient>,
    AuthUser(user): AuthUser,
    Form(form): Form<BroadcastForm>,
) -> Result<Response> {
    user.check_not_sandbox()?;
    user.check_not_suspended()?;
    // Only creators can broadcast
    if !user.can_create_projects {
        return Ok(Html(
            FormStatusTemplate {
                success: false,
                message: "Creator access required".to_string(),
            }
            .render_string()?,
        )
        .into_response());
    }

    // Validate subject and body
    let subject = form.subject.trim();
    let body = form.body.trim();

    if subject.is_empty() || subject.chars().count() > 200 {
        return Ok(Html(
            FormStatusTemplate {
                success: false,
                message: "Subject must be between 1 and 200 characters".to_string(),
            }
            .render_string()?,
        )
        .into_response());
    }

    if body.is_empty() || body.chars().count() > 5000 {
        return Ok(Html(
            FormStatusTemplate {
                success: false,
                message: "Body must be between 1 and 5000 characters".to_string(),
            }
            .render_string()?,
        )
        .into_response());
    }

    // Rate limit: one broadcast per 24 hours
    if !db::users::try_set_broadcast_at(&db, user.id).await? {
        return Ok(Html(
            FormStatusTemplate {
                success: false,
                message: "You can only send one broadcast per 24 hours".to_string(),
            }
            .render_string()?,
        )
        .into_response());
    }

    // Get follower emails, enforcing the broadcast recipient cap at the type
    // level, `BoundedRecipients::new` is the only way to obtain a sendable list.
    let followers = db::follows::get_follower_emails(&db, user.id).await?;
    let recipients = match crate::email::BoundedRecipients::new(followers) {
        Ok(r) => r,
        Err(count) => {
            // Roll back the 24h rate-limit slot so the creator can try again after lifting the cap.
            let _ = db::users::clear_broadcast_at(&db, user.id).await;
            return Ok(Html(FormStatusTemplate {
                success: false,
                message: format!(
                    "Broadcast would reach {count} followers, above the per-send limit of 10,000. Email info@makenot.work to lift the cap for your account."
                ),
            }.render_string()?).into_response());
        }
    };
    let count = recipients.len();

    // The monthly allowance, on top of the per-send cap above and the 24h slot
    // below it. Those two bound one send and its rate; this one bounds the count
    // over the billing month, which is the number the shared Postmark IP pool's
    // reputation actually follows (`db::mail_caps`).
    let verdict =
        db::mail_caps::reserve(&db, user.id, i64::try_from(count).unwrap_or(i64::MAX)).await?;
    if let Some(message) = verdict.refusal_message() {
        // Give the 24h slot back: the creator has not spent their broadcast on
        // a send that never left, and the same rollback the per-send cap does.
        let _ = db::users::clear_broadcast_at(&db, user.id).await;
        return Ok(Html(
            FormStatusTemplate {
                success: false,
                message,
            }
            .render_string()?,
        )
        .into_response());
    }

    if count == 0 {
        return Ok(Html(
            FormStatusTemplate {
                success: true,
                message: "No followers to notify".to_string(),
            }
            .render_string()?,
        )
        .into_response());
    }

    // Get creator name
    let db_user = db::users::get_user_by_id(&db, user.id)
        .await?
        .ok_or(AppError::NotFound)?;
    let creator_name = db_user.display_name.as_deref().unwrap_or(&db_user.username);

    // The fan-out, recorded before it happens so a complaint about it can be
    // traced back here (`93f23f00`). No list: a broadcast goes to a creator's
    // followers, which is not one. `None` on failure and the mail still goes --
    // a broadcast is worth more than its attribution, and the cost is a
    // denominator rather than a send.
    let send = db::mail_attribution::record_send(
        &db,
        user.id,
        None,
        db::mail_attribution::SendKind::Broadcast,
        i64::try_from(count).unwrap_or(i64::MAX),
    )
    .await
    .inspect_err(|error| {
        tracing::warn!(error = ?error, user_id = %user.id,
            "could not record the send; this broadcast will be unattributed");
    })
    .ok();

    // Send to each follower (fire-and-forget)
    let subject = subject.to_string();
    let body = body.to_string();
    let creator_name = creator_name.to_string();
    let creator_id = user.id;
    let email_client = email.clone();
    let host_url = config.host_url.clone();
    let signing_secret = config.signing_secret.clone();

    let followers = recipients.into_inner();
    tokio::spawn(async move {
        let mut set = tokio::task::JoinSet::new();
        let chunk_delay = std::time::Duration::from_millis(constants::BROADCAST_CHUNK_DELAY_MS);

        for follower in followers {
            if set.len() >= constants::BROADCAST_PARALLELISM {
                let _ = set.join_next().await;
            }

            let email_client = email_client.clone();
            let host_url = host_url.clone();
            let signing_secret = signing_secret.clone();
            let creator_name = creator_name.clone();
            let subject = subject.clone();
            let body = body.clone();
            let creator_id_str = creator_id.to_string();

            set.spawn(async move {
                let unsub_url = crate::email::generate_unsubscribe_url(
                    &host_url,
                    follower.id,
                    crate::email::UnsubscribeAction::Broadcast,
                    &creator_id_str,
                    &signing_secret,
                );
                if let Err(e) = email_client
                    .send_broadcast(
                        &follower.email,
                        follower.display_name.as_deref(),
                        &creator_name,
                        &subject,
                        &body,
                        crate::email::Fanout {
                            unsub_url: Some(&unsub_url),
                            send,
                        },
                    )
                    .await
                {
                    tracing::warn!(error = ?e, to = %follower.email, "broadcast email failed");
                }
            });

            tokio::time::sleep(chunk_delay).await;
        }

        while set.join_next().await.is_some() {}
    });

    tracing::info!(user_id = %user.id, recipient_count = count, "broadcast sent");

    Ok(Html(
        FormStatusTemplate {
            success: true,
            message: format!(
                "Broadcast sent to {} follower{}",
                count,
                if count == 1 { "" } else { "s" }
            ),
        }
        .render_string()?,
    )
    .into_response())
}