//! Switching this node to a committed epoch.
//!
//! The worker watches the remake records. When a remake of the local epoch is
//! committed, it waits one keepalive interval (so the commit record can reach peers
//! through ordinary sync before this node stops syncing the old epoch), then:
//!
//! 1. clones the live document under the read lock;
//! 2. without any lock: takes the coordinator's prebuilt snapshot or builds it, checks
//!    that its hash is the committed epoch ID, and replays the clone's changes since
//!    the base into it;
//! 3. under the write lock: replays whatever arrived since the clone, swaps document and
//!    epoch, and hands back the old document;
//! 4. archives the old document according to the policy.
//!
//! The write lock is held only for step 3, which is proportional to the changes made
//! during steps 1-2.
//!
//! A node whose build doesn't match the committed epoch, or that never saw the commit,
//! rejoins the new epoch from peers instead (`catchup`).

use std::time::Duration;

use automerge::{ActorId, AutoCommit, ChangeHash};
use tokio::sync::broadcast;
use tracing::{error, info, warn};

use super::catchup;
use super::delta::{self, ReplayStats};
use super::kind::Translator;
use super::local::LocalStatus;
use super::quarantine;
use super::records::{self, unix_millis, EpochRecord, Record};
use super::snapshot::{self, META_ROOT, REMAKES};
use super::Host;
use crate::document::replica::Epoch;
use crate::document::store;
use crate::swarm::sync::paths_related;

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

pub(crate) async fn run_worker<H: Host>(host: H) {
    info!("Starting remake worker");

    let records_dir = vec![META_ROOT.to_owned(), REMAKES.to_owned()];
    let mut changes = host.subscribe_changes();
    let mut changes_closed = false;
    let mut tick = tokio::time::interval(TICK);
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        // React promptly to changes of the records; the timer covers the switch delay.
        let should_check = tokio::select! {
            _ = tick.tick() => true,
            received = changes.recv(), if !changes_closed => match received {
                Ok(set) => set.paths.iter().any(|path| paths_related(path, &records_dir)),
                Err(broadcast::error::RecvError::Lagged(_)) => true,
                Err(broadcast::error::RecvError::Closed) => {
                    warn!("Change events closed; remake worker continues on its timer only");
                    changes_closed = true;
                    false
                }
            }
        };

        if should_check {
            check(&host).await;
        }
    }
}

async fn check<H: Host>(host: &H) {
    let local = host.current_epoch();
    host.remake().prune_archive(local.generation);

    // While rejoining, the document is being filled from peers: its records are not
    // complete enough to act on.
    if host.remake().is_rejoining() {
        return catchup::check(host).await;
    }

    let all = host.read_doc(records::load_all).await;
    let local_record = EpochRecord::from(local);

    let committed: Vec<(String, Record)> = all
        .into_iter()
        .filter(|(_, record)| {
            record.proposal.from_epoch == local_record && record.committed_epoch().is_some()
        })
        .collect();

    let (id, record) = match committed.as_slice() {
        // Nothing committed here; maybe peers moved on without us.
        [] => return catchup::check(host).await,
        [(id, record)] => (id.clone(), record.clone()),
        several => {
            // The settle protocol prevents this; never guess which one is right. Peers
            // that did switch can still be followed.
            let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect();
            error!(remakes = ?ids, "Several remakes committed for this epoch; not switching on my own");
            return catchup::check(host).await;
        }
    };

    if host
        .remake()
        .status(&id)
        .is_some_and(|status| status.error.is_some())
    {
        // Our own switch failed (reported in the remake's status): fetch the epoch
        // from a peer that made it.
        return catchup::check(host).await;
    }

    // As proposed, or one keepalive interval.
    let switch_delay = record
        .proposal
        .switch_delay_ms
        .map(Duration::from_millis)
        .unwrap_or(host.dormant_after() / 4);
    if host.remake().first_seen(&id).elapsed() < switch_delay {
        return;
    }

    match switch(host, &id, &record).await {
        Ok((stats, quarantine)) => {
            host.remake().set_status(
                &id,
                LocalStatus {
                    switched_at: Some(unix_millis()),
                    replayed_paths: Some(stats.replayed),
                    failed_paths: Some(stats.failed),
                    quarantine,
                    error: None,
                },
            );
        }
        Err(reason) => {
            error!(remake = %id, reason = %reason, "Epoch switch failed");
            host.remake().set_status(
                &id,
                LocalStatus {
                    error: Some(reason),
                    ..LocalStatus::default()
                },
            );
        }
    }
}

