//! `/admin/compaction(s)` handlers.
//!
//! * `POST /admin/compaction` proposes a compaction of the current epoch. The receiving
//!   node becomes its coordinator. `202` with the report, or with `?wait=true` the final
//!   report once decided. A dropped request does not abort anything. `?force=true`
//!   accepts a `settleSecs` below the dormancy threshold (tests, single-site clusters).
//! * `GET /admin/compactions` lists all compactions.
//! * `GET /admin/compactions/{id}` reports one; `?wait=true` long-polls until decided
//!   (bounded by the proposal's own expiry).
//! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether
//!   the node is catching up with a newer epoch from a peer.
//! * `DELETE /admin/compactions/{id}` asks the coordinator to abort. Only possible while
//!   proposed; `409` once committed (a switch can't be undone) or already over.
//!   `DELETE /admin/compaction` does the same for the compaction of the local epoch that
//!   is in progress, without having to know its ID.
//!
//! Request body for POST (all optional):
//! `{ "replayWindow": {"epochs": 1, "seconds": 604800},
//!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
//!    "settleSecs": 480 }`

use std::time::Duration;

use axum::{
    body::Bytes,
    extract::{Path, Query, State},
    http::{header::LOCATION, StatusCode},
    response::{IntoResponse, Response},
    Json,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value as JsonValue};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
use uuid::Uuid;

use super::coordinator;
use super::local::LocalStatus;
use super::records::{
    self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention,
};
use super::Host;
use crate::store::StoreError;

#[derive(Debug, Default, Deserialize)]
pub(crate) struct WaitQuery {
    #[serde(default)]
    wait: bool,
}

#[derive(Debug, Default, Deserialize)]
pub(crate) struct StartQuery {
    #[serde(default)]
    wait: bool,
    /// Allows a settle period shorter than the dormancy threshold.
    #[serde(default)]
    force: bool,
}

#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct StartRequest {
    replay_window: Option<Retention>,
    archive_retention: Option<Retention>,
    settle_secs: Option<u64>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct Report {
    id: String,
    phase: Phase,
    #[serde(flatten)]
    record: Record,
    /// What happened to this compaction on the node answering the request.
    #[serde(skip_serializing_if = "Option::is_none")]
    local: Option<LocalStatus>,
}

