//! Alerts that repeat until a human says they read them.
//!
//! Most mail is sent and hoped for. A few messages carry something only the
//! recipient can act on, where a message that silently failed to land is
//! indistinguishable from one that landed and was ignored, and the difference
//! matters. Those get a row here instead of a single send.
//!
//! The shape is: a caller [`open`]s a row, the scheduler sends it and keeps
//! sending it weekly, and following the link and pressing the button
//! [`acknowledge`]s it and stops the sending. Nothing sends from [`open`]
//! itself, so the first message and the repeats travel the same path and cannot
//! drift apart.
//!
//! **Acknowledgement requires a POST.** The whole value of the row is evidence
//! a human read the message, and link prefetchers and corporate mail scanners
//! follow GETs. Email verification gets away with a bare GET because a
//! prefetcher confirming an address that the address owner did receive is
//! harmless; here a prefetcher would record a person's attention that never
//! happened, which is worse than recording nothing.
//! <!-- wiki: mnw-server-overview -->

use sqlx::PgPool;

use super::enums::AckKind;
use super::id_types::{PendingAcknowledgementId, UserId};
use crate::error::Result;

/// What the acknowledgement page renders. Deliberately not the whole row: the
/// page is reachable without a session, by anyone holding the link, so it says
/// what happened and nothing about the account it belongs to.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct AcknowledgementView {
    pub kind: AckKind,
    pub details: serde_json::Value,
    pub acknowledged: bool,
}

/// One alert waiting on its recipient.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct PendingAcknowledgement {
    pub id: PendingAcknowledgementId,
    pub user_id: UserId,
    pub kind: AckKind,
    pub dedup_key: String,
    pub details: serde_json::Value,
    pub notify_count: i32,
    pub email: String,
    pub display_name: Option<String>,
}

impl PendingAcknowledgement {
    /// This row's link token. Derived, so every send produces the same one and
    /// the link in an early message still works when a later one arrives.
    pub fn token(&self, signing_secret: &str) -> String {
        crate::email::acknowledgement_token(
            self.user_id,
            &self.kind.to_string(),
            &self.dedup_key,
            signing_secret,
        )
    }
}

/// Open an alert, or return the one already open for the same thing.
///
/// Reports whether a row was opened, so a caller can log the new case without
/// logging every restatement of it. Nothing is sent here: the scheduler owns
/// every send, including the first.
///
/// Idempotent by `(user_id, kind, dedup_key)` while unacknowledged. A Stripe
/// account whose currency flaps between the same two values produces one row,
/// not a pile; a genuinely different change has a different `dedup_key` and
/// opens its own.
#[tracing::instrument(skip_all)]
pub async fn open(
    pool: &PgPool,
    user_id: UserId,
    kind: AckKind,
    dedup_key: &str,
    details: serde_json::Value,
    signing_secret: &str,
) -> Result<bool> {
    let token_hash = crate::email::hash_opaque_token(&crate::email::acknowledgement_token(
        user_id,
        &kind.to_string(),
        dedup_key,
        signing_secret,
    ));

    // DO NOTHING rather than DO UPDATE: an open row is already saying this, and
    // resetting its notify_count would restart a nag that is halfway to
    // escalation.
    let inserted = sqlx::query_scalar::<_, PendingAcknowledgementId>(
        "INSERT INTO pending_acknowledgements (user_id, kind, dedup_key, details, token_hash) \
         VALUES ($1, $2, $3, $4, $5) \
         ON CONFLICT (user_id, kind, dedup_key) WHERE acknowledged_at IS NULL \
         DO NOTHING \
         RETURNING id",
    )
    .bind(user_id)
    .bind(kind.to_string())
    .bind(dedup_key)
    .bind(&details)
    .bind(&token_hash)
    .fetch_optional(pool)
    .await?;

    Ok(inserted.is_some())
}

/// Alerts that should be sent now: never sent, or last sent longer ago than
/// `repeat_days`. Escalated rows are excluded, having been handed to a person.
///
/// Joined to `users` here rather than by the caller so the send path cannot
/// mail a suspended or deleted account, and bounded by `limit` so one pass
/// cannot unbox an unbounded set into a mail queue.
#[tracing::instrument(skip_all)]
pub async fn due(
    pool: &PgPool,
    repeat_days: i64,
    limit: i64,
) -> Result<Vec<PendingAcknowledgement>> {
    let rows = sqlx::query_as::<_, PendingAcknowledgement>(
        "SELECT pa.id, pa.user_id, pa.kind, pa.dedup_key, pa.details, pa.notify_count, \
                u.email, u.display_name \
         FROM pending_acknowledgements pa \
         JOIN users u ON u.id = pa.user_id \
         WHERE pa.acknowledged_at IS NULL \
           AND pa.escalated_at IS NULL \
           AND u.suspended_at IS NULL \
           AND (pa.last_notified_at IS NULL \
                OR pa.last_notified_at < NOW() - make_interval(days => $1::int)) \
         ORDER BY pa.last_notified_at ASC NULLS FIRST \
         LIMIT $2",
    )
    .bind(i32::try_from(repeat_days).unwrap_or(7))
    .bind(limit)
    .fetch_all(pool)
    .await?;
    Ok(rows)
}

