//! Telling a peer which of the changes it brought were refused (REFUSALS), so a node
//! that took a change the swarm's rules refuse finds out and sets itself right.
//!
//! A node takes a change the others refuse only if it didn't judge it as they do: under
//! a policy it didn't know yet (it receives without judging then, `access::policy`), or
//! one that came after. Everything it writes from then on builds on that change and is
//! refused too, and nobody tells it so. Hence: a node that refuses changes a peer
//! brought tells it which broke a rule (`tell`), at most every `TELL_EVERY` and
//! `MAX_LISTED` of them. The peer doesn't take the word for it (a member could tell
//! anything): it judges each it holds itself, as its rules have them now
//! (`received`). If its rules refuse one too, it took it unjudged, and drops it with
//! everything built on it: its document as it would be without them, its own writes
//! among them quarantined (`AppState::drop_taken`). If its rules take it, the two
//! policies disagree, which it logs; finding out which is right is left
//! (`future-work.md`, Write protection by checking changes).

use std::time::Duration;

use tracing::{debug, warn};

use crate::document::ChangeHash;
use crate::document::replica::Epoch;
use crate::state::AppState;
use crate::transport::{ParsedMessage, RefusalsMessage};

/// Changes listed in one REFUSALS, at most.
pub(crate) const MAX_LISTED: usize = 64;
/// How often a peer is told, at most.
pub(crate) const TELL_EVERY: Duration = Duration::from_secs(30);

/// Tells `peer` that `changes` it brought, of our epoch, broke the swarm's rules.
pub(crate) async fn tell(
    state: &AppState,
    peer: &str,
    epoch: Epoch,
    epoch_seq: u32,
    mut changes: Vec<ChangeHash>,
) {
    changes.truncate(MAX_LISTED);
    let message = ParsedMessage::Refusals(RefusalsMessage {
        epoch,
        epoch_seq,
        changes,
    });
    if let Err(error) = state.transport.send_message(peer, &message).await {
        // Told again with the next refusal.
        debug!(peer = %peer, error = %format!("{error:#}"), "Couldn't tell refusals");
    }
}

/// REFUSALS from `peer`, of our epoch: each change listed that we hold judged as our
/// rules have it now; if they refuse one, we took it unjudged, and drop it.
pub(crate) async fn received(state: &AppState, peer: &str, message: RefusalsMessage) {
    let doc = std::sync::Arc::clone(&state.doc);
    let changes: Vec<ChangeHash> = message.changes.into_iter().take(MAX_LISTED).collect();
    let judged = tokio::task::spawn_blocking(move || {
        let replica = doc.blocking_read();
        if replica.epoch != message.epoch {
            return None;
        }
        let policy = crate::access::policy::of(&replica.doc);
        if !policy.enforced() || matches!(policy, crate::access::policy::Policy::Unknown(_)) {
            // We can't judge as they do: receiving without judging until upgraded.
            return None;
        }
        let mut refused = None;
        let mut taken = Vec::new();
        for hash in changes {
            match crate::access::rules::judge_held(&replica.doc, replica.epoch, &hash) {
                Some(Err(why)) => {
                    refused = Some((hash, why));
                    break;
                }
                Some(Ok(())) => taken.push(hash),
                None => {}
            }
        }
        Some((refused, taken))
    })
    .await
    .ok()
    .flatten();
    let Some((refused, taken)) = judged else {
        return;
    };
    if let Some((hash, why)) = refused {
        warn!(peer = %peer, change = %hash, why = %why, "We hold a change the swarm's rules refuse, taken unjudged: dropping it and what builds on it (our writes among them go to the quarantine)");
        state.drop_taken(hash).await;
    } else if !taken.is_empty() {
        warn!(peer = %peer, changes = ?taken, "A peer refuses changes our rules take: its policy or ours is off");
    }
}