//! Switching this node to a committed epoch.
//!
//! The worker watches the compaction records. When a compaction 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::local::LocalStatus;
use super::records::{self, unix_millis, EpochRecord, Record};
use super::snapshot::{self, COMPACTIONS, META_ROOT};
use super::Host;
use crate::replica::Epoch;
use crate::store;
use crate::sync::paths_related;

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

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

    let records_dir = vec![META_ROOT.to_owned(), COMPACTIONS.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; compaction 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.compaction().prune_archive(local.generation);

    // While rejoining, the document is being filled from peers: its records are not
    // complete enough to act on.
    if host.compaction().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!(compactions = ?ids, "Several compactions committed for this epoch; not switching on my own");
            return catchup::check(host).await;
        }
    };

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

    let switch_delay = host.dormant_after() / 4; // one keepalive interval
    if host.compaction().first_seen(&id).elapsed() < switch_delay {
        return;
    }

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

async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> {
    let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id))
    else {
        return Err("unreadable compaction record".to_owned());
    };
    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!(compaction = %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
        .compaction()
        .take_prebuilt(id)
        .filter(|snapshot| snapshot.epoch == expected);

    // 2. Build (or reuse), verify and replay without holding any lock.
    let (new_doc, copy_heads, 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 compaction'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 compaction record into the new epoch");
                }
            }

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

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

    // 4. Keep the old document as configured.
    host.compaction()
        .archive(previous, old, retention, expected.generation);

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

/// 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>,
) -> Result<AutoCommit, String> {
    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);
        if late.replayed + late.failed > 0 {
            info!(
                replayed = late.replayed,
                failed = late.failed,
                "Replayed late changes"
            );
        }
        Ok(new_doc)
    })
    .await
}