/// Record that one more message went out.
///
/// Stamped only after a successful send, so a mail provider outage re-sends
/// next pass rather than burning a message towards escalation.
#[tracing::instrument(skip_all)]
pub async fn mark_notified(pool: &PgPool, id: PendingAcknowledgementId) -> Result<()> {
    sqlx::query(
        "UPDATE pending_acknowledgements \
         SET last_notified_at = NOW(), notify_count = notify_count + 1 \
         WHERE id = $1",
    )
    .bind(id)
    .execute(pool)
    .await?;
    Ok(())
}

/// Hand the alert to a person and stop sending it.
///
/// Returns whether this call was the one that escalated, so the caller files
/// exactly one ticket. Without that, every subsequent pass would file another.
#[tracing::instrument(skip_all)]
pub async fn escalate(pool: &PgPool, id: PendingAcknowledgementId) -> Result<bool> {
    let moved = sqlx::query(
        "UPDATE pending_acknowledgements SET escalated_at = NOW() \
         WHERE id = $1 AND escalated_at IS NULL AND acknowledged_at IS NULL",
    )
    .bind(id)
    .execute(pool)
    .await?
    .rows_affected()
        > 0;
    Ok(moved)
}

/// The open alert a link token points at, or `None`.
///
/// Acknowledged and escalated rows both still resolve. Someone clicking an old
/// link should be told what it was about rather than shown a dead end, and the
/// page decides what to say from `acknowledged`.
#[tracing::instrument(skip_all)]
pub async fn find_by_token(pool: &PgPool, token: &str) -> Result<Option<AcknowledgementView>> {
    let hash = crate::email::hash_opaque_token(token);
    let row = sqlx::query_as::<_, AcknowledgementView>(
        "SELECT pa.kind, pa.details, (pa.acknowledged_at IS NOT NULL) AS acknowledged \
         FROM pending_acknowledgements pa \
         WHERE pa.token_hash = $1",
    )
    .bind(&hash)
    .fetch_optional(pool)
    .await?;
    Ok(row)
}

/// Record that a human read it. Idempotent: a second POST reports `false` and
/// is not an error, because a double-submitted form is not a failure.
#[tracing::instrument(skip_all)]
pub async fn acknowledge(pool: &PgPool, token: &str) -> Result<bool> {
    let hash = crate::email::hash_opaque_token(token);
    let moved = sqlx::query(
        "UPDATE pending_acknowledgements SET acknowledged_at = NOW() \
         WHERE token_hash = $1 AND acknowledged_at IS NULL",
    )
    .bind(&hash)
    .execute(pool)
    .await?
    .rows_affected()
        > 0;
    Ok(moved)
}

/// The body copy for one alert, rendered from its `details`.
///
/// One source for the email and for the page the email links to. Rendered
/// rather than stored so fixing the wording is an edit and not a migration over
/// rows written months ago, and so the two can never say different things about
/// the same event.
///
/// Paragraphs are separated by a blank line, which is what the mail body wants;
/// the page splits on it.
pub fn detail_copy(kind: AckKind, details: &serde_json::Value) -> String {
    match kind {
        AckKind::SettlementCurrencyChanged => {
            let from = details["from"].as_str().unwrap_or("its old currency");
            let to = details["to"].as_str().unwrap_or("a new currency");
            format!(
                "Your Stripe account now settles in {to}, where it used to settle in {from}.\n\n\
                 Every price you have already set is stored as a plain number, so those numbers \
                 now mean {to}. A price that was 10 {from} is now 10 {to}. We have not converted \
                 anything and we have not changed any of your prices, because guessing an \
                 exchange rate on your behalf is not ours to do.\n\n\
                 Please check your prices."
            )
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn kind_round_trip() {
        for k in AckKind::ALL {
            assert_eq!(k.to_string().parse::<AckKind>().unwrap(), *k);
        }
    }

    /// Every `AckKind` is accepted by the database.
    ///
    /// Same guard, and the same reason, as
    /// `db::lists::every_kind_is_allowed_by_the_check_constraint`: a kind lives
    /// in this enum and in a CHECK constraint, and adding it to one only fails
    /// at INSERT on a deployed database, a long way from the edit that caused
    /// it.
    #[test]
    fn every_kind_is_allowed_by_the_check_constraint() {
        const SQL: &str = include_str!("../../migrations/196_pending_acknowledgements.sql");
        let clause = SQL
            .split_once("kind TEXT NOT NULL CHECK (kind IN (")
            .expect("the constraint has the expected shape")
            .1
            .split_once("))")
            .expect("the constraint list is closed")
            .0;
        let allowed: Vec<&str> = clause
            .split(',')
            .map(|s| s.trim().trim_matches('\'').trim())
            .filter(|s| !s.is_empty())
            .collect();

        for kind in AckKind::ALL {
            let s = kind.to_string();
            assert!(
                allowed.contains(&s.as_str()),
                "AckKind::{kind:?} (\"{s}\") is not in the pending_acknowledgements kind \
                 constraint. Adding a kind takes a migration as well as an enum variant.",
            );
        }
        assert_eq!(allowed.len(), AckKind::ALL.len());
    }

    /// The title is user-facing copy, so it follows the house rules.
    #[test]
    fn titles_are_clean_copy() {
        for k in AckKind::ALL {
            let t = k.title();
            assert!(!t.is_empty());
            assert!(
                !t.contains('\u{2014}') && !t.contains(" -- "),
                "{t}: no connective dashes in user-facing copy"
            );
        }
    }
}