//! The change publisher: after document changes, which store paths they touched
//! (`AppState::change_events`, for the API's streams, the dashboard and remakes), and
//! a nudge to topology when the member list changed.

use std::sync::Arc;

use tracing::{info, warn};

use crate::document::store::ChangeSet;
use crate::state::AppState;

fn patch_store_path(patch: &automerge::Patch) -> Vec<String> {
    let mut path: Vec<String> = patch
        .path
        .iter()
        .map(|(_, prop)| match prop {
            automerge::Prop::Map(key) => key.clone(),
            automerge::Prop::Seq(index) => index.to_string(),
        })
        .collect();

    match &patch.action {
        automerge::PatchAction::PutMap { key, .. }
        | automerge::PatchAction::DeleteMap { key, .. } => path.push(key.clone()),
        // List, text and counter changes: the containing object changed.
        _ => {}
    }
    path
}

pub(crate) async fn run_change_publisher(state: AppState) {
    let mut changed = state.change_tx.subscribe();
    let mut last_heads = state.write_doc(|doc| doc.get_heads()).await;

    info!("Starting change publisher");

    while changed.changed().await.is_ok() {
        let before = last_heads.clone();

        let (heads, paths, conflicts) = state
            .write_doc(move |doc| {
                let heads = doc.get_heads();
                if heads == before {
                    return (heads, Vec::new(), Vec::new());
                }

                // After an epoch switch the previous heads belong to the old document:
                // report the root as changed, which every stream is related to.
                if !before.iter().all(|h| doc.get_change_by_hash(h).is_some()) {
                    return (heads, vec![Vec::new()], Vec::new());
                }

                let mut paths = Vec::new();
                let mut conflicts = Vec::new();
                for patch in doc.diff(&before, &heads) {
                    let path = patch_store_path(&patch);
                    if matches!(
                        patch.action,
                        automerge::PatchAction::PutMap { conflict: true, .. }
                    ) {
                        conflicts.push(path.clone());
                    }
                    paths.push(path);
                }
                (heads, paths, conflicts)
            })
            .await;

        if heads == last_heads {
            continue;
        }
        last_heads = heads;

        for path in &conflicts {
            warn!(path = ?path, "Concurrent writes produced a conflict");
        }

        if paths
            .iter()
            .any(|path| crate::swarm::topology::affects_neighbours(path))
        {
            state.topology_changed.notify_one();
        }

        // Fails only when nobody is subscribed.
        let _ = state.change_events.send(Arc::new(ChangeSet { paths }));
    }
}