//! 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::replica::Epoch;
use crate::document::{ChangeHash, ReadDoc};
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 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");
    }
}

/// `hash`, which `doc` (of `epoch`) holds, judged by its rules now: why they refuse it,
/// if they do; None if `doc` doesn't hold it.
fn judge_held(
    doc: &crate::document::Document,
    epoch: Epoch,
    hash: &ChangeHash,
) -> Option<Result<(), String>> {
    let change = doc.get_change_by_hash(hash)?;
    // An epoch's root is what the document is: never refused, whatever a peer says.
    if change.deps().is_empty() {
        return Some(Ok(()));
    }
    let decoded = change.decode();
    let ops = crate::document::inspect::Inspector::new(doc)
        .ops(std::slice::from_ref(&decoded))
        .pop()
        .unwrap_or_default();
    let paths = ops.iter().map(|op| op.path.clone()).collect();
    let committer = doc
        .get_author_for_actor(change.actor_id())
        .and_then(|author| {
            std::str::from_utf8(author.as_bytes())
                .ok()
                .map(str::to_owned)
        });
    let user = change
        .message()
        .and_then(crate::document::replica::user_of)
        .map(str::to_owned);
    let deps = change.deps().to_vec();
    let judged = crate::access::rules::judge(
        doc,
        epoch,
        &crate::access::rules::Change {
            hash: *hash,
            deps: &deps,
            seen: &deps,
            committer: committer.as_deref(),
            user: user.as_deref(),
            paths: &paths,
            ops: &ops,
        },
    );
    Some(match judged {
        Ok(()) => Ok(()),
        Err(crate::access::rules::Refusal::Rule(why)) => Err(why),
        // Its dependencies are in the document: it is.
        Err(crate::access::rules::Refusal::Unready) => Ok(()),
    })
}