//! GET /admin/crawler — composed status snapshot
//! GET /admin/crawler/stream — Server-Sent Events live status
//!
//! The composed response merges the in-process status surface
//! ([`crate::crawler::status`]) with two DB-derived counts
//! (job-state breakdown and missing-cover backlog) so the dashboard
//! reads the same shape from one one-shot fetch and from each SSE
//! frame. The streaming path debounces a burst of pokes into a single
//! frame and memoizes the DB counts for the [`QUEUE_MEMO_TTL`] window
//! to avoid hammering Postgres.

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

use axum::extract::State;
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::routing::get;
use axum::{Json, Router};
use futures_util::stream::Stream;
use serde::Serialize;

use crate::app::AppState;
use crate::auth::extractor::RequireAdmin;
use crate::crawler::browser_manager::RestartPhase;
use crate::crawler::status::{ActiveChapter, CoverTarget, LastPass, Phase};
use crate::error::AppResult;
use crate::repo;

/// Backstop recompose interval for the SSE stream. Phase/worker/session
/// changes push instantly via the status `watch`; this only bounds the
/// staleness of DB-derived queue counts and the browser phase when those
/// change without an accompanying status poke.
const SSE_BACKSTOP: Duration = Duration::from_secs(5);

/// Coalesce a burst of status pokes (e.g. one per stored page during a
/// chapter download) into a single SSE frame. After the first change
/// fires we sleep this long and absorb any further changes that arrive
/// in the window before emitting. Tighter than human-perceivable jitter
/// (~16-100 ms is the rule of thumb for "instant").
const SSE_DEBOUNCE: Duration = Duration::from_millis(250);

/// Per-connection memo TTL for DB-derived queue counts. With a busy
/// chapter pass bumping the watch up to several times per second, the
/// memo collapses N stream wakeups into one round-trip per second
/// — typically a ~10x reduction in steady-state DB QPS per subscriber.
const QUEUE_MEMO_TTL: Duration = Duration::from_millis(1000);

pub(super) fn routes() -> Router<AppState> {
    Router::new()
        .route("/admin/crawler", get(get_status))
        .route("/admin/crawler/stream", get(stream_status))
}

#[derive(Debug, Serialize)]
struct QueueCounts {
    pending: i64,
    running: i64,
    dead: i64,
}

#[derive(Debug, Serialize)]
struct SessionStatus {
    /// Whether the sticky session-expired flag is set (chapter workers idle).
    expired: bool,
    /// Whether a PHPSESSID is currently configured at all.
    configured: bool,
}

#[derive(Debug, Serialize)]
struct CrawlerStatusResponse {
    /// `"running"` | `"disabled"`.
    daemon: &'static str,
    phase: Option<Phase>,
    /// Configured chapter-worker count (for "N busy / M workers").
    worker_count: usize,
    /// Chapters being crawled right now, with live page counts.
    active_chapters: Vec<ActiveChapter>,
    /// The cover being fetched right now, if any.
    current_cover: Option<CoverTarget>,
    /// Mangas still queued for a cover fetch.
    covers_queued: i64,
    last_pass: LastPass,
    session: SessionStatus,
    /// `"healthy"` | `"draining"` | `"restarting"` | `"down"`.
    browser: &'static str,
    queue: QueueCounts,
}

/// Per-stream memo of the two DB-derived counts shared by every SSE
/// frame: crawler-job state breakdown and missing-cover backlog. Held
/// across iterations of the unfold loop so a burst of status pokes
/// emits one frame from cached counts. A fresh memo (`new`) always
/// misses on first call, so `get_status` (one-shot) sees no stale data.
struct QueueCountsMemo {
    cached: Option<(std::time::Instant, (i64, i64, i64), i64)>,
}

impl QueueCountsMemo {
    fn new() -> Self {
        Self { cached: None }
    }

    async fn get(
        &mut self,
        db: &sqlx::PgPool,
    ) -> sqlx::Result<((i64, i64, i64), i64)> {
        if let Some((at, qc, cv)) = self.cached.as_ref() {
            if at.elapsed() < QUEUE_MEMO_TTL {
                return Ok((*qc, *cv));
            }
        }
        let qc = repo::crawler::job_state_counts(db).await?;
        let cv = repo::crawler::count_missing_covers(db).await?;
        self.cached = Some((std::time::Instant::now(), qc, cv));
        Ok((qc, cv))
    }
}

