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

use chrono::{DateTime, Utc};
use futures_util::future;
use models::event_outbox;
use opentelemetry::metrics::{Counter, Gauge};
use opentelemetry::{KeyValue, global};
use queue::{DeliveryClass, EVENT_PREFIX, Envelope, Publisher, QueueError, SCHEMA_VERSION};
use sea_orm::ActiveValue::Set;
use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, DbErr, EntityTrait, Statement};
use serde::Serialize;
use serde_json::Value;
use tokio::task::{JoinHandle, JoinSet};
use tokio_util::sync::CancellationToken;
use tracing::Instrument;
use uuid::Uuid;

use crate::handler::concrete_kind;
use crate::{EventHandler, EventKind, Labels, label};

const BATCH_SIZE: i64 = 16;
const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
const POLL_INTERVAL: Duration = Duration::from_millis(500);
const BACKLOG_SAMPLE_INTERVAL: Duration = Duration::from_secs(10);
/// 单条事件的重试上限。超过后标记为 discarded 并打 ERROR：否则一条毒消息会永久堵住它所在的
/// ordering_key（`auth:{user_id}` 之类），后面的事件永远发不出去。
const MAX_PUBLISH_ATTEMPTS: i32 = 50;
/// 已投递行的保留时长与清理间隔。outbox 只是投递队列，不承担归档职责。
/// discarded 行保留更久，便于事后回答「哪些事件丢了、影响谁」。
const PURGE_INTERVAL: Duration = Duration::from_secs(600);
const PURGED_RETENTION_SECS: i64 = 7 * 24 * 60 * 60;
const DISCARDED_RETENTION_SECS: i64 = 90 * 24 * 60 * 60;

static BACKLOG_AGE: LazyLock<Gauge<f64>> = LazyLock::new(|| {
    global::meter("events")
        .f64_gauge("event.outbox.oldest.ready.age")
        .with_unit("s")
        .build()
});

static PUBLISHED: LazyLock<Counter<u64>> = LazyLock::new(|| {
    global::meter("events")
        .u64_counter("event.outbox.published")
        .build()
});
static RETRIED: LazyLock<Counter<u64>> = LazyLock::new(|| {
    global::meter("events")
        .u64_counter("event.outbox.retried")
        .build()
});
static DISCARDED: LazyLock<Counter<u64>> = LazyLock::new(|| {
    global::meter("events")
        .u64_counter("event.outbox.discarded")
        .build()
});
static LEASE_LOST: LazyLock<Counter<u64>> = LazyLock::new(|| {
    global::meter("events")
        .u64_counter("event.outbox.lease_lost")
        .build()
});
static ENQUEUE_FAILED: LazyLock<Counter<u64>> = LazyLock::new(|| {
    global::meter("events")
        .u64_counter("event.outbox.enqueue_failed")
        .build()
});

const ENQUEUE_ATTEMPTS: usize = 3;

// `published_at IS NULL` 只是滚动升级期间的兼容条件：旧 relay 领取/投递后只写
// published_at，不会写 status；这些行必须视为已终结，否则会被重投或永久堵住 key。
const CLAIM: &str = "
WITH pending AS (
    SELECT candidate.id FROM event_outbox AS candidate
    WHERE candidate.status = 0 AND candidate.published_at IS NULL
      AND candidate.available_at <= CURRENT_TIMESTAMP
      AND (candidate.leased_until IS NULL OR candidate.leased_until <= CURRENT_TIMESTAMP)
      AND NOT EXISTS (
          SELECT 1 FROM event_outbox AS earlier
          WHERE earlier.ordering_key = candidate.ordering_key
            AND earlier.status = 0 AND earlier.published_at IS NULL
            AND earlier.seq < candidate.seq
      )
    ORDER BY candidate.available_at, candidate.seq
    LIMIT $1 FOR UPDATE OF candidate SKIP LOCKED
)
UPDATE event_outbox AS events
SET lease_id = $2, leased_until = CURRENT_TIMESTAMP + INTERVAL '60 seconds',
    attempts = LEAST(events.attempts, 2147483646) + 1
FROM pending WHERE events.id = pending.id
RETURNING events.id, events.subject, events.kind, events.envelope, events.attempts
";

fn require_durable(kind: EventKind) -> Result<(), DbErr> {
    if kind.durable() {
        return Ok(());
    }
    Err(DbErr::Custom(format!(
        "event `{kind}` requires an outbox-enabled business domain"
    )))
}

