//! The member registry: every node that has joined the swarm, under
//! `_continuum/members/<pk>`, replicated with the document (and carried into new
//! epochs with the rest of `_continuum`).
//!
//! A pure registry: it decides nothing about admission (that's the swarm key). Each
//! node registers itself once it is ready, and re-registers if its entry disappears
//! while it is alive. An operator's forget (`DELETE /peers/{pk}`) removes the entry
//! together with writing the forget record; a forgotten node that turns out to be alive
//! cancels the record and registers again.
//!
//! Besides its tier, an entry holds the alias a node gives itself (`--alias`), a name
//! for people reading /peers, metrics and logs. Not unique and never an identity: the
//! public key stays that.

use std::collections::HashMap;

use automerge::AutoCommit;
use serde_json::{Value as JsonValue, json};
use tracing::{info, warn};

use crate::document::store::{self, StoreError};
use crate::state::{AppState, unix_millis};

const META_ROOT: &str = "_continuum";

fn members_dir() -> Vec<String> {
    vec![META_ROOT.to_owned(), "members".to_owned()]
}

pub(crate) fn member_path(pk: &str) -> Vec<String> {
    let mut path = members_dir();
    path.push(pk.to_owned());
    path
}

/// Public keys of all registered members, sorted.
pub(crate) fn members(doc: &AutoCommit) -> Vec<String> {
    let mut members = store::keys_at(doc, &members_dir()).unwrap_or_default();
    members.sort();
    members
}

/// A member's stability tier, judged by itself (`swarm::stability`).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) enum Tier {
    Stable,
    #[default]
    Transient,
}

impl Tier {
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            Self::Stable => "stable",
            Self::Transient => "transient",
        }
    }
}

fn tier_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("tier".to_owned());
    path
}

/// `pk`'s tier; transient unless its entry says stable.
pub(crate) fn tier(doc: &AutoCommit, pk: &str) -> Tier {
    match store::read_path(doc, &tier_path(pk)) {
        Ok(Some(JsonValue::String(tier))) if tier == "stable" => Tier::Stable,
        _ => Tier::Transient,
    }
}

pub(crate) fn set_tier(doc: &mut AutoCommit, pk: &str, tier: Tier) -> Result<(), StoreError> {
    store::write_path(doc, &tier_path(pk), &json!(tier.as_str()))
}

fn alias_path(pk: &str) -> Vec<String> {
    let mut path = member_path(pk);
    path.push("alias".to_owned());
    path
}

/// `pk`'s alias, if it gave itself one.
pub(crate) fn alias(doc: &AutoCommit, pk: &str) -> Option<String> {
    match store::read_path(doc, &alias_path(pk)) {
        Ok(Some(JsonValue::String(alias))) => Some(alias),
        _ => None,
    }
}

/// The aliases of all members that have one.
pub(crate) fn aliases(doc: &AutoCommit) -> HashMap<String, String> {
    members(doc)
        .into_iter()
        .filter_map(|pk| alias(doc, &pk).map(|alias| (pk, alias)))
        .collect()
}

fn set_alias(doc: &mut AutoCommit, pk: &str, alias: Option<&str>) -> Result<(), StoreError> {
    match alias {
        Some(alias) => store::write_path(doc, &alias_path(pk), &json!(alias)),
        None => match store::delete_path(doc, &alias_path(pk), false) {
            Err(StoreError::NotFound) => Ok(()),
            result => result,
        },
    }
}

/// Registers this node if it is ready and its entry is missing, and keeps its alias
/// as configured (set, changed or removed after a restart with another `--alias`).
pub(crate) async fn ensure_registered(state: &AppState) {
    if !state.is_ready().await {
        return;
    }
    let local = state.local_id().to_owned();
    let own = member_path(&local);
    let (registered, recorded_alias) = {
        let replica = state.doc.read().await;
        let registered = store::read_path(&replica.doc, &own)
            .ok()
            .flatten()
            .is_some();
        (registered, alias(&replica.doc, &local))
    };
    let wanted = state.alias.clone();

    if !registered {
        let mut entry = json!({ "since": unix_millis() });
        if let Some(alias) = &wanted {
            entry["alias"] = json!(alias);
        }
        match state
            .mutate_doc(move |doc| store::write_path(doc, &own, &entry))
            .await
        {
            Ok(()) => info!(
                alias = wanted.as_deref(),
                "Registered in the swarm's member list"
            ),
            Err(error) => warn!(error = %error, "Failed to register in the member list"),
        }
        return;
    }

    if recorded_alias == wanted {
        return;
    }
    let alias = wanted.clone();
    match state
        .mutate_doc(move |doc| set_alias(doc, &local, alias.as_deref()))
        .await
    {
        Ok(()) => info!(
            alias = wanted.as_deref(),
            previous = recorded_alias.as_deref(),
            "Alias updated in the member list"
        ),
        Err(error) => warn!(error = %error, "Failed to record our alias"),
    }
}

/// Removes `pk`'s entry (an operator forgot it). Absent is fine.
pub(crate) fn unregister(doc: &mut AutoCommit, pk: &str) -> Result<(), StoreError> {
    match store::delete_path(doc, &member_path(pk), false) {
        Err(StoreError::NotFound) => Ok(()),
        result => result,
    }
}