//! The worker that keeps the document file (`document::file`) up to date: it looks at
//! the node (ready, the document's epoch and heads, this node's actors), and the file
//! says what is due. While a ready node rejoins, the file follows the document before
//! the rejoin, which takes the node's writes meanwhile (`state::StaleView`): a restart
//! then resumes on it, and rejoins again.

use std::sync::Arc;

use crate::document::Document;
use crate::document::file::Due;
use crate::document::replica::Epoch;
use crate::locks::RwLockExt;
use crate::state::AppState;

/// Keeps the document file up to date. Runs for as long as the node does.
pub(crate) async fn run(state: AppState) {
    let Some(file) = state.document_file.clone() else {
        return;
    };
    let mut tick = tokio::time::interval(file.append_every());
    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        tick.tick().await;
        flush(&state).await;
    }
}

/// At a clean stop, after the last `flush`: the file marked as holding everything this
/// node wrote (`DocumentFile::close`), where it does (ready, not rejoining, up to the
/// document's heads), so a restart may write on as the same actor (`--keep-actor`).
pub(crate) async fn close(state: &AppState) {
    let Some(file) = state.document_file.clone() else {
        return;
    };
    if state.remake.is_rejoining() || !state.is_ready().await {
        return;
    }
    let heads = state.doc.read().await.doc.get_heads();
    file.close(&heads);
}

/// Writes what is due now (`DocumentFile::due`). Also called at shutdown.
pub(crate) async fn flush(state: &AppState) {
    let Some(file) = state.document_file.clone() else {
        return;
    };
    if state.remake.is_rejoining() {
        if let Some((epoch, stale)) = state.stale_document() {
            flush_before_rejoin(&file, epoch, stale).await;
        }
        return;
    }
    if !state.is_ready().await {
        return;
    }
    let own = state.remake.own_actors();
    let (epoch, heads, signatures) = {
        let replica = state.doc.read().await;
        (
            replica.epoch,
            replica.heads().to_vec(),
            replica.signatures.count(),
        )
    };
    match file.due(epoch, own.len(), &heads, Some(signatures)) {
        Due::Rewrite(reason) => {
            let (epoch, copy, signatures) = {
                let replica = state.doc.read().await;
                (
                    replica.epoch,
                    replica.doc.clone(),
                    replica.signatures.since(0),
                )
            };
            file.rewrite(epoch, copy, signatures, own, reason).await;
        }
        Due::Append {
            epoch,
            since,
            signatures,
        } => {
            let (heads, increment, signatures) = {
                let replica = state.doc.read().await;
                // Switched since: the next flush rewrites the file instead.
                if replica.epoch != epoch {
                    return;
                }
                let heads = replica.heads().to_vec();
                (
                    heads,
                    crate::document::save_since(&replica.doc, &since),
                    replica.signatures.since(signatures),
                )
            };
            file.append(since, heads, increment, signatures).await;
        }
        Due::Nothing => {}
    }
}

/// [`flush`] of the document before a rejoin (`epoch`'s), while it takes this node's
/// writes. Only appends: a rewrite (another epoch, or the increments outgrew the full
/// save) waits for the end of the rejoin, so the file keeps the actors it had.
async fn flush_before_rejoin(
    file: &Arc<crate::document::file::DocumentFile>,
    epoch: Epoch,
    stale: Arc<std::sync::RwLock<Document>>,
) {
    let heads = {
        let stale = Arc::clone(&stale);
        // Sorted, as the live document's (`Replica::heads`) the file compares with.
        tokio::task::spawn_blocking(move || {
            let mut heads = stale.read_or_recover().get_heads();
            heads.sort_unstable();
            heads
        })
        .await
        .expect("document read task panicked")
    };
    let Some(actors) = file.actors() else {
        return;
    };
    // Its signatures aren't kept: those of the document after the rejoin are.
    if let Due::Append {
        epoch: written,
        since,
        ..
    } = file.due(epoch, actors, &heads, None)
        && written == epoch
    {
        let since_then = since.clone();
        let (heads, increment) = tokio::task::spawn_blocking(move || {
            let doc = stale.read_or_recover();
            let mut heads = doc.get_heads();
            heads.sort_unstable();
            (heads, crate::document::save_since(&doc, &since_then))
        })
        .await
        .expect("document read task panicked");
        file.append(since, heads, increment, Vec::new()).await;
    }
}