#[tracing::instrument(skip_all, fields(kind = %kind, request_id = %envelope::context::resolve_request_id(_request_id)))]
pub async fn enqueue_at<T: Serialize>(
    conn: &impl ConnectionTrait,
    kind: EventKind,
    labels: Labels,
    payload: T,
    trace_id: &str,
    available_at: DateTime<Utc>,
    _request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
    require_durable(kind)?;
    let (id, model) = build_model(kind, labels, payload, trace_id, available_at)?;
    event_outbox::Entity::insert(model)
        .exec_without_returning(conn)
        .await?;
    Ok(id)
}

pub async fn enqueue_now<T: Serialize>(
    conn: &impl ConnectionTrait,
    kind: EventKind,
    labels: Labels,
    payload: T,
    trace_id: &str,
    request_id: Option<&str>,
) -> Result<Uuid, DbErr> {
    enqueue_at(
        conn,
        kind,
        labels,
        payload,
        trace_id,
        Utc::now(),
        request_id,
    )
    .await
}

#[tracing::instrument(skip_all, fields(kind = %kind))]
pub async fn emit_retry<T: Serialize>(
    conn: &impl ConnectionTrait,
    kind: EventKind,
    labels: Labels,
    payload: &T,
) -> Result<Uuid, DbErr> {
    let mut attempt = 1_usize;
    loop {
        match crate::emit::emit(conn, kind, labels.clone(), payload).await {
            Ok(id) => return Ok(id),
            Err(error) if attempt >= ENQUEUE_ATTEMPTS => {
                ENQUEUE_FAILED.add(1, &[KeyValue::new("domain", kind.domain())]);
                return Err(error);
            }
            Err(error) => {
                tracing::warn!(kind = %kind, attempt, %error, "retrying the outbox enqueue");
                tokio::time::sleep(Duration::from_millis(50) * attempt as u32).await;
                attempt += 1;
            }
        }
    }
}

fn build_model<T: Serialize>(
    kind: EventKind,
    labels: Labels,
    payload: T,
    trace_id: &str,
    available_at: DateTime<Utc>,
) -> Result<(Uuid, event_outbox::ActiveModel), DbErr> {
    require_durable(kind)?;
    let ordering_key = kind.ordering_key(&labels)?;
    let labels = labels.without(label::ORDERING_KEY);
    let mut envelope = Envelope::new(kind.as_str(), payload)
        .with_delivery(DeliveryClass::BusinessEvent)
        .with_source(kind.domain())
        .with_labels(labels);
    if !trace_id.is_empty() {
        envelope = envelope.with_trace_id(trace_id);
    }
    let subject = envelope
        .bound_subject(EVENT_PREFIX)
        .map_err(|error| DbErr::Custom(format!("invalid outbox event kind: {error}")))?;
    let encoded = envelope
        .encode()
        .map_err(|error| DbErr::Custom(format!("encoding outbox event: {error}")))?;
    if encoded.len() > queue::MAX_ENVELOPE_BYTES {
        return Err(DbErr::Custom(
            "outbox event exceeds the size limit".to_owned(),
        ));
    }
    let stored = serde_json::from_slice(&encoded)
        .map_err(|error| DbErr::Custom(format!("storing outbox event: {error}")))?;
    let model = event_outbox::ActiveModel {
        id: Set(envelope.id),
        subject: Set(subject),
        kind: Set(envelope.kind.clone()),
        ordering_key: Set(ordering_key),
        envelope: Set(stored),
        occurred_at: Set(envelope.occurred_at),
        available_at: Set(available_at),
        ..Default::default()
    };
    Ok((envelope.id, model))
}

struct PendingEvent {
    id: Uuid,
    subject: String,
    kind: String,
    envelope: Value,
    attempts: i32,
}

fn retry_delay(attempts: i32) -> f64 {
    f64::from(2_u32.pow(attempts.clamp(1, 8) as u32))
}

async fn publish_identity(
    handler: Option<&EventHandler>,
    row: &PendingEvent,
) -> Result<(), QueueError> {
    let handler =
        handler.ok_or_else(|| QueueError::publish("identity event handler is not configured"))?;
    let request_id = row
        .envelope
        .get("labels")
        .and_then(|labels| labels.get(label::REQUEST_ID))
        .or_else(|| {
            row.envelope
                .get("payload")
                .and_then(|payload| payload.get(label::REQUEST_ID))
        })
        .and_then(Value::as_str)
        .unwrap_or_default();
    let mut envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;
    if envelope.id != row.id || envelope.schema_version != SCHEMA_VERSION {
        return Err(QueueError::payload(
            "outbox event identity or schema is invalid",
        ));
    }
    envelope.attempt = row.attempts.max(1) as u32;
    handler
        .publish_durable(&row.subject, &envelope, Some(request_id))
        .await
}

