//! `/admin/remakes` handlers.
//!
//! * `POST /admin/remakes` proposes a remake of the current epoch (`POST
//!   /admin/compaction` is a shortcut, kept from before). 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 swarms).
//! * `GET /admin/remakes` lists all remakes.
//! * `GET /admin/remakes/{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/remakes/{id}` asks the coordinator to abort. Only possible while
//!   proposed; `409` once committed (a switch can't be undone) or already over.
//!   `DELETE /admin/remakes` does the same for the remake of the local epoch that
//!   is in progress, without having to know its ID.
//!
//! Request body for POST (all optional):
//! `{ "kind": "compact", "spec": null, "baseHeads": ["<hex>", …],
//!    "replayWindow": {"epochs": 1, "seconds": 604800},
//!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
//!    "settleSecs": 480, "switchDelaySecs": 60 }`
//!
//! `?force=true` accepts `settleSecs` below the dormancy threshold and `switchDelaySecs`
//! below one keepalive interval: for tests and debugging, not for real swarms.
//!
//! `kind` is `compact` (no spec) or `replace`, whose spec is
//! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
//! (see `kind::Replace`). Reports leave the replacement document out (`contentBytes`).

use std::time::Duration;

use automerge::ChangeHash;
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::{kind, Host};
use crate::document::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 {
    /// The remake kind; `compact` if absent.
    kind: Option<String>,
    /// Kind-specific part of the request.
    spec: Option<JsonValue>,
    /// The base: heads (hex) whose content the new document accounts for; writes after
    /// them are carried over. The current heads if absent. For a remake derived from
    /// a read, use the heads that read returned (`GET /store?heads=true`).
    base_heads: Option<Vec<String>>,
    replay_window: Option<Retention>,
    archive_retention: Option<Retention>,
    settle_secs: Option<u64>,
    /// How long nodes wait after seeing the commit before switching; one keepalive
    /// interval if absent. Shorter only with `?force=true`.
    switch_delay_secs: Option<u64>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct Report {
    id: String,
    phase: Phase,
    #[serde(flatten)]
    record: Record,
    /// Size of the kind's build input (e.g. a replacement document), which reports
    /// leave out.
    #[serde(skip_serializing_if = "Option::is_none")]
    content_bytes: Option<usize>,
    /// What happened to this remake on the node answering the request.
    #[serde(skip_serializing_if = "Option::is_none")]
    local: Option<LocalStatus>,
}

#[derive(Debug, thiserror::Error)]
enum StartError {
    #[error("remake {0} is already open for this epoch")]
    Open(String),
    #[error("base head {0} is not in this node's document")]
    UnknownBase(String),
    #[error(transparent)]
    Store(#[from] StoreError),
}

pub(super) 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, mut record: Record) -> Report {
    let content = std::mem::take(&mut record.proposal.content);
    Report {
        phase: record.phase(unix_millis(), grace(host)),
        local: host.remake().status(&id),
        content_bytes: (!content.is_null()).then(|| content.to_string().len()),
        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 remake 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> {
    // A record that was readable may be briefly unreadable or absent while concurrent
    // changes merge; give up only if that persists.
    const MAX_MISSES: u32 = 5;

    let mut changes = host.subscribe_changes();
    let mut seen = false;
    let mut misses = 0;
    loop {
        let Some(report) = load_report(host, id).await else {
            misses += 1;
            warn!(remake = %id, seen, misses, "Remake record not readable while waiting");
            if !seen || misses >= MAX_MISSES {
                return None;
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
            continue;
        };
        seen = true;
        misses = 0;
        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;
        }
    }
}

/// `POST /admin/remakes`.
pub(crate) async fn start<H: Host>(
    State(host): State<H>,
    Query(query): Query<StartQuery>,
    body: Bytes,
) -> Response {
    start_kind(host, query, body, None).await
}

/// `POST /admin/compaction`: the original endpoint, for compacting remakes only.
pub(crate) async fn start_compaction<H: Host>(
    State(host): State<H>,
    Query(query): Query<StartQuery>,
    body: Bytes,
) -> Response {
    start_kind(host, query, body, Some(kind::DEFAULT_KIND)).await
}

async fn start_kind<H: Host>(
    host: H,
    query: StartQuery,
    body: Bytes,
    only: Option<&'static str>,
) -> 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 kind_name = request.kind.as_deref().unwrap_or(kind::DEFAULT_KIND);
    if only.is_some_and(|only| only != kind_name) {
        return error_response(
            StatusCode::BAD_REQUEST,
            format!("this endpoint only starts {} remakes", kind::DEFAULT_KIND),
            None,
        );
    }
    let Some(kind) = kind::by_name(kind_name) else {
        return error_response(
            StatusCode::BAD_REQUEST,
            format!("unknown remake kind {kind_name:?}"),
            None,
        );
    };
    let prepared = match kind.propose(request.spec.as_ref().unwrap_or(&JsonValue::Null)) {
        Ok(prepared) => prepared,
        Err(reason) => {
            return error_response(
                StatusCode::BAD_REQUEST,
                format!("invalid {kind_name} spec: {reason}"),
                None,
            )
        }
    };

    let base = match &request.base_heads {
        None => None,
        Some(heads) => {
            let mut parsed = Vec::with_capacity(heads.len());
            for head in heads {
                let Some(bytes) = records::decode_hex32(head.trim()) else {
                    return error_response(
                        StatusCode::BAD_REQUEST,
                        format!("invalid base head {head:?}"),
                        None,
                    );
                };
                parsed.push(ChangeHash(bytes));
            }
            Some(parsed)
        }
    };

    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,
        );
    }

    // The delay lets the commit reach every node through ordinary sync before any of
    // them stops syncing the old epoch.
    let keepalive = dormant_after / 4;
    let switch_delay = request.switch_delay_secs.map(Duration::from_secs);
    match switch_delay {
        Some(delay) if delay < keepalive && query.force => {
            warn!(
                switch_delay_secs = delay.as_secs(),
                keepalive_secs = keepalive.as_secs(),
                "Forced switch delay shorter than a keepalive interval; slow nodes may have to rejoin"
            );
        }
        Some(delay) if delay < keepalive => {
            return error_response(
                StatusCode::BAD_REQUEST,
                format!(
                    "switchDelaySecs must be at least {} (one keepalive interval), so the commit reaches every node before they switch; add ?force=true to override",
                    keepalive.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 remake; 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 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,
        switch_delay_ms: switch_delay.map(|delay| delay.as_millis() as u64),
        peers_considered,
        kind: kind.name().to_owned(),
        content: prepared.content,
        translation: prepared.translation,
    };

    // 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,
        base,
        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 remake of this epoch is still open",
                Some(json!({ "remake": open })),
            );
        }
        Err(error @ StartError::UnknownBase(_)) => {
            return error_response(StatusCode::BAD_REQUEST, error.to_string(), None);
        }
        Err(StartError::Store(error)) => {
            error!(error = %format!("{error:#}"), "Failed to write remake proposal");
            return error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to write the proposal",
                None,
            );
        }
    };

    info!(
        remake = %id,
        settle_secs = settle.as_secs(),
        "Remake proposed"
    );

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

    let location = format!("/admin/remakes/{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,
    base: Option<Vec<ChangeHash>>,
    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 the requested heads, or everything before the proposal (the
            // proposal itself is never in it).
            let base = match base {
                Some(base) => {
                    if let Some(unknown) = base
                        .iter()
                        .find(|head| doc.get_change_by_hash(head).is_none())
                    {
                        return Err(StartError::UnknownBase(unknown.to_string()));
                    }
                    base
                }
                None => doc.get_heads(),
            };
            let mut heads: Vec<String> = base.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 remake", None),
    }
}

/// `DELETE /admin/remakes/{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/remakes`: aborts whichever remake 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 remake 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 remake", None);
    };

    match current.phase {
        Phase::Proposed => {}
        Phase::Committed => {
            return error_response(
                StatusCode::CONFLICT,
                "already committed; the switch cannot be undone",
                Some(json!({ "remake": current })),
            )
        }
        Phase::Aborted | Phase::Expired => {
            return error_response(
                StatusCode::CONFLICT,
                "already over",
                Some(json!({ "remake": 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!(remake = %id, error = %format!("{error:#}"), "Failed to write abort request");
            return error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to record the abort request",
                None,
            );
        }
        info!(remake = %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 remake", 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.remake().archived_epochs(),
        "catchUp": host.remake().catch_up_report(),
        "quarantine": host.remake().list_quarantine(|entry| entry.id.clone()),
    }))
    .into_response()
}