fn browser_phase_str(p: RestartPhase) -> &'static str {
    match p {
        RestartPhase::Healthy => "healthy",
        RestartPhase::Draining => "draining",
        RestartPhase::Restarting => "restarting",
    }
}

/// Compose a full status snapshot from the in-memory status, the
/// browser/session flags, and DB queue-count queries (routed through
/// a memo so a burst of pokes doesn't hammer Postgres). Shared by
/// `get_status` (with a fresh per-call memo) and `stream_status`
/// (with a per-stream memo held across iterations).
async fn compose_status(
    state: &AppState,
    memo: &mut QueueCountsMemo,
) -> AppResult<CrawlerStatusResponse> {
    let ((pending, running, dead), covers_queued) = memo.get(&state.db).await?;
    let queue = QueueCounts {
        pending,
        running,
        dead,
    };

    Ok(match state.crawler().as_ref() {
        None => CrawlerStatusResponse {
            daemon: "disabled",
            phase: None,
            worker_count: 0,
            active_chapters: Vec::new(),
            current_cover: None,
            covers_queued,
            last_pass: LastPass::default(),
            session: SessionStatus {
                expired: false,
                configured: false,
            },
            browser: "down",
            queue,
        },
        Some(c) => {
            let snap = c.status.snapshot().await;
            CrawlerStatusResponse {
                daemon: "running",
                phase: Some(snap.phase),
                worker_count: snap.worker_count,
                active_chapters: snap.active_chapters,
                current_cover: snap.current_cover,
                covers_queued,
                last_pass: snap.last_pass,
                session: SessionStatus {
                    expired: c.session.is_expired(),
                    configured: c.session.current().await.is_some(),
                },
                browser: browser_phase_str(c.browser_manager.phase()),
                queue,
            }
        }
    })
}

async fn get_status(
    State(state): State<AppState>,
    _admin: RequireAdmin,
) -> AppResult<Json<CrawlerStatusResponse>> {
    // Fresh memo — one-shot calls always query the DB. The wrapper exists
    // only so compose_status can share its signature with the streaming
    // path; there is no caching across one-shot calls.
    let mut memo = QueueCountsMemo::new();
    Ok(Json(compose_status(&state, &mut memo).await?))
}

/// Push live status to the dashboard instead of polling. Emits a snapshot
/// immediately on connect, then on every status change (instant, via the
/// `watch` notifier) and on a [`SSE_BACKSTOP`] tick (to refresh DB queue
/// counts / browser phase that change without a status poke). The browser
/// opens this only while the crawler page is mounted and closes it on
/// navigate-away, so the subscription is scoped to the active page.
async fn stream_status(
    State(state): State<AppState>,
    _admin: RequireAdmin,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    // Subscribe before the first emit so no change between the initial
    // snapshot and the first await is lost.
    let rx = state.crawler().as_ref().map(|c| c.status.subscribe());
    let memo = QueueCountsMemo::new();

    let stream = futures_util::stream::unfold(
        (state, rx, memo, true),
        |(state, mut rx, mut memo, first)| async move {
            // After the first immediate emit, wait for a change or the
            // backstop tick before recomposing. On a change, debounce a
            // short window so a burst of pokes (one per stored page
            // during a chapter download, etc.) coalesces into a single
            // frame instead of hammering subscribers.
            if !first {
                match rx.as_mut() {
                    Some(rx) => {
                        tokio::select! {
                            _ = rx.changed() => {
                                tokio::time::sleep(SSE_DEBOUNCE).await;
                                // Mark any pokes that arrived during the
                                // debounce window as observed so the next
                                // iteration only fires on NEW changes.
                                rx.borrow_and_update();
                            }
                            _ = tokio::time::sleep(SSE_BACKSTOP) => {}
                        }
                    }
                    None => tokio::time::sleep(SSE_BACKSTOP).await,
                }
            }
            // Compose; on a transient DB error, emit a keep-alive comment
            // rather than tearing down the stream.
            let event = match compose_status(&state, &mut memo).await {
                Ok(resp) => Event::default()
                    .event("status")
                    .json_data(&resp)
                    .unwrap_or_else(|_| Event::default().comment("serialize error")),
                Err(_) => Event::default().comment("status unavailable"),
            };
            Some((Ok(event), (state, rx, memo, false)))
        },
    );

    Sse::new(stream).keep_alive(KeepAlive::default())
}
