//! `/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. The remake
//!   commits as soon as a majority of its electorate voted for it (`vote`), and aborts
//!   once that is out of reach or `settleSecs` passed without it.
//! * `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
//!   undecided; `409` once committed (a switch can't be undone) or aborted. An expired
//!   remake is aborted right away by the node asked: its coordinator is gone, and
//!   can't commit it any more anyway (`coordinator`). `DELETE /admin/remakes` does the
//!   same for the undecided remake of the local epoch, 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=` overrides checks: `true` all of them, or a comma-separated list of
//! * `switchDelaySecs`: accept one below a keepalive interval (for tests and debugging,
//!   not for real swarms);
//! * `supersede`: abort expired remakes of this epoch and propose in their place.
//!   Without it, an expired remake still holding votes answers `409`.
//!
//! `kind` is `compact` (no spec), `replace`, whose spec is
//! `{ "document": {…store root…}, "rules": [{"from": "a", "to": "b"}, {"from": "c", "drop": true}] }`
//! (see `kind::Replace`), or `import`, whose spec is
//! `{ "document": "<base64 of an Automerge document>", "rules": […] }` (see
//! `kind::Import`; `GET /admin/document` exports one). The document stays on the
//! coordinator; the proposal refers to it by hash (`proposal.content`), and every voter
//! downloads it (`content`). Requests may be up to 64 MiB.
//!
//! An import can also be uploaded as is, with `Content-Type: application/octet-stream`;
//! then `rules` (JSON), `baseHeads` (comma-separated), `settleSecs` and
//! `switchDelaySecs` go in the query. Restoring a backup, keeping writes made since at
//! their paths:
//! `curl --data-binary @backup.automerge -H 'Content-Type: application/octet-stream' --url-query 'rules=[{"from":"","to":""}]' …/admin/remakes`

use std::collections::HashSet;
use std::time::Duration;

use automerge::ChangeHash;
use axum::{
    Json,
    body::Bytes,
    extract::{Path, Query, State},
    http::{
        HeaderMap, StatusCode,
        header::{CONTENT_TYPE, LOCATION},
    },
    response::{IntoResponse, Response},
};
use base64::engine::{Engine, general_purpose::STANDARD as BASE64};
use serde::{Deserialize, Serialize};
use serde_json::{Value as JsonValue, json};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
use uuid::Uuid;

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

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

#[derive(Debug, Default, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct StartQuery {
    #[serde(default)]
    wait: bool,
    /// The checks to override (`Force`).
    force: Option<String>,
    // Only for an import uploaded as `application/octet-stream`, which has no JSON
    // body to carry them:
    /// The rules, as JSON (as in the spec).
    rules: Option<String>,
    /// Comma-separated hex.
    base_heads: Option<String>,
    settle_secs: Option<u64>,
    switch_delay_secs: Option<u64>,
}

impl StartQuery {
    fn has_request_fields(&self) -> bool {
        self.rules.is_some()
            || self.base_heads.is_some()
            || self.settle_secs.is_some()
            || self.switch_delay_secs.is_some()
    }

    /// An import request for a document uploaded as is.
    fn import_request(&self, document: &[u8]) -> Result<StartRequest, String> {
        let rules = match &self.rules {
            None => JsonValue::Array(Vec::new()),
            Some(rules) => serde_json::from_str(rules)
                .map_err(|error| format!("rules are not valid JSON: {error}"))?,
        };
        Ok(StartRequest {
            kind: Some("import".to_owned()),
            spec: Some(json!({
                "document": BASE64.encode(document),
                "rules": rules,
            })),
            base_heads: self.base_heads.as_ref().map(|heads| {
                heads
                    .split(',')
                    .filter(|head| !head.trim().is_empty())
                    .map(str::to_owned)
                    .collect()
            }),
            settle_secs: self.settle_secs,
            switch_delay_secs: self.switch_delay_secs,
            ..StartRequest::default()
        })
    }
}

fn is_octet_stream(headers: &HeaderMap) -> bool {
    headers
        .get(CONTENT_TYPE)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.split(';').next())
        .is_some_and(|media| media.trim() == "application/octet-stream")
}

#[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>,
    /// How long votes are collected; twice the dormancy threshold if absent.
    settle_secs: Option<u64>,
    /// How long nodes wait after seeing the commit before switching; one keepalive
    /// interval if absent. Shorter only with `?force=switchDelaySecs`.
    switch_delay_secs: Option<u64>,
}

/// The checks `?force=` overrides: `true` all of them, or a comma-separated list.
#[derive(Debug, Default, Clone, Copy)]
struct Force {
    switch_delay: bool,
    supersede: bool,
}