#[derive(Debug, thiserror::Error)]
enum StartError {
    #[error("compaction {0} is already open for this epoch")]
    Open(String),
    #[error(transparent)]
    Store(#[from] StoreError),
}

fn error_response(
    status: StatusCode,
    message: impl Into<String>,
    details: Option<JsonValue>,
) -> Response {
    let mut body = json!({ "error": message.into() });
    if let (Some(JsonValue::Object(extra)), Some(object)) = (details, body.as_object_mut()) {
        object.extend(extra);
    }
    (status, Json(body)).into_response()
}

fn grace<H: Host>(host: &H) -> u64 {
    host.dormant_after().as_millis() as u64
}

fn report<H: Host>(host: &H, id: String, record: Record) -> Report {
    Report {
        phase: record.phase(unix_millis(), grace(host)),
        local: host.compaction().status(&id),
        id,
        record,
    }
}

async fn load_report<H: Host>(host: &H, id: &str) -> Option<Report> {
    let wanted = id.to_owned();
    let record = host
        .read_doc(move |doc| records::load(doc, &wanted))
        .await?;
    Some(report(host, id.to_owned(), record))
}

/// Waits until the compaction leaves `proposed`. Bounded by the proposal's own expiry,
/// at which point it is reported as `expired`.
async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
    let mut changes = host.subscribe_changes();
    loop {
        let report = load_report(host, id).await?;
        if report.phase.is_final() {
            return Some(report);
        }
        // Re-read on every published change, and at least once a second: expiry is a
        // matter of time, not of document changes.
        if let Ok(Err(broadcast::error::RecvError::Closed)) =
            tokio::time::timeout(Duration::from_secs(1), changes.recv()).await
        {
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    }
}

pub(crate) async fn start<H: Host>(
    State(host): State<H>,
    Query(query): Query<StartQuery>,
    body: Bytes,
) -> Response {
    if !host.is_node_ready().await {
        return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
    }

    let request: StartRequest = if body.is_empty() {
        StartRequest::default()
    } else {
        match serde_json::from_slice(&body) {
            Ok(request) => request,
            Err(error) => {
                return error_response(
                    StatusCode::BAD_REQUEST,
                    format!("invalid request: {error}"),
                    None,
                )
            }
        }
    };

    let dormant_after = host.dormant_after();
    let settle = request
        .settle_secs
        .map(Duration::from_secs)
        .unwrap_or(dormant_after * 2);
    if settle < dormant_after && query.force {
        warn!(
            settle_secs = settle.as_secs_f64(),
            dormant_after_secs = dormant_after.as_secs(),
            "Forced settle period shorter than the dormancy threshold; a partition may go unnoticed"
        );
    } else if settle < dormant_after {
        return error_response(
            StatusCode::BAD_REQUEST,
            format!(
                "settleSecs must be at least {} (the dormancy threshold), so a partition is detected before deciding; add ?force=true to override",
                dormant_after.as_secs()
            ),
            None,
        );
    }

    // Fail fast instead of proposing something that will abort anyway.
    let peers = host.peer_summaries().await;
    let dormant: Vec<String> = peers
        .iter()
        .filter(|peer| !peer.active)
        .map(|peer| peer.id.clone())
        .collect();
    if !dormant.is_empty() {
        return error_response(
            StatusCode::CONFLICT,
            "dormant peers block compaction; forget them first (DELETE /peers/{pk})",
            Some(json!({ "dormantPeers": dormant })),
        );
    }

    let mut peers_considered: Vec<String> = peers.into_iter().map(|peer| peer.id).collect();
    peers_considered.sort();

    let now = unix_millis();
    let settle_ms = settle.as_millis() as u64;
    let id = Uuid::new_v4().to_string();
    let grace = grace(&host);

    let mut proposal = Proposal {
        coordinator: host.node_id(),
        from_epoch: host.current_epoch().into(),
        base_heads: Vec::new(),
        policy: Policy {
            replay_window: request
                .replay_window
                .unwrap_or_else(Policy::default_replay_window),
            archive_retention: request
                .archive_retention
                .unwrap_or_else(Policy::default_archive_retention),
        },
        settle_ms,
        proposed_at: now,
        settle_until: now + settle_ms,
        peers_considered,
    };

    // From here on, nothing depends on the client: writing the proposal and starting the
    // coordinator run in their own task, so a client that disconnects mid-request can't
    // leave a proposal without a coordinator. Dropping the handle doesn't cancel it.
    let proposing = tokio::spawn(propose(host.clone(), id.clone(), proposal, now, grace));
    let written = match proposing.await {
        Ok(written) => written,
        Err(error) => {
            error!(error = %error, "Proposal task failed");
            return error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to write the proposal",
                None,
            );
        }
    };

    let record = match written {
        Ok(record) => record,
        Err(StartError::Open(open)) => {
            return error_response(
                StatusCode::CONFLICT,
                "another compaction of this epoch is still open",
                Some(json!({ "compaction": open })),
            );
        }
        Err(StartError::Store(error)) => {
            error!(error = %format!("{error:#}"), "Failed to write compaction proposal");
            return error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to write the proposal",
                None,
            );
        }
    };

    info!(
        compaction = %id,
        settle_secs = settle.as_secs(),
        "Compaction proposed"
    );

    if query.wait {
        return match wait_until_final(&host, &id).await {
            Some(report) => Json(report).into_response(),
            None => error_response(StatusCode::NOT_FOUND, "compaction record vanished", None),
        };
    }

    let location = format!("/admin/compactions/{id}");
    (
        StatusCode::ACCEPTED,
        [(LOCATION, location)],
        Json(report(&host, id, record)),
    )
        .into_response()
}

/// Writes the proposal (unless another one is open for the epoch) and starts its
/// coordinator. Runs as its own task, independent of the HTTP request.
async fn propose<H: Host>(
    host: H,
    id: String,
    mut proposal: Proposal,
    now: u64,
    grace: u64,
) -> Result<Record, StartError> {
    let record_id = id.clone();
    let record = host
        .commit_doc(move |doc| -> Result<Record, StartError> {
            if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| {
                record.proposal.from_epoch == proposal.from_epoch
                    && record.phase(now, grace).is_open()
            }) {
                return Err(StartError::Open(open));
            }

            // The base is everything before the proposal; the proposal itself is not in it.
            let mut heads: Vec<String> = doc.get_heads().iter().map(|h| h.to_string()).collect();
            heads.sort();
            proposal.base_heads = heads;

            records::write_proposal(doc, &record_id, &proposal)?;
            Ok(Record {
                proposal,
                decision: None,
                abort_requested: None,
            })
        })
        .await?;

    tokio::spawn(coordinator::run(host, id));
    Ok(record)
}

pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
    let all = host.read_doc(records::load_all).await;
    let mut reports: Vec<Report> = all
        .into_iter()
        .map(|(id, record)| report(&host, id, record))
        .collect();
    reports.sort_by_key(|report| report.record.proposal.proposed_at);
    Json(reports).into_response()
}

pub(crate) async fn get<H: Host>(
    State(host): State<H>,
    Path(id): Path<String>,
    Query(query): Query<WaitQuery>,
) -> Response {
    let found = if query.wait {
        wait_until_final(&host, &id).await
    } else {
        load_report(&host, &id).await
    };
    match found {
        Some(report) => Json(report).into_response(),
        None => error_response(StatusCode::NOT_FOUND, "no such compaction", None),
    }
}

/// `DELETE /admin/compactions/{id}`.
pub(crate) async fn abort<H: Host>(
    State(host): State<H>,
    Path(id): Path<String>,
    Query(query): Query<WaitQuery>,
) -> Response {
    abort_by_id(host, id, query.wait).await
}

/// `DELETE /admin/compaction`: aborts whichever compaction of the local epoch is still
/// proposed. There is at most one, unless two coordinators raced (then both abort anyway
/// and this picks the oldest).
pub(crate) async fn abort_current<H: Host>(
    State(host): State<H>,
    Query(query): Query<WaitQuery>,
) -> Response {
    let local = EpochRecord::from(host.current_epoch());
    let now = unix_millis();
    let grace = grace(&host);
    let open = host
        .read_doc(move |doc| {
            records::load_all(doc)
                .into_iter()
                .filter(|(_, record)| {
                    record.proposal.from_epoch == local
                        && record.phase(now, grace) == Phase::Proposed
                })
                .min_by_key(|(_, record)| record.proposal.proposed_at)
                .map(|(id, _)| id)
        })
        .await;

    match open {
        Some(id) => abort_by_id(host, id, query.wait).await,
        None => error_response(
            StatusCode::NOT_FOUND,
            "no compaction of the local epoch is in progress",
            None,
        ),
    }
}

async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
    let Some(current) = load_report(&host, &id).await else {
        return error_response(StatusCode::NOT_FOUND, "no such compaction", None);
    };

    match current.phase {
        Phase::Proposed => {}
        Phase::Committed => {
            return error_response(
                StatusCode::CONFLICT,
                "already committed; the switch cannot be undone",
                Some(json!({ "compaction": current })),
            )
        }
        Phase::Aborted | Phase::Expired => {
            return error_response(
                StatusCode::CONFLICT,
                "already over",
                Some(json!({ "compaction": current })),
            )
        }
    }

    // The coordinator applies the request on its next check (within about a second).
    // If it is gone, the proposal expires anyway.
    if current.record.abort_requested.is_none() {
        let request = AbortRequest {
            by: host.node_id(),
            at: unix_millis(),
        };
        let record_id = id.clone();
        if let Err(error) = host
            .commit_doc(move |doc| records::write_abort_request(doc, &record_id, &request))
            .await
        {
            error!(compaction = %id, error = %format!("{error:#}"), "Failed to write abort request");
            return error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to record the abort request",
                None,
            );
        }
        info!(compaction = %id, "Abort requested");
    }

    let report = if wait {
        wait_until_final(&host, &id).await
    } else {
        load_report(&host, &id).await
    };
    match report {
        Some(report) if report.phase.is_final() => Json(report).into_response(),
        Some(report) => (StatusCode::ACCEPTED, Json(report)).into_response(),
        None => error_response(StatusCode::NOT_FOUND, "no such compaction", None),
    }
}

/// `GET /admin/epochs`: the local epoch, the previous epochs this node still keeps, and
/// an ongoing catch-up with a peer's newer epoch (`null` if none).
pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
    let current = EpochRecord::from(host.current_epoch());
    Json(json!({
        "current": current,
        "archived": host.compaction().archived_epochs(),
        "catchUp": host.compaction().catch_up_report(),
    }))
    .into_response()
}