//! The state file (`--state-file`): what a node remembers across restarts, so it
//! rejoins its swarm on its own: the swarm ID and some members to reach it through.
//! Not the document: a restarted node still starts empty and syncs.
//!
//! Written rarely: once the node is ready, then when the remembered members change
//! (checked every few minutes, written at most once an hour), and at shutdown if they
//! changed since, atomically (`files`).

use std::path::PathBuf;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;

use serde::{Deserialize, Serialize};
use tokio::time::Instant;
use tracing::{error, info, warn};

use crate::files::{FileKind, Files};
use crate::state::AppState;
use crate::swarm::members;

/// Members remembered as seeds.
const MAX_SEEDS: usize = 16;
/// How often the remembered members are compared with the registry.
const CHECK_INTERVAL: Duration = Duration::from_secs(300);
/// Writes at least this far apart, but for the first one and at shutdown.
const MIN_WRITE_INTERVAL: Duration = Duration::from_secs(3600);

/// The file's content.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Remembered {
    pub(crate) swarm_id: String,
    /// Public keys of members to join through (sorted).
    #[serde(default)]
    pub(crate) seeds: Vec<String>,
}

pub(crate) struct StateFile {
    path: PathBuf,
    files: Arc<Files>,
    // A std mutex: never held across an await.
    inner: Mutex<Inner>,
}

#[derive(Default)]
struct Inner {
    written: Option<Remembered>,
    written_at: Option<Instant>,
    checked_at: Option<Instant>,
}

impl StateFile {
    /// Opens `path`, returning what it remembers. A missing file remembers nothing; an
    /// unreadable one is moved aside (`.bad`) and logged, and remembers nothing either.
    pub(crate) fn open(path: PathBuf, files: Arc<Files>) -> (Self, Option<Remembered>) {
        let remembered = match files.read(FileKind::State, &path) {
            Ok(Some(bytes)) => match serde_json::from_slice::<Remembered>(&bytes) {
                Ok(remembered) if !remembered.swarm_id.is_empty() => Some(remembered),
                Ok(_) => None,
                Err(error) => {
                    error!(
                        path = %path.display(),
                        error = %error,
                        "Unreadable state file; moving it aside and remembering nothing"
                    );
                    match files.move_aside(&path, ".bad") {
                        Ok(aside) => {
                            warn!(moved_to = %aside.display(), "Moved the state file aside")
                        }
                        Err(error) => warn!(error = %error, "Couldn't move the state file aside"),
                    }
                    None
                }
            },
            Ok(None) => None,
            Err(error) => {
                error!(path = %path.display(), error = %error, "Couldn't read the state file; remembering nothing");
                None
            }
        };
        let file = Self {
            path,
            files,
            inner: Mutex::new(Inner {
                written: remembered.clone(),
                ..Inner::default()
            }),
        };
        (file, remembered)
    }

    fn lock(&self) -> MutexGuard<'_, Inner> {
        self.inner
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
    }

    fn write(&self, remembered: &Remembered) -> std::io::Result<()> {
        let mut bytes = serde_json::to_vec_pretty(remembered)?;
        bytes.push(b'\n');
        self.files.write_atomic(FileKind::State, &self.path, &bytes)
    }
}

/// Brings the state file up to date if it is due: `now` for the first write and at
/// shutdown, otherwise only every few minutes and at most once an hour. Called by the
/// peer worker on its maintenance tick.
pub(crate) async fn update(state: &AppState, now: bool) {
    let Some(file) = state.state_file.clone() else {
        return;
    };
    if !state.is_ready().await {
        return;
    }
    let Some(swarm_id) = state.swarm_id().await else {
        return;
    };
    {
        let mut inner = file.lock();
        if !now
            && inner
                .checked_at
                .is_some_and(|at| at.elapsed() < CHECK_INTERVAL)
        {
            return;
        }
        inner.checked_at = Some(Instant::now());
    }

    let seeds = {
        let replica = state.doc.read().await;
        seeds(
            &replica.doc,
            state.local_id(),
            state.electorate_window,
            state.clock.now_millis(),
        )
    };
    let remembered = Remembered { swarm_id, seeds };

    {
        let inner = file.lock();
        if inner.written.as_ref() == Some(&remembered) {
            return;
        }
        let first = inner.written_at.is_none();
        if !now
            && !first
            && inner
                .written_at
                .is_some_and(|at| at.elapsed() < MIN_WRITE_INTERVAL)
        {
            return;
        }
    }

    let writer = Arc::clone(&file);
    let content = remembered.clone();
    match tokio::task::spawn_blocking(move || writer.write(&content)).await {
        Ok(Ok(())) => {
            info!(
                path = %file.path.display(),
                seeds = remembered.seeds.len(),
                "Wrote the state file"
            );
            let mut inner = file.lock();
            inner.written = Some(remembered);
            inner.written_at = Some(Instant::now());
        }
        Ok(Err(error)) => {
            error!(path = %file.path.display(), error = %error, "Couldn't write the state file");
        }
        Err(error) => error!(error = %error, "State file write task failed"),
    }
}

/// Members to rejoin through: those alive within the electorate window, stable ones
/// first, at most `MAX_SEEDS`, sorted. Changes only when members join, leave or change
/// tier, so the file rarely does.
fn seeds(doc: &automerge::AutoCommit, me: &str, window: Duration, now: u64) -> Vec<String> {
    let (mut stable, transient): (Vec<String>, Vec<String>) = members::electorate(doc, now, window)
        .into_iter()
        .filter(|pk| pk != me)
        .partition(|pk| members::tier(doc, pk) == members::Tier::Stable);
    stable.extend(transient);
    stable.truncate(MAX_SEEDS);
    stable.sort();
    stable
}