impl Force {
    fn parse(value: Option<&str>) -> Result<Self, String> {
        let mut force = Force::default();
        match value.map(str::trim) {
            None | Some("" | "false") => {}
            Some("true") => {
                force = Force {
                    switch_delay: true,
                    supersede: true,
                };
            }
            Some(list) => {
                for name in list.split(',').map(str::trim) {
                    match name {
                        "switchDelaySecs" => force.switch_delay = true,
                        "supersede" => force.supersede = true,
                        other => {
                            return Err(format!(
                                "unknown force {other:?}: use true, or a list of \
                                 switchDelaySecs and supersede"
                            ));
                        }
                    }
                }
            }
        }
        Ok(force)
    }
}

/// An expired remake of the epoch a proposal would supersede.
#[derive(Debug)]
struct ExpiredRemake {
    id: String,
    coordinator: String,
    /// When it expired (Unix milliseconds).
    since: u64,
    tally: vote::Tally,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct Report {
    id: String,
    phase: Phase,
    #[serde(flatten)]
    record: Record,
    /// 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),
    /// Expired remakes still hold votes; superseding them takes `force=supersede`.
    #[error("expired remakes of this epoch still hold votes")]
    Expired(Vec<ExpiredRemake>),
    /// The proposer votes for its own remake, so it must be free to.
    #[error("this node can't vote for a new remake: {0}")]
    VoteHeld(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, record: Record) -> Report {
    Report {
        phase: record.phase(unix_millis(), grace(host)),
        local: host.remake().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 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>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    start_kind(host, query, &headers, 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>,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    start_kind(host, query, &headers, body, Some(kind::DEFAULT_KIND)).await
}

async fn start_kind<H: Host>(
    host: H,
    query: StartQuery,
    headers: &HeaderMap,
    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 force = match Force::parse(query.force.as_deref()) {
        Ok(force) => force,
        Err(reason) => return error_response(StatusCode::BAD_REQUEST, reason, None),
    };

    let request: StartRequest = if is_octet_stream(headers) {
        match query.import_request(&body) {
            Ok(request) => request,
            Err(reason) => {
                return error_response(
                    StatusCode::BAD_REQUEST,
                    format!("invalid request: {reason}"),
                    None,
                );
            }
        }
    } else if query.has_request_fields() {
        return error_response(
            StatusCode::BAD_REQUEST,
            "rules, baseHeads, settleSecs and switchDelaySecs go in the query only for an application/octet-stream import; put them in the JSON body",
            None,
        );
    } else 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)
        }
    };

    // The voting period. No lower bound: the majority protects against competitors,
    // a short period only risks an abort for lack of votes.
    let dormant_after = host.dormant_after();
    let settle = request
        .settle_secs
        .map(Duration::from_secs)
        .unwrap_or(dormant_after * 2);

    // 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 && force.switch_delay => {
            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=switchDelaySecs to override",
                    keepalive.as_secs()
                ),
                None,
            );
        }
        _ => {}
    }

    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,
        switch_delay_ms: switch_delay.map(|delay| delay.as_millis() as u64),
        // Filled in when the proposal is written, from the document at that moment.
        electorate: Vec::new(),
        quorum: 0,
        kind: kind.name().to_owned(),
        content: None,
        translation: prepared.translation,
    };
    // The content stays here, referred to by hash; voters fetch it from us.
    if let Some(bytes) = prepared.content {
        let reference = ContentRef::of(&bytes);
        if let Some(hash) = reference.hash() {
            host.remake().put_content(hash, std::sync::Arc::new(bytes));
        }
        proposal.content = Some(reference);
    }

    // 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,
        force.supersede,
        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(StartError::Expired(expired)) => {
            let active: HashSet<String> = host
                .peer_summaries()
                .await
                .into_iter()
                .filter(|peer| peer.active)
                .map(|peer| peer.id)
                .collect();
            let me = host.node_id();
            let expired: Vec<JsonValue> = expired
                .iter()
                .map(|remake| {
                    json!({
                        "id": remake.id,
                        "coordinator": remake.coordinator,
                        "coordinatorActive": remake.coordinator == me
                            || active.contains(&remake.coordinator),
                        "expiredAt": remake.since,
                        "tally": remake.tally.to_string(),
                    })
                })
                .collect();
            return error_response(
                StatusCode::CONFLICT,
                "an expired remake of this epoch still holds votes: abort it, or add \
                 ?force=supersede to abort it and propose in its place",
                Some(json!({ "expired": expired, "force": "supersede" })),
            );
        }
        Err(error @ StartError::VoteHeld(_)) => {
            return error_response(StatusCode::CONFLICT, error.to_string(), None);
        }
        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>>,
    supersede: bool,
    now: u64,
    grace: u64,
) -> Result<Record, StartError> {
    let record_id = id.clone();
    let electorate_host = host.clone();
    let me = host.node_id();
    let started = host.remake().started_at() + grace;
    let record = host
        .commit_doc(move |doc| -> Result<Record, StartError> {
            let mut all = records::load_all(doc);
            if let Some((open, _)) = all.iter().find(|(_, record)| {
                record.proposal.from_epoch == proposal.from_epoch
                    && record.phase(now, grace).is_open()
            }) {
                return Err(StartError::Open(open.clone()));
            }
            let mut register = records::load_register(doc);

            // Expired remakes hold their votes; their coordinators can't commit them
            // any more, so they may be aborted here.
            let expired: Vec<&String> = all
                .iter()
                .filter(|(_, record)| {
                    record.proposal.from_epoch == proposal.from_epoch
                        && record.phase(now, grace) == Phase::Expired
                })
                .map(|(id, _)| id)
                .collect();
            if !expired.is_empty() {
                if !supersede {
                    let details = all
                        .iter()
                        .filter(|(id, _)| expired.contains(&id))
                        .map(|(id, record)| ExpiredRemake {
                            id: id.clone(),
                            coordinator: record.proposal.coordinator.clone(),
                            since: record.proposal.settle_until.saturating_add(grace),
                            tally: vote::tally(id, record, &all, &register),
                        })
                        .collect();
                    return Err(StartError::Expired(details));
                }
                let decision = Decision::Aborted {
                    reason: format!("superseded by remake {record_id}"),
                    decided_at: now,
                };
                for id in expired {
                    records::write_decision(doc, id, &decision)?;
                    info!(remake = %id, by = %record_id, "Superseded expired remake");
                }
                all = records::load_all(doc);
                register = records::load_register(doc);
            }

            match vote::freedom(&me, &proposal.from_epoch, &all, &register, started) {
                vote::Freedom::Any => {}
                vote::Freedom::Only(id) => {
                    return Err(StartError::VoteHeld(format!(
                        "it may have voted for remake {id} before its start"
                    )));
                }
                vote::Freedom::Held(reason) => return Err(StartError::VoteHeld(reason)),
            }

            // 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;

            // We vote for our own remake, so we are always in the electorate.
            let mut electorate = electorate_host.electorate(doc, now);
            if let Err(position) = electorate.binary_search(&me) {
                electorate.insert(position, me.clone());
            }
            proposal.quorum = vote::quorum(electorate.len());
            proposal.electorate = electorate;

            records::write_proposal(doc, &record_id, &proposal)?;
            let own_vote = Vote {
                remake: record_id.clone(),
                at: now,
            };
            records::write_vote(doc, &me, &own_vote)?;
            Ok(Record {
                proposal,
                decision: None,
                abort_requested: None,
                contested: false,
                votes: [(me, own_vote)].into(),
            })
        })
        .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
/// undecided (proposed or expired), the oldest if there are several.
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 open = host
        .read_doc(move |doc| {
            records::load_all(doc)
                .into_iter()
                .filter(|(_, record)| {
                    record.proposal.from_epoch == local && record.decision.is_none()
                })
                .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::Expired => return abort_expired(&host, &id).await,
        Phase::Committed => {
            return error_response(
                StatusCode::CONFLICT,
                "already committed; the switch cannot be undone",
                Some(json!({ "remake": current })),
            );
        }
        Phase::Aborted => {
            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, and can then be aborted right away.
    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),
    }
}

/// Aborts expired remake `id` from this node: its coordinator is gone, and can't commit
/// it any more (`coordinator`). Answers with the final report.
async fn abort_expired<H: Host>(host: &H, id: &str) -> Response {
    let record_id = id.to_owned();
    let decision = Decision::Aborted {
        reason: format!(
            "aborted by operator via {} after it expired",
            host.node_id()
        ),
        decided_at: unix_millis(),
    };
    let grace = grace(host);
    let written = host
        .commit_doc(move |doc| -> Result<bool, StoreError> {
            let expired = records::load(doc, &record_id)
                .is_some_and(|record| record.phase(unix_millis(), grace) == Phase::Expired);
            if expired {
                records::write_decision(doc, &record_id, &decision)?;
            }
            Ok(expired)
        })
        .await;
    match written {
        Ok(true) => info!(remake = %id, "Aborted expired remake"),
        // Decided meanwhile: the report says how.
        Ok(false) => {}
        Err(error) => {
            error!(remake = %id, error = %format!("{error:#}"), "Failed to abort expired remake");
            return error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                "failed to record the abort",
                None,
            );
        }
    }
    match load_report(host, id).await {
        Some(report) => 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()
}