/// Returns the replay numbers and the quarantine entry for writes the remake had no
/// place for, if any.
async fn switch<H: Host>(
    host: &H,
    id: &str,
    record: &Record,
) -> Result<(ReplayStats, Option<String>), String> {
    let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id))
    else {
        return Err("unreadable remake record".to_owned());
    };
    let translator = request
        .translator()
        .ok_or_else(|| format!("unknown remake kind {:?}", request.kind))?;
    let late_translator = translator.clone();
    let previous = request.previous;
    let base = request.base_heads.clone();
    let retention = record.proposal.policy.archive_retention.clone();
    // Only the coordinator carries the record into the new epoch: one writer, so the
    // nodes don't all write the same path concurrently right after switching (a merge
    // conflict, plus one change per node to sync). The epoch metadata in the snapshot
    // keeps the essentials on every node anyway.
    let carry_record = record.proposal.coordinator == host.node_id();
    let record_path = records::record_path(id, "");

    info!(remake = %id, from = %previous, to = %expected, "Switching epoch");

    // 1. Consistent copy, under the read lock only.
    let copy = host.read_doc(|doc| doc.clone()).await;
    let prebuilt = host
        .remake()
        .take_prebuilt(id)
        .filter(|snapshot| snapshot.epoch == expected);

    // 2. Build (or reuse), verify and replay without holding any lock.
    let (new_doc, copy_heads, mut stats) = tokio::task::spawn_blocking(
        move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
            let mut copy = copy;
            let snapshot = match prebuilt {
                Some(snapshot) => snapshot,
                None => snapshot::build(&mut copy, &request.params())
                    .map_err(|error| format!("snapshot build failed: {error}"))?,
            };
            if snapshot.epoch != expected {
                return Err(format!(
                    "built snapshot {} does not match the committed epoch {expected}",
                    snapshot.epoch
                ));
            }

            // A fresh actor: this node's writes in the new epoch.
            let mut doc = snapshot.doc.with_actor(ActorId::random());

            // Keep this remake's record visible in the new epoch, for its history.
            if carry_record {
                if let Err(error) = store::copy_path(&copy, &mut doc, &record_path) {
                    warn!(error = %error, "Could not carry the remake record into the new epoch");
                }
            }

            let heads = copy.get_heads();
            let stats = delta::replay(&mut copy, &base, &heads, &mut doc, &translator);
            Ok((doc, heads, stats))
        },
    )
    .await
    .map_err(|error| format!("switch task failed: {error}"))??;

    // 3. Swap, replaying what arrived since the copy.
    let (old, late) = swap_in(host, expected, new_doc, copy_heads, late_translator).await?;
    stats.absorb(late);

    // 4. Writes the remake had no place for wait for an operator (the old document
    //    holds them); the old document is kept as configured either way.
    let quarantine = quarantine::add(
        host,
        previous,
        expected,
        &old,
        std::mem::take(&mut stats.quarantined),
    );
    host.remake()
        .archive(previous, old, retention, expected.generation);

    info!(
        remake = %id,
        epoch = %expected,
        replayed = stats.replayed,
        failed = stats.failed,
        dropped = stats.dropped,
        quarantine = ?quarantine,
        "Switched to the new epoch"
    );
    Ok((stats, quarantine))
}

/// Under the write lock: replays into `new_doc` whatever the live document received
/// since `copy_heads` (the heads of the copy `new_doc` was prepared from), then swaps it
/// in as `target`. Returns the old document.
pub(super) async fn swap_in<H: Host>(
    host: &H,
    target: Epoch,
    new_doc: AutoCommit,
    copy_heads: Vec<ChangeHash>,
    translator: Translator,
) -> Result<(AutoCommit, ReplayStats), String> {
    // The closure runs on the blocking pool; its replay numbers come back through this.
    let late_stats = std::sync::Arc::new(std::sync::Mutex::new(ReplayStats::default()));
    let late_out = std::sync::Arc::clone(&late_stats);
    let old = host
        .switch_epoch(target, move |live| {
            // Only one worker switches, but be sure the copy still describes the live document.
            if !copy_heads
                .iter()
                .all(|head| live.get_change_by_hash(head).is_some())
            {
                return Err("the document was replaced while preparing the switch".to_owned());
            }
            let mut new_doc = new_doc;
            let live_heads = live.get_heads();
            let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc, &translator);
            if late.replayed + late.failed > 0 {
                info!(
                    replayed = late.replayed,
                    failed = late.failed,
                    "Replayed late changes"
                );
            }
            *late_out
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner()) = late;
            Ok(new_doc)
        })
        .await?;
    let late = std::mem::take(
        &mut *late_stats
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()),
    );
    Ok((old, late))
}