//! 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::collections::HashMap;
use std::time::Duration;

use automerge::AutoCommit;
use serde_json::Value as JsonValue;
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()
}

/// The forget records this node acted on: peer → the record as it read then.
pub(super) type AppliedForgets = HashMap<String, JsonValue>;

/// Acts on each forget record once (`applied`). A peer forgotten here may be heard of
/// again before its cancel arrives (named in a peer list, or alive again), and is kept
/// then; forgetting it again every check would only make it flap. A new record (another
/// `DELETE`) is acted on again.
pub(super) async fn apply_forget_records(state: &AppState, applied: &mut AppliedForgets) {
    // Before the initial sync we can't judge the records, and we have nothing to delete.
    if !state.is_ready().await {
        return;
    }

    let forgotten: Vec<(String, JsonValue)> = {
        let replica = state.doc.read().await;
        forgotten_peers(&replica.doc)
            .into_iter()
            .filter_map(|pk| {
                let record = store::read_path(&replica.doc, &forgotten_path(&pk)).ok()??;
                Some((pk, record))
            })
            .collect()
    };
    // Cancelled records are done with.
    applied.retain(|pk, _| forgotten.iter().any(|(forgotten, _)| forgotten == pk));
    if forgotten.is_empty() {
        return;
    }

    let now = Instant::now();
    let mut cancel = false;

    for (pk, record) in forgotten {
        if pk == state.local_id() {
            cancel = true; // we are evidently alive
            continue;
        }
        if applied.get(&pk) == Some(&record) {
            continue;
        }

        // An alive peer is kept, and the record looked at again next time. Only the
        // peer itself cancels its record: it gets the record from whoever it is alive
        // to, and one cancel is one change instead of one from every node that hears
        // from it. Unknown here counts as acted on: one heard of later is kept.
        if state.get_peer_state(&pk).await.is_none() {
            applied.insert(pk, record);
        } else if forget_unless_alive(state, &pk, now).await {
            info!(peer = %pk, "Forgetting peer per swarm-wide record");
            applied.insert(pk, record);
        }
    }

    if !cancel {
        return;
    }

    // NotFound: gone since we looked.
    let local_id = state.local_id().to_owned();
    let _ = state
        .mutate_doc(move |doc| {
            match store::delete_path(&mut *doc, &forgotten_path(&local_id), false) {
                Ok(()) => {
                    info!("We were forgotten swarm-wide but are alive; cancelling the record")
                }
                Err(StoreError::NotFound) => {}
                Err(error) => warn!(error = %error, "Failed to cancel forget record"),
            }
            Ok::<(), StoreError>(())
        })
        .await;
}

/// Forgets `pk` here unless it is alive, and returns whether it did. Judged and removed
/// under the same locks, so a message that arrives meanwhile can't be lost to a peer
/// state that is already gone. A peer whose state is in use is left for the next check.
pub(crate) async fn forget_unless_alive(state: &AppState, pk: &str, now: Instant) -> bool {
    let mut peer_states = state.peer_states.write().await;
    let Some(peer_state) = peer_states.get(pk) else {
        return false;
    };
    // Not waiting: holding the map, we must not wait for a peer's lock (its holder may
    // be waiting for the map).
    let Ok(peer_state) = peer_state.try_lock() else {
        return false;
    };
    if peer_state.is_alive(&state.timing, now) {
        return false;
    }
    drop(peer_state);
    peer_states.remove(pk);
    true
}

/// 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"
    );
}