//! `/admin/quarantine`: this node's own changes that were not replayed when it rejoined
//! a newer epoch (outside the replay window, or impossible to tell apart).
//!
//! * `GET /admin/quarantine` lists the entries.
//! * `GET /admin/quarantine/{id}` shows the writes: every changed path with the value
//!   the old document's merged view holds there (`deleted` if none).
//! * `GET /admin/quarantine/{id}/document` downloads the old document
//!   (`AutoCommit::save()`), for inspection with other tools.
//! * `POST /admin/quarantine/{id}/replay` writes them into the current epoch anyway (as
//!   fresh writes, so they win over what is there) and removes the entry.
//! * `DELETE /admin/quarantine/{id}` discards the entry.
//!
//! Quarantine lives in memory on the node that holds it.

use std::convert::Infallible;

use axum::{
    extract::{Path, State},
    http::{header, StatusCode},
    response::{IntoResponse, Response},
    Json,
};
use serde::Serialize;
use serde_json::{json, Value as JsonValue};
use tracing::info;

use super::api::error_response;
use super::delta;
use super::local::Quarantined;
use super::Host;
use crate::store;

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct Summary {
    id: String,
    /// The epoch the changes were made in.
    epoch: String,
    /// The epoch the node rejoined.
    target: String,
    created_at: u64,
    reason: String,
    paths: usize,
}

fn summary(entry: &Quarantined) -> Summary {
    Summary {
        id: entry.id.clone(),
        epoch: entry.epoch.to_string(),
        target: entry.target.to_string(),
        created_at: entry.created_at,
        reason: entry.reason.clone(),
        paths: entry.paths.len(),
    }
}

fn not_found() -> Response {
    error_response(StatusCode::NOT_FOUND, "no such quarantine entry", None)
}

pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
    Json(host.compaction().list_quarantine(summary)).into_response()
}

pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
    let Some((summary, document, paths)) = host.compaction().with_quarantined(&id, |entry| {
        (summary(entry), entry.document.clone(), entry.paths.clone())
    }) else {
        return not_found();
    };

    let writes = tokio::task::spawn_blocking(move || {
        paths
            .into_iter()
            .map(|path| match store::read_path(&document, &path) {
                Ok(Some(value)) => json!({ "path": path, "value": value }),
                Ok(None) => json!({ "path": path, "deleted": true }),
                Err(error) => json!({ "path": path, "error": error.to_string() }),
            })
            .collect::<Vec<JsonValue>>()
    })
    .await;

    match writes {
        Ok(writes) => Json(json!({ "entry": summary, "writes": writes })).into_response(),
        Err(_) => error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "reading the writes failed",
            None,
        ),
    }
}

pub(crate) async fn document<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
    let Some(mut document) = host
        .compaction()
        .with_quarantined(&id, |entry| entry.document.clone())
    else {
        return not_found();
    };

    match tokio::task::spawn_blocking(move || document.save()).await {
        Ok(bytes) => (
            [
                (header::CONTENT_TYPE, "application/octet-stream".to_owned()),
                (
                    header::CONTENT_DISPOSITION,
                    format!("attachment; filename=\"quarantine-{id}.automerge\""),
                ),
            ],
            bytes,
        )
            .into_response(),
        Err(_) => error_response(
            StatusCode::INTERNAL_SERVER_ERROR,
            "saving the document failed",
            None,
        ),
    }
}

pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
    if !host.is_node_ready().await {
        return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
    }
    let Some(entry) = host.compaction().take_quarantined(&id) else {
        return not_found();
    };

    let (stats, entry) = host
        .commit_doc(move |doc| {
            let stats = delta::replay_paths(&entry.document, doc, &entry.paths);
            Ok::<_, Infallible>((stats, entry))
        })
        .await
        .unwrap_or_else(|never| match never {});

    if stats.failed > 0 && stats.replayed == 0 {
        // Nothing went in; keep the entry for another look.
        host.compaction().restore_quarantined(entry);
        return error_response(
            StatusCode::CONFLICT,
            "none of the writes could be replayed",
            Some(json!({ "failed": stats.failed })),
        );
    }

    info!(quarantine = %id, replayed = stats.replayed, failed = stats.failed, "Replayed quarantined changes");
    Json(json!({ "replayed": stats.replayed, "failed": stats.failed })).into_response()
}

pub(crate) async fn discard<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
    match host.compaction().take_quarantined(&id) {
        Some(_) => {
            info!(quarantine = %id, "Discarded quarantined changes");
            StatusCode::NO_CONTENT.into_response()
        }
        None => not_found(),
    }
}