//! The coordinator: the node that received the proposal watches the cluster during the
//! settle period and then writes the decision.
//!
//! It aborts if, at any point before deciding:
//! * an operator requested an abort,
//! * a competing proposal for the same epoch appeared (both coordinators abort),
//! * a peer it considered went dormant (a partition may hide a competitor),
//! * the local epoch changed, or
//! * the decision would come later than the settle period plus the grace margin
//!   (other nodes report the proposal as expired by then).
//!
//! Otherwise it builds the snapshot at the base heads and commits, recording the new
//! epoch so every node can verify its own build against it.
//!
//! The task lives only in memory. If the coordinator dies, nobody decides and the
//! proposal expires.

use std::collections::HashMap;
use std::time::Duration;

use tracing::{info, warn};

use super::records::{self, Decision, Record};
use super::snapshot;
use super::Host;
use crate::store::StoreError;

const TICK: Duration = Duration::from_secs(1);

pub(crate) async fn run<H: Host>(host: H, id: String) {
    info!(compaction = %id, "Coordinating compaction");

    loop {
        tokio::time::sleep(TICK).await;

        match step(&host, &id).await {
            Ok(true) => return,
            Ok(false) => {}
            Err(error) => {
                warn!(compaction = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying");
            }
        }
    }
}

/// One evaluation. Returns true once the coordinator is done.
async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
    let wanted = id.to_owned();
    let (record, all) = host
        .read_doc(move |doc| (records::load(doc, &wanted), records::load_all(doc)))
        .await;

    let Some(record) = record else {
        warn!(compaction = %id, "Compaction record vanished; stopping coordinator");
        return Ok(true);
    };
    if record.decision.is_some() {
        return Ok(true);
    }

    let now = records::unix_millis();
    let grace = host.dormant_after().as_millis() as u64;

    if let Some(reason) = abort_reason(host, id, &record, &all, now, grace).await {
        decide_abort(host, id, reason).await?;
        return Ok(true);
    }

    if now < record.proposal.settle_until {
        return Ok(false);
    }

    commit(host, id, &record).await?;
    Ok(true)
}

async fn abort_reason<H: Host>(
    host: &H,
    id: &str,
    record: &Record,
    all: &[(String, Record)],
    now: u64,
    grace: u64,
) -> Option<String> {
    let proposal = &record.proposal;

    if let Some(request) = &record.abort_requested {
        return Some(format!("aborted by operator via {}", request.by));
    }

    if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch {
        return Some("local epoch changed during settle".to_owned());
    }

    if let Some((other, _)) = all.iter().find(|(other, candidate)| {
        other != id
            && candidate.proposal.from_epoch == proposal.from_epoch
            && candidate.phase(now, grace).is_open()
    }) {
        return Some(format!("competing compaction {other} for the same epoch"));
    }

    // Peers forgotten in the meantime are ignored: an operator decided they are gone.
    let peers: HashMap<String, bool> = host
        .peer_summaries()
        .await
        .into_iter()
        .map(|peer| (peer.id, peer.active))
        .collect();
    let dormant: Vec<&str> = proposal
        .peers_considered
        .iter()
        .filter(|peer| peers.get(peer.as_str()) == Some(&false))
        .map(String::as_str)
        .collect();
    if !dormant.is_empty() {
        return Some(format!(
            "peers went dormant during settle: {}",
            dormant.join(", ")
        ));
    }

    if now > proposal.settle_until.saturating_add(grace) {
        return Some("decision deadline passed".to_owned());
    }

    None
}

async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> {
    let Some(request) = record.snapshot_request(id) else {
        return decide_abort(
            host,
            id,
            "unreadable epoch or base heads in proposal".into(),
        )
        .await;
    };

    // Clone under the read lock, build without any lock.
    let copy = host.read_doc(|doc| doc.clone()).await;
    let mut snapshot = match snapshot::build_detached(copy, request).await {
        Ok((snapshot, _copy)) => snapshot,
        Err(error) => {
            return decide_abort(host, id, format!("snapshot build failed: {error}")).await;
        }
    };

    let to_epoch = snapshot.epoch;
    let snapshot_bytes = snapshot.doc.save().len() as u64;

    let decision = Decision::Committed {
        to_epoch: to_epoch.into(),
        snapshot_bytes,
        decided_at: records::unix_millis(),
    };
    write_decision(host, id, decision).await?;

    // The local switch reuses it instead of building again.
    host.compaction().store_prebuilt(id, snapshot);

    info!(compaction = %id, to_epoch = %to_epoch, snapshot_bytes, "Compaction committed");
    Ok(())
}

async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> {
    warn!(compaction = %id, reason = %reason, "Compaction aborted");
    write_decision(
        host,
        id,
        Decision::Aborted {
            reason,
            decided_at: records::unix_millis(),
        },
    )
    .await
}

async fn write_decision<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> {
    let id = id.to_owned();
    host.commit_doc(move |doc| records::write_decision(doc, &id, &decision))
        .await
}