//! Signatures of changes sent on their own (SIGNATURES), where no sync message took
//! them: a signature made after its change went (`--sign-every`), or one received
//! without the change it vouches for, as from a peer that had it already. Every
//! `SEND_EVERY`, each active peer on our epoch is sent the signatures it wasn't sent yet
//! (`Signatures::newer_than`), if any; it checks them as it checks those sync messages
//! bring (`Signatures::offer`), and sends them on in turn: each actor's latest goes to
//! each node about once.

use std::time::Duration;

use tracing::debug;

use crate::state::AppState;
use crate::transport::{ParsedMessage, SignaturesMessage};

/// How often the signatures the peers weren't sent yet go out.
pub(crate) const SEND_EVERY: Duration = Duration::from_secs(1);

/// Sends the signatures due every `SEND_EVERY`. Runs for as long as the node does.
pub(crate) async fn run(state: AppState) {
    let mut ticks = tokio::time::interval(SEND_EVERY);
    ticks.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        ticks.tick().await;
        send_due(&state).await;
    }
}

/// Each active peer on our epoch: the signatures it wasn't sent yet, if any.
async fn send_due(state: &AppState) {
    for peer in state.active_peers_on_local_epoch().await {
        let Some(peer_state) = state.get_peer_state(&peer).await else {
            continue;
        };
        let doc = std::sync::Arc::clone(&state.doc);
        // Lock order doc -> peer, as in sync generation.
        let message = tokio::task::spawn_blocking(move || {
            let replica = doc.blocking_read();
            let mut peer_state = peer_state.blocking_lock();
            let (signatures, sent) = replica.signatures.newer_than(peer_state.signatures_sent);
            if signatures.is_empty() {
                return None;
            }
            peer_state.signatures_sent = sent;
            Some(ParsedMessage::Signatures(SignaturesMessage {
                epoch: replica.epoch,
                epoch_seq: replica.epoch_seq,
                signatures: serde_bytes::ByteBuf::from(crate::document::signatures::pack(
                    &signatures,
                )),
            }))
        })
        .await;
        if let Ok(Some(message)) = message
            && let Err(error) = state.transport.send_message(&peer, &message).await
        {
            // Lost, they come again with the next sync session (`PeerState::reset_sync`).
            debug!(peer = %peer, error = %format!("{error:#}"), "Couldn't send signatures");
        }
    }
}

/// SIGNATURES from `peer`, of our epoch: checked and kept as those of sync messages.
pub(crate) async fn received(state: &AppState, message: SignaturesMessage) {
    let Some(signatures) = crate::document::signatures::unpack(&message.signatures)
        .filter(|signatures| signatures.len() <= crate::document::signatures::MAX_PER_MESSAGE)
    else {
        return;
    };
    let doc = std::sync::Arc::clone(&state.doc);
    let screening = std::sync::Arc::clone(&state.screening);
    // How many waiting changes went in, if any: published as received ones are.
    let applied = tokio::task::spawn_blocking(move || {
        let mut replica = doc.blocking_write();
        if replica.epoch != message.epoch {
            return 0;
        }
        {
            let (epoch, epoch_seq) = (replica.epoch, replica.epoch_seq);
            let replica = &mut *replica;
            replica.signatures.offer(&replica.doc, signatures);
            // With signatures required, changes waiting for them judged again
            // (`access::screen`).
            if crate::access::policy::of(&replica.doc).required() {
                let applied = replica.logged_signed(|doc, known, log| {
                    let applied = crate::access::screen::apply_bytes(
                        doc,
                        epoch,
                        epoch_seq,
                        &[],
                        &screening,
                        Some(known),
                        log,
                    );
                    (applied, true)
                });
                replica.refresh_heads();
                return match applied {
                    Ok(outcome) => outcome.applied,
                    Err(error) => {
                        tracing::warn!(error = %format!("{error:#}"), "Couldn't judge the changes waiting for signatures");
                        0
                    }
                };
            }
            0
        }
    })
    .await
    .unwrap_or(0);
    if applied > 0 {
        state.publish_change();
    }
}