//! 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. And
//! locally, strangers that went silent.

use std::time::Duration;

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, Standing};

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

/// How long a peer that never became a member is kept once nothing more is heard from
/// it: one listed to us that never answered, or one that keeps failing the handshake
/// (another swarm, a wrong key). Anyone can name peers to a node, so without a limit
/// the map of peers would only grow.
const STRANGER_KEPT: Duration = Duration::from_secs(24 * 3600);

/// Forgets peers that never became members and were silent for `STRANGER_KEPT`. Only
/// here, not swarm-wide; one that comes back is a stranger again.
pub(super) async fn forget_strangers(state: &AppState) {
    let now = Instant::now();
    let mut strangers = Vec::new();
    for (peer, peer_state) in state.peers_snapshot().await {
        let peer_state = peer_state.lock().await;
        let quiet_since = peer_state.last_seen.unwrap_or(peer_state.added_at);
        if !matches!(peer_state.standing, Standing::Member { .. })
            && now.saturating_duration_since(quiet_since) >= STRANGER_KEPT
        {
            strangers.push(peer);
        }
    }
    if strangers.is_empty() {
        return;
    }
    let mut peer_states = state.peer_states.write().await;
    for peer in &strangers {
        peer_states.remove(peer);
    }
    info!(
        forgotten = strangers.len(),
        "Forgot peers that never became members and were silent for a day"
    );
}