async fn publish_app(publisher: Option<&Publisher>, row: &PendingEvent) -> Result<(), QueueError> {
    let publisher =
        publisher.ok_or_else(|| QueueError::publish("app event publisher is not configured"))?;
    let mut envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;
    if envelope.id != row.id
        || envelope.kind != row.kind
        || envelope.schema_version != SCHEMA_VERSION
    {
        return Err(QueueError::payload("outbox event identity is invalid"));
    }
    envelope.verify_event_subject(&row.subject)?;
    envelope.attempt = row.attempts.max(1) as u32;
    let kind = concrete_kind(&envelope.kind)?;
    let topic = publisher.topic(kind.topic());
    let body = envelope.encode()?;
    let payload = serde_json::to_value(&envelope.payload).ok();
    let tag = kind.tag(payload.as_ref());
    publisher
        .publish(
            &topic,
            tag,
            vec![row.id.to_string()],
            envelope.properties(payload.as_ref(), None),
            body,
        )
        .await
}

enum PublishOutcome {
    Published,
    Retried,
    Discarded(&'static str),
}

async fn publish(
    db: &DatabaseConnection,
    identity: Option<&EventHandler>,
    app: Option<&Publisher>,
    lease_id: Uuid,
    row: PendingEvent,
) -> Result<(), DbErr> {
    let request_id = row
        .envelope
        .get("labels")
        .and_then(|labels| labels.get(label::REQUEST_ID))
        .or_else(|| {
            row.envelope
                .get("payload")
                .and_then(|payload| payload.get(label::REQUEST_ID))
        })
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned();
    let span = tracing::info_span!(
        "outbox.publish",
        event_id = %row.id,
        subject = %row.subject,
        kind = %row.kind,
        request_id = %request_id,
    );
    let result = async {
        match concrete_kind(&row.kind) {
            Ok(kind) if kind.identity() => publish_identity(identity, &row).await,
            _ => publish_app(app, &row).await,
        }
    }
    .instrument(span);
    let result = tokio::time::timeout(PUBLISH_TIMEOUT, result)
        .await
        .unwrap_or_else(|_| Err(QueueError::publish("outbox publish timed out")));
    let (statement, outcome) = match result {
        Ok(()) => (
            Statement::from_sql_and_values(
                DbBackend::Postgres,
                "UPDATE event_outbox SET status = 1, published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = NULL WHERE id = $1 AND lease_id = $2 AND status = 0",
                [row.id.into(), lease_id.into()],
            ),
            PublishOutcome::Published,
        ),
        Err(error) => {
            let permanent = !error.is_retryable();
            let error_text: String = error.to_string().chars().take(2048).collect();
            if permanent || row.attempts >= MAX_PUBLISH_ATTEMPTS {
                tracing::error!(
                    event_id = %row.id,
                    subject = %row.subject,
                    kind = %row.kind,
                    attempt = row.attempts,
                    permanent,
                    error = %error_text,
                    "outbox event was discarded"
                );
                (
                    Statement::from_sql_and_values(
                        DbBackend::Postgres,
                        "UPDATE event_outbox SET status = 2, discarded_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL, last_error = $3 WHERE id = $1 AND lease_id = $2 AND status = 0",
                        [row.id.into(), lease_id.into(), error_text.into()],
                    ),
                    PublishOutcome::Discarded(if permanent {
                        "permanent"
                    } else {
                        "retry_budget"
                    }),
                )
            } else {
                tracing::warn!(event_id = %row.id, subject = %row.subject, kind = %row.kind, attempt = row.attempts, error = %error_text, "outbox publish will retry");
                let delay = retry_delay(row.attempts) + f64::from(row.id.as_bytes()[15] % 5);
                (
                    Statement::from_sql_and_values(
                        DbBackend::Postgres,
                        "UPDATE event_outbox SET available_at = CURRENT_TIMESTAMP + ($3 * INTERVAL '1 second'), lease_id = NULL, leased_until = NULL, last_error = $4 WHERE id = $1 AND lease_id = $2 AND status = 0",
                        [
                            row.id.into(),
                            lease_id.into(),
                            delay.into(),
                            error_text.into(),
                        ],
                    ),
                    PublishOutcome::Retried,
                )
            }
        }
    };
    let event_domain = concrete_kind(&row.kind)
        .map(EventKind::domain)
        .unwrap_or_default();
    let updated = db.execute_raw(statement).await?;
    if updated.rows_affected() == 0 {
        LEASE_LOST.add(1, &[KeyValue::new("domain", event_domain)]);
        tracing::warn!(event_id = %row.id, "outbox lease was reclaimed by another relay");
    } else {
        match outcome {
            PublishOutcome::Published => {
                PUBLISHED.add(1, &[KeyValue::new("domain", event_domain)]);
            }
            PublishOutcome::Retried => {
                RETRIED.add(1, &[KeyValue::new("domain", event_domain)]);
            }
            PublishOutcome::Discarded(reason) => {
                DISCARDED.add(
                    1,
                    &[
                        KeyValue::new("domain", event_domain),
                        KeyValue::new("reason", reason),
                    ],
                );
            }
        }
    }
    Ok(())
}

#[tracing::instrument(skip_all, fields(request_id = %envelope::context::resolve_request_id(_request_id)))]
async fn run_batch(
    db: &DatabaseConnection,
    identity: Option<&EventHandler>,
    app: Option<&Publisher>,
    _request_id: Option<&str>,
) -> Result<usize, DbErr> {
    let lease_id = Uuid::now_v7();
    let claimed = db
        .query_all_raw(Statement::from_sql_and_values(
            DbBackend::Postgres,
            CLAIM,
            [BATCH_SIZE.into(), lease_id.into()],
        ))
        .await?;
    let mut pending = Vec::with_capacity(claimed.len());
    for row in claimed {
        pending.push(PendingEvent {
            id: row.try_get("", "id")?,
            subject: row.try_get("", "subject")?,
            kind: row.try_get("", "kind")?,
            envelope: row.try_get("", "envelope")?,
            attempts: row.try_get("", "attempts")?,
        });
    }
    let count = pending.len();
    let results = future::join_all(
        pending
            .into_iter()
            .map(|row| publish(db, identity, app, lease_id, row)),
    )
    .await;
    for result in results {
        result?;
    }
    Ok(count)
}

async fn purge_rows(db: &DatabaseConnection, state: &'static str, sql: &str) {
    match db.execute_unprepared(sql).await {
        Ok(result) if result.rows_affected() > 0 => {
            tracing::info!(rows = result.rows_affected(), state, "purged outbox events");
        }
        Ok(_) => {}
        Err(error) => tracing::warn!(%error, state, "outbox purge failed"),
    }
}

async fn purge_finished(db: &DatabaseConnection) {
    purge_rows(
        db,
        "published",
        &format!(
            // status 0 + published_at set 只会来自滚动升级期间的旧 relay，按 published 清掉。
            "DELETE FROM event_outbox WHERE id IN (SELECT id FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds' LIMIT 10000)"
        ),
    )
    .await;
    purge_rows(
        db,
        "discarded",
        &format!(
            "DELETE FROM event_outbox WHERE id IN (SELECT id FROM event_outbox WHERE status = 2 AND discarded_at < CURRENT_TIMESTAMP - INTERVAL '{DISCARDED_RETENTION_SECS} seconds' LIMIT 10000)"
        ),
    )
    .await;
}

async fn run_database(
    domain: &'static str,
    db: DatabaseConnection,
    identity: Option<EventHandler>,
    app: Option<Publisher>,
    cancel: CancellationToken,
) {
    let mut sample_at = Instant::now();
    let mut purge_at = Instant::now() + PURGE_INTERVAL;
    loop {
        if cancel.is_cancelled() {
            return;
        }
        if Instant::now() >= purge_at {
            purge_at = Instant::now() + PURGE_INTERVAL;
            purge_finished(&db).await;
        }
        if Instant::now() >= sample_at {
            sample_at = Instant::now() + BACKLOG_SAMPLE_INTERVAL;
            let sample = db.query_one_raw(Statement::from_string(
                DbBackend::Postgres,
                "SELECT COALESCE(EXTRACT(EPOCH FROM (CURRENT_TIMESTAMP - (SELECT available_at FROM event_outbox WHERE status = 0 AND published_at IS NULL AND available_at <= CURRENT_TIMESTAMP ORDER BY available_at, seq LIMIT 1))), 0)::double precision AS age",
            )).await;
            match sample {
                Ok(Some(row)) => match row.try_get::<f64>("", "age") {
                    Ok(age) => BACKLOG_AGE.record(age.max(0.0), &[KeyValue::new("domain", domain)]),
                    Err(error) => {
                        tracing::warn!(%domain, %error, "outbox backlog sample is invalid")
                    }
                },
                Ok(None) => {}
                Err(error) => tracing::warn!(%domain, %error, "outbox backlog sample failed"),
            }
        }
        match run_batch(&db, identity.as_ref(), app.as_ref(), None).await {
            Ok(count) if count > 0 => continue,
            Ok(_) => {}
            Err(error) => {
                tracing::error!(%domain, %error, "outbox relay database operation failed")
            }
        }
        tokio::select! {
            _ = cancel.cancelled() => return,
            _ = tokio::time::sleep(POLL_INTERVAL) => {}
        }
    }
}

pub struct OutboxRelay {
    cancel: CancellationToken,
    join: Option<JoinHandle<Result<(), DbErr>>>,
}

impl OutboxRelay {
    pub fn start(
        identity: Option<EventHandler>,
        app: Option<Publisher>,
        databases: Vec<(&'static str, DatabaseConnection)>,
    ) -> Self {
        let cancel = CancellationToken::new();
        let join = if identity.is_some() || app.is_some() {
            let cancel = cancel.clone();
            Some(tokio::spawn(async move {
                let mut workers = JoinSet::new();
                for (domain, db) in databases {
                    workers.spawn(run_database(
                        domain,
                        db,
                        identity.clone(),
                        app.clone(),
                        cancel.clone(),
                    ));
                }
                while let Some(result) = workers.join_next().await {
                    result.map_err(|error| {
                        DbErr::Custom(format!("outbox relay task failed: {error}"))
                    })?;
                    if !cancel.is_cancelled() {
                        return Err(DbErr::Custom(
                            "outbox relay stopped unexpectedly".to_owned(),
                        ));
                    }
                }
                Ok(())
            }))
        } else {
            None
        };
        if join.is_none() {
            tracing::warn!(
                "event forwarding is disabled; reliable events remain pending in the database outbox"
            );
        }
        Self { cancel, join }
    }

    pub async fn wait(&mut self) -> Result<(), DbErr> {
        let Some(join) = self.join.as_mut() else {
            return std::future::pending().await;
        };
        let result = join.await;
        self.join.take();
        result.map_err(|error| DbErr::Custom(format!("outbox relay task failed: {error}")))??;
        Err(DbErr::Custom(
            "outbox relay stopped unexpectedly".to_owned(),
        ))
    }

    pub async fn shutdown(mut self) -> Result<(), DbErr> {
        self.cancel.cancel();
        let Some(mut join) = self.join.take() else {
            return Ok(());
        };
        match tokio::time::timeout(SHUTDOWN_TIMEOUT, &mut join).await {
            Ok(result) => result
                .map_err(|error| DbErr::Custom(format!("outbox relay task failed: {error}")))?,
            Err(_) => {
                join.abort();
                let _ = join.await;
                Err(DbErr::Custom(
                    "outbox relay shutdown timed out; pending leases will recover".to_owned(),
                ))
            }
        }
    }
}

impl Drop for OutboxRelay {
    fn drop(&mut self) {
        self.cancel.cancel();
    }
}

#[cfg(test)]
mod tests {
    use crate::{Labels, label};

