//! Swarm-wide forget records (`_continuum/forgotten/<pk>`): a peer forgotten on one
//! node is forgotten by all; a forgotten node that is alive cancels its record.

use automerge::AutoCommit;
use tokio::time::Instant;
use tracing::{info, warn};

use crate::document::meta;
use crate::document::store::{self, StoreError};
use crate::state::AppState;

fn forgotten_dir() -> Vec<String> {
    meta::path(&[meta::FORGOTTEN])
}

pub(crate) fn forgotten_path(pk: &str) -> Vec<String> {
    let mut path = forgotten_dir();
    path.push(pk.to_owned());
    path
}

fn forgotten_peers(doc: &AutoCommit) -> Vec<String> {
    store::keys_at(doc, &forgotten_dir()).unwrap_or_default()
}

pub(super) async fn apply_forget_records(state: &AppState) {
    // Before the initial sync we can't judge the records, and we have nothing to delete.
    if !state.is_ready().await {
        return;
    }

    let forgotten = forgotten_peers(&state.doc.read().await.doc);
    if forgotten.is_empty() {
        return;
    }

    let now = Instant::now();
    let mut cancel = Vec::new();

    for pk in forgotten {
        if pk == state.local_id() {
            cancel.push(pk); // we are evidently alive
            continue;
        }

        let Some(peer_state_mutex) = state.get_peer_state(&pk).await else {
            continue; // already forgotten here
        };

        let alive = peer_state_mutex.lock().await.is_alive(&state.timing, now);

        if alive {
            cancel.push(pk);
        } else if state.peer_states.write().await.remove(&pk).is_some() {
            info!(peer = %pk, "Forgetting peer per swarm-wide record");
        }
    }

    if cancel.is_empty() {
        return;
    }

    // Individual failures are logged inside; NotFound means another node already
    // cancelled the record.
    let _ = state
        .mutate_doc(move |doc| {
            for pk in &cancel {
                match store::delete_path(&mut *doc, &forgotten_path(pk), false) {
                    Ok(()) => info!(peer = %pk, "Peer is alive; cancelling swarm-wide forget"),
                    Err(StoreError::NotFound) => {}
                    Err(error) => {
                        warn!(peer = %pk, error = %error, "Failed to cancel forget record")
                    }
                }
            }
            Ok::<(), StoreError>(())
        })
        .await;
}