use std::convert::Infallible;
use std::time::Duration;

use axum::Json;
use axum::extract::{Query, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use futures::stream::Stream;
use serde::{Deserialize, Serialize};
use tokio_stream::StreamExt;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::wrappers::errors::BroadcastStreamRecvError;

use crate::auth::middleware::AuthUser;
use crate::error::AppError;
use crate::models::session::Session;
use crate::state::AppState;

#[derive(Deserialize)]
pub struct SseQuery {
    pub ticket: String,
}

#[derive(Serialize)]
pub struct StreamTicketResponse {
    pub ticket: String,
    /// Server clock at mint time — the client seeds its SSE reconnect cursor from this
    /// instead of `new Date()`, so a skewed browser clock can't drop uploads. See
    /// `DeltaResponse::server_time`.
    pub server_time: chrono::DateTime<chrono::Utc>,
}

/// Mint a short-lived single-use SSE ticket. The browser's `EventSource` cannot
/// send an `Authorization` header, so the alternative used to be passing the JWT
/// as `?token=...` — which leaks the bearer token into access logs, referer
/// headers, and browser history. The client now exchanges its Bearer token for
/// an opaque ticket via this endpoint and passes that on the stream open.
pub async fn issue_ticket(
    State(state): State<AppState>,
    auth: AuthUser,
) -> Result<Json<StreamTicketResponse>, AppError> {
    let ticket = state.sse_tickets.issue(auth.token_hash);
    let server_time = sqlx::query_scalar("SELECT NOW()")
        .fetch_one(&state.pool)
        .await?;
    Ok(Json(StreamTicketResponse {
        ticket,
        server_time,
    }))
}

/// SSE stream endpoint. Authenticates via a single-use ticket (see
/// [`issue_ticket`]) — never the raw JWT.
pub async fn stream(
    State(state): State<AppState>,
    Query(q): Query<SseQuery>,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, AppError> {
    let token_hash = state
        .sse_tickets
        .consume(&q.ticket)
        .ok_or_else(|| AppError::Unauthorized("Ticket ungültig oder abgelaufen.".into()))?;

    // NOTE: this authenticates via ticket→session, not the `AuthUser` extractor. The
    // SSE stream is read-only, and under the read-only-ban model (USER_JOURNEYS §10)
    // banned users retain read access — so both minting a ticket and holding a stream
    // open are intentionally allowed for banned users; only writes are blocked.
    Session::find_by_token_hash(&state.pool, &token_hash)
        .await
        .map_err(|e| AppError::Internal(e.into()))?
        .ok_or_else(|| AppError::Unauthorized("Sitzung nicht gefunden.".into()))?;

    let rx = state.sse_tx.subscribe();
    let events = BroadcastStream::new(rx).filter_map(|msg| match msg {
        Ok(sse_event) => Some(Ok(Event::default()
            .event(sse_event.event_type)
            .data(sse_event.data))),
        // A consumer that falls behind the broadcast buffer would otherwise silently
        // lose events, leaving its feed permanently stale. Instead of dropping the
        // gap, tell the client to resync — the frontend responds by running a full
        // feed-delta fetch (which reconciles both new uploads and deletions).
        Err(BroadcastStreamRecvError::Lagged(n)) => {
            tracing::warn!("SSE consumer lagged, dropped {n} event(s); emitting resync");
            Some(Ok(Event::default().event("resync").data(n.to_string())))
        }
    });

    // The session is only checked once at open. Re-validate it periodically so a
    // logged-out, expired, OR banned session's stream is closed rather than kept alive
    // until the client happens to disconnect. Bounds a stale stream to ~60s. NOTE: a ban is
    // deliberately read-only and does NOT revoke sessions (see `ban_user`), so the
    // `is_banned` re-read below is LOAD-BEARING — it is the only thing that stops a banned
    // user's live push stream. Do not remove it. Only a *definitive* gone/expired/banned
    // state ends the stream; a transient DB error just retries next tick.
    let pool = state.pool.clone();
    let session_hash = token_hash.clone();
    let session_gone = async move {
        let mut ticker = tokio::time::interval(Duration::from_secs(60));
        ticker.tick().await; // consume the immediate first tick
        loop {
            ticker.tick().await;
            match Session::find_user_by_token_hash(&pool, &session_hash).await {
                Ok(Some(user)) if !user.is_banned => {} // still valid — keep streaming
                Ok(Some(_)) => break,                   // banned — cut the stream
                Ok(None) => break,                      // logged out or expired — stop
                Err(e) => {
                    tracing::warn!(error = ?e, "SSE session revalidation query failed; retrying");
                }
            }
        }
    };

    let stream = futures::StreamExt::take_until(events, Box::pin(session_gone));

    Ok(Sse::new(stream).keep_alive(
        KeepAlive::new()
            .interval(Duration::from_secs(30))
            .text("ping"),
    ))
}