    #[test]
    fn auth_events_serialize_per_user_and_fall_back_to_request_id() {
        let user = Labels::new()
            .with(label::USER_ID, 42)
            .with(label::REQUEST_ID, "req-1");
        assert_eq!(
            label::AUTH_LOGIN_CREATE.ordering_key(&user).unwrap(),
            "auth:42"
        );

        let anonymous = Labels::new().with(label::REQUEST_ID, "req-2");
        assert_eq!(
            label::AUTH_LOGIN_CREATE.ordering_key(&anonymous).unwrap(),
            "auth:req-2"
        );
    }

    #[test]
    fn organization_and_email_use_their_own_resource_ids() {
        let org = Labels::new()
            .with(label::ORGANIZATION_ID, "org-1")
            .with(label::ACTOR_USER_ID, 7);
        assert_eq!(
            label::ORGANIZATION_LIFE_CYCLE_CREATE
                .ordering_key(&org)
                .unwrap(),
            "organization:org-1"
        );

        let mail = Labels::new().with(label::REQUEST_ID, "mail-1");
        assert_eq!(
            label::EMAIL_DELIVERY_CREATE.ordering_key(&mail).unwrap(),
            "email:mail-1"
        );
    }

    #[test]
    fn events_without_any_correlation_label_are_rejected() {
        let error = label::AUTH_LOGIN_CREATE
            .ordering_key(&Labels::new())
            .unwrap_err();
        assert!(
            error
                .to_string()
                .contains("requires a user, organization or request id")
        );
    }
}