use std::time::Duration;

use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, DbErr, Statement};
use serde_json::Value;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use crate::envelope::Envelope;
use crate::error::QueueError;
use crate::publisher::Publisher;
use crate::route;

const BATCH_SIZE: i64 = 16;
const POLL_INTERVAL: Duration = Duration::from_millis(500);
const PUBLISH_TIMEOUT: Duration = Duration::from_secs(10);
const MAX_PUBLISH_ATTEMPTS: i32 = 50;
const PURGE_INTERVAL: Duration = Duration::from_secs(600);
const PURGED_RETENTION_SECS: i64 = 7 * 24 * 60 * 60;

const CLAIM: &str = "
WITH pending AS (
    SELECT candidate.id FROM event_outbox AS candidate
    WHERE 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.published_at IS NULL
            AND (earlier.occurred_at, earlier.id) < (candidate.occurred_at, candidate.id)
      )
    ORDER BY candidate.occurred_at, candidate.id
    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.kind, events.envelope, events.attempts
";

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

pub async fn run(db: DatabaseConnection, publisher: Publisher, cancel: CancellationToken) {
    let mut purge_at = tokio::time::Instant::now() + PURGE_INTERVAL;
    loop {
        if cancel.is_cancelled() {
            return;
        }
        if tokio::time::Instant::now() >= purge_at {
            purge_at = tokio::time::Instant::now() + PURGE_INTERVAL;
            purge_published(&db).await;
        }
        match run_batch(&db, &publisher).await {
            Ok(count) if count > 0 => continue,
            Ok(_) => {}
            Err(error) => tracing::error!(%error, "event outbox relay failed"),
        }
        tokio::select! {
            _ = cancel.cancelled() => return,
            _ = tokio::time::sleep(POLL_INTERVAL) => {}
        }
    }
}

async fn run_batch(db: &DatabaseConnection, publisher: &Publisher) -> 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")?,
            kind: row.try_get("", "kind")?,
            envelope: row.try_get("", "envelope")?,
            attempts: row.try_get("", "attempts")?,
        });
    }

    let count = pending.len();
    for row in pending {
        publish(db, publisher, lease_id, row).await?;
    }
    Ok(count)
}

async fn publish(
    db: &DatabaseConnection,
    publisher: &Publisher,
    lease_id: Uuid,
    row: PendingEvent,
) -> Result<(), DbErr> {
    let result = tokio::time::timeout(PUBLISH_TIMEOUT, publish_body(publisher, &row))
        .await
        .unwrap_or_else(|_| Err(QueueError::publish("outbox publish timed out")));

    let statement = match result {
        Ok(()) => Statement::from_sql_and_values(
            DbBackend::Postgres,
            "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
            [row.id.into(), lease_id.into()],
        ),
        Err(error) => {
            if row.attempts >= MAX_PUBLISH_ATTEMPTS {
                tracing::error!(
                    event_id = %row.id,
                    kind = %row.kind,
                    attempt = row.attempts,
                    %error,
                    "outbox event exceeded the retry budget and was discarded"
                );
                Statement::from_sql_and_values(
                    DbBackend::Postgres,
                    "UPDATE event_outbox SET published_at = CURRENT_TIMESTAMP, lease_id = NULL, leased_until = NULL WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
                    [row.id.into(), lease_id.into()],
                )
            } else {
                tracing::warn!(
                    event_id = %row.id,
                    kind = %row.kind,
                    attempt = row.attempts,
                    %error,
                    "outbox publish will retry"
                );
                let delay = retry_delay(row.attempts);
                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 WHERE id = $1 AND lease_id = $2 AND published_at IS NULL",
                    [row.id.into(), lease_id.into(), delay.into()],
                )
            }
        }
    };

    let updated = db.execute_raw(statement).await?;
    if updated.rows_affected() == 0 {
        tracing::warn!(event_id = %row.id, "outbox lease was reclaimed by another relay");
    }
    Ok(())
}

async fn publish_body(publisher: &Publisher, row: &PendingEvent) -> Result<(), QueueError> {
    let envelope: Envelope<Value> = serde_json::from_value(row.envelope.clone())?;
    if envelope.id != row.id || envelope.kind != row.kind {
        return Err(QueueError::payload("outbox event identity is invalid"));
    }
    let route = route::route(&envelope.kind)?;
    let topic = publisher.topic(&route.area);
    let body = envelope.summary_body(&route.tag)?;
    let properties = envelope.properties();
    publisher
        .publish(
            &topic,
            &route.tag,
            vec![row.id.to_string()],
            properties,
            body,
        )
        .await
}

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

async fn purge_published(db: &DatabaseConnection) {
    match db
        .execute_unprepared(&format!(
            "DELETE FROM event_outbox WHERE published_at < CURRENT_TIMESTAMP - INTERVAL '{PURGED_RETENTION_SECS} seconds'"
        ))
        .await
    {
        Ok(result) if result.rows_affected() > 0 => {
            tracing::info!(
                rows = result.rows_affected(),
                "purged published outbox events"
            );
        }
        Ok(_) => {}
        Err(error) => tracing::warn!(%error, "outbox purge failed"),
    }
}