//! The replication worker. One task drains the queue: debounced per-repo pushes (5 s after the
//! last ref update), a database snapshot every `db_snapshot_minutes`, and a reconcile pass every
//! `reconcile_minutes` that pushes every repo, removes deleted repos from the standby, and pings
//! the healthcheck. Runs start once at boot, so a fresh standby is filled without waiting.
//!
//! Every outcome is recorded in `replication_status` (see `forge_core::store::replication`), which
//! `/admin/replication` shows.

use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use forge_core::Store;
use forge_core::store::replication::DB_SUBJECT;
use forge_core::store::snapshot::verify_snapshot;
use forge_git::GitEnv;
use forge_git::backup::refs_checksum;

use super::{
    KEEP_SNAPSHOTS, Layout, SshTransport, StandbyTransport, Target, is_snapshot_name, push_repo,
    remote_refs_checksum, sh_quote, snapshot_stamp, wiki_subject, wiki_subject_repo,
};
use crate::backup::healthcheck;
use crate::state::AppState;

/// Longest the worker sleeps with nothing due. Guards against a missed wake, not polling.
const BACKSTOP: Duration = Duration::from_secs(15 * 60);

/// Everything a replication pass needs. The web side builds one from `AppState`; tests build one
/// around a `LocalTransport`.
pub struct Ctx<T> {
    pub store: Store,
    pub git: GitEnv,
    pub data_dir: PathBuf,
    pub layout: Layout,
    pub transport: T,
}

/// What a reconcile pass did. `failures` holds one line per repo or step that failed.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct ReconcileReport {
    pub replicated: usize,
    pub failures: Vec<String>,
}

/// Starts the worker task. It runs until the process exits. An invalid target is logged and the
/// worker does not start; config validation makes that unlikely.
pub fn spawn(state: AppState) {
    tokio::spawn(run(state));
}

async fn run(state: AppState) {
    let config = state.config.replication.clone();
    let target = match Target::parse(&config.target, config.ssh_port) {
        Ok(target) => target,
        Err(e) => {
            tracing::error!(error = %e, "replication: invalid target, worker not started");
            return;
        }
    };
    let layout = match Layout::new(&target.base) {
        Ok(layout) => layout,
        Err(e) => {
            tracing::error!(error = %e, "replication: invalid remote base, worker not started");
            return;
        }
    };
    let transport = SshTransport::new(
        target,
        config.identity_file.clone(),
        config.known_hosts_file.clone(),
    );
    let ctx = Ctx {
        store: state.store.clone(),
        git: (*state.git).clone(),
        data_dir: state.git.data_dir.clone(),
        layout,
        transport,
    };
    tracing::info!(target = %config.target, "replication: worker started");

    let db_every = Duration::from_secs(config.db_snapshot_minutes.max(1) * 60);
    let reconcile_every = Duration::from_secs(config.reconcile_minutes.max(1) * 60);
    let mut next_db = Instant::now();
    let mut next_reconcile = Instant::now();
    loop {
        // Sleep until the earliest debounce or calendar timer, or until something is queued.
        let backstop = Instant::now() + BACKSTOP;
        let next = [
            state.replication.next_due(),
            Some(next_db),
            Some(next_reconcile),
            Some(backstop),
        ]
        .into_iter()
        .flatten()
        .min()
        .unwrap_or(backstop);
        tokio::select! {
            () = tokio::time::sleep_until(tokio::time::Instant::from_std(next)) => {}
            () = state.replication.wait() => {}
        }
        for repo_id in state.replication.take_due(Instant::now()) {
            // A repo deleted since the update is skipped; the next reconcile removes its copy.
            match ctx.store.repo_by_id(&repo_id).await {
                Ok(Some(_)) => {
                    let _ = replicate_repo(&ctx, &repo_id).await;
                }
                Ok(None) => {}
                Err(e) => tracing::error!(error = %e, "replication: could not look up repo"),
            }
        }
        let now = Instant::now();
        if now >= next_db {
            next_db = now + db_every;
            if let Err(e) = ship_db(&ctx).await {
                tracing::error!(error = %e, "replication: database snapshot not shipped");
            }
        }
        // Read the request even when the timer is due, so one request is one pass.
        let requested = state.replication.take_reconcile_request();
        if requested || now >= next_reconcile {
            next_reconcile = Instant::now() + reconcile_every;
            let report = reconcile_once(&ctx).await;
            tracing::info!(
                replicated = report.replicated,
                failures = report.failures.len(),
                "replication: reconcile finished"
            );
            ping(&config.healthcheck_url, &report).await;
        }
    }
}

/// Pings the healthcheck after a reconcile: success when every step worked, failure otherwise.
async fn ping(url: &str, report: &ReconcileReport) {
    if url.is_empty() {
        return;
    }
    if report.failures.is_empty() {
        healthcheck::success(url).await;
    } else {
        healthcheck::failure(url, &report.failures.join("; ")).await;
    }
}

/// Records an outcome for `subject` and passes the error on. `refs` is the checksum to store on
/// success (repos only).
async fn record(
    store: &Store,
    subject: &str,
    result: Result<Option<String>, String>,
) -> Result<(), String> {
    match result {
        Ok(refs) => {
            if let Err(e) = store.record_replication_ok(subject, refs.as_deref()).await {
                tracing::error!(error = %e, subject, "replication: could not record success");
            }
            Ok(())
        }
        Err(e) => {
            tracing::warn!(subject, error = %e, "replication: step failed");
            if let Err(db) = store.record_replication_error(subject, &e).await {
                tracing::error!(error = %db, subject, "replication: could not record failure");
            }
            Err(e)
        }
    }
}

/// Pushes one repo to the standby and checks that its refs match. A mismatch triggers one
/// re-push; a second mismatch is an error. Records the outcome for the repo.
///
/// The repo's wiki, when it has one, is replicated the same way after the repo, under the subject
/// `<id>.wiki`. A failure in either is returned.
pub async fn replicate_repo<T: StandbyTransport>(
    ctx: &Ctx<T>,
    repo_id: &str,
) -> Result<(), String> {
    let result = replicate_inner(ctx, repo_id).await;
    let repo_outcome = record(&ctx.store, repo_id, result).await;
    let wiki = forge_git::repo::wiki_path(&ctx.data_dir, repo_id).map_err(|e| e.to_string())?;
    let wiki_outcome = if wiki.join("HEAD").is_file() {
        let result = replicate_wiki_inner(ctx, repo_id).await;
        record(&ctx.store, &wiki_subject(repo_id), result).await
    } else {
        Ok(())
    };
    repo_outcome?;
    wiki_outcome
}

async fn replicate_inner<T: StandbyTransport>(
    ctx: &Ctx<T>,
    repo_id: &str,
) -> Result<Option<String>, String> {
    let local = forge_git::repo::repo_path(&ctx.data_dir, repo_id).map_err(|e| e.to_string())?;
    if !local.is_dir() {
        return Err(format!(
            "local repo directory is missing: {}",
            local.display()
        ));
    }
    let remote = ctx.layout.repo_dir(repo_id)?;
    replicate_dir(ctx, &local, &remote).await
}

async fn replicate_wiki_inner<T: StandbyTransport>(
    ctx: &Ctx<T>,
    repo_id: &str,
) -> Result<Option<String>, String> {
    let local = forge_git::repo::wiki_path(&ctx.data_dir, repo_id).map_err(|e| e.to_string())?;
    let remote = ctx.layout.wiki_dir(repo_id)?;
    replicate_dir(ctx, &local, &remote).await
}

/// Pushes the bare repo at `local` to `remote` and checks that the refs match, retrying once.
async fn replicate_dir<T: StandbyTransport>(
    ctx: &Ctx<T>,
    local: &Path,
    remote: &str,
) -> Result<Option<String>, String> {
    // Idempotent: creates the bare repo on first use, and does nothing to an existing one.
    ctx.transport
        .run(&format!("git init --bare --quiet {}", sh_quote(remote)))
        .await?;

    push_repo(&ctx.transport, &ctx.git, local, remote).await?;
    let mut local_sum = local_checksum(&ctx.git, local).await?;
    let mut remote_sum = remote_refs_checksum(&ctx.transport, remote).await?;
    if remote_sum != local_sum {
        push_repo(&ctx.transport, &ctx.git, local, remote).await?;
        local_sum = local_checksum(&ctx.git, local).await?;
        remote_sum = remote_refs_checksum(&ctx.transport, remote).await?;
        if remote_sum != local_sum {
            return Err(format!(
                "refs checksum mismatch after re-push: local {local_sum}, standby {remote_sum}"
            ));
        }
    }
    Ok(Some(local_sum))
}

async fn local_checksum(git: &GitEnv, local: &Path) -> Result<String, String> {
    refs_checksum(git, local).await.map_err(|e| e.to_string())
}

/// Snapshots the database, verifies the copy, ships it as `db/forge-<stamp>.db`, points `latest`
/// at it and prunes the standby to the newest [`KEEP_SNAPSHOTS`]. Records the outcome as `db`.
pub async fn ship_db<T: StandbyTransport>(ctx: &Ctx<T>) -> Result<(), String> {
    let result = ship_db_inner(ctx).await;
    record(&ctx.store, DB_SUBJECT, result.map(|()| None)).await
}

async fn ship_db_inner<T: StandbyTransport>(ctx: &Ctx<T>) -> Result<(), String> {
    let run_dir = ctx.data_dir.join("run");
    std::fs::create_dir_all(&run_dir)
        .map_err(|e| format!("cannot create {}: {e}", run_dir.display()))?;
    let snap = run_dir.join("replica-snap.db");
    remove_snapshot_files(&snap)?;
    ctx.store
        .snapshot_to(&snap)
        .await
        .map_err(|e| e.to_string())?;
    let shipped = ship_snapshot(ctx, &snap).await;
    // The local copy is only a staging file; a failure to delete it is not a replication failure.
    let _ = remove_snapshot_files(&snap);
    shipped
}

async fn ship_snapshot<T: StandbyTransport>(ctx: &Ctx<T>, snap: &Path) -> Result<(), String> {
    verify_snapshot(snap).await.map_err(|e| e.to_string())?;

    let db_dir = ctx.layout.db_dir();
    ctx.transport
        .run(&format!("mkdir -p {}", sh_quote(&db_dir)))
        .await?;

    let stamp = snapshot_stamp();
    let name = format!("forge-{stamp}.db");
    ctx.transport
        .upload(snap, &ctx.layout.snapshot_file(&stamp))
        .await?;

    // `ln -sfn` writes the link under a temporary name, and `mv -T` swaps it in atomically.
    let latest_tmp = format!("{db_dir}/.latest-tmp");
    ctx.transport
        .run(&format!(
            "ln -sfn -- {} {} && mv -T -- {} {}",
            sh_quote(&name),
            sh_quote(&latest_tmp),
            sh_quote(&latest_tmp),
            sh_quote(&ctx.layout.latest_link())
        ))
        .await?;

    prune_snapshots(ctx, &name).await
}

/// Deletes all but the newest [`KEEP_SNAPSHOTS`] snapshot files. `keep_always` is never deleted,
/// even when clock skew makes it sort below older files.
async fn prune_snapshots<T: StandbyTransport>(
    ctx: &Ctx<T>,
    keep_always: &str,
) -> Result<(), String> {
    let db_dir = ctx.layout.db_dir();
    let listing = ctx
        .transport
        .run(&format!("ls -1 -- {}", sh_quote(&db_dir)))
        .await?;
    let text = String::from_utf8_lossy(&listing);
    let mut names: Vec<&str> = text.lines().filter(|name| is_snapshot_name(name)).collect();
    names.sort_unstable();
    let excess = names.len().saturating_sub(KEEP_SNAPSHOTS);
    let victims: Vec<String> = names
        .iter()
        .take(excess)
        .filter(|name| **name != keep_always)
        .map(|name| sh_quote(&format!("{db_dir}/{name}")))
        .collect();
    if victims.is_empty() {
        return Ok(());
    }
    ctx.transport
        .run(&format!("rm -f -- {}", victims.join(" ")))
        .await
        .map(|_| ())
}

/// Removes `snap` and its `-wal` and `-shm` sidecars when present.
fn remove_snapshot_files(snap: &Path) -> Result<(), String> {
    for suffix in ["", "-wal", "-shm"] {
        let mut path = snap.as_os_str().to_owned();
        path.push(suffix);
        match std::fs::remove_file(&path) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(format!("cannot remove {}: {e}", Path::new(&path).display())),
        }
    }
    Ok(())
}

/// Replicates every repo, removes the standby copies of repos that no longer exist, and reports.
/// A repo that fails is counted and does not stop the others.
pub async fn reconcile_once<T: StandbyTransport>(ctx: &Ctx<T>) -> ReconcileReport {
    let mut report = ReconcileReport::default();
    let ids = match ctx.store.repo_ids().await {
        Ok(ids) => ids,
        Err(e) => {
            report.failures.push(format!("cannot list repos: {e}"));
            return report;
        }
    };
    for id in &ids {
        match replicate_repo(ctx, id).await {
            Ok(()) => report.replicated += 1,
            Err(e) => report.failures.push(format!("{id}: {e}")),
        }
    }

    let statuses = match ctx.store.list_replication_status().await {
        Ok(statuses) => statuses,
        Err(e) => {
            report
                .failures
                .push(format!("cannot list replication status: {e}"));
            return report;
        }
    };
    for status in statuses {
        // A wiki's status belongs to its repo: it stays while the repo does.
        let owner = wiki_subject_repo(&status.subject).unwrap_or(&status.subject);
        if status.subject == DB_SUBJECT || ids.iter().any(|id| id == owner) {
            continue;
        }
        match remove_remote_repo(ctx, &status.subject).await {
            Ok(()) => {
                if let Err(e) = ctx.store.delete_replication_status(&status.subject).await {
                    report.failures.push(format!("{}: {e}", status.subject));
                }
            }
            Err(e) => report
                .failures
                .push(format!("{}: remove failed: {e}", status.subject)),
        }
    }
    report
}

/// Deletes exactly `<base>/repos/<id>.git` on the standby. The id is validated first.
async fn remove_remote_repo<T: StandbyTransport>(
    ctx: &Ctx<T>,
    subject: &str,
) -> Result<(), String> {
    let remote = match wiki_subject_repo(subject) {
        Some(repo_id) => ctx.layout.wiki_dir(repo_id)?,
        None => ctx.layout.repo_dir(subject)?,
    };
    ctx.transport
        .run(&format!("rm -rf -- {}", sh_quote(&remote)))
        .await
        .map(|_| ())
}

#[cfg(test)]
mod tests {
    use std::path::Path;

    use forge_core::Store;
    use forge_core::Visibility;
    use forge_core::store::replication::DB_SUBJECT;
    use forge_core::store::snapshot::verify_snapshot;
    use forge_git::GitEnv;
    use forge_git::backup::refs_checksum;

    use super::super::{Layout, LocalTransport, remote_refs_checksum};
    use super::{Ctx, ReconcileReport, reconcile_once, replicate_repo, ship_db};

    /// Plain git against a bare repo, for making commits in the fixture.
    fn plumbing(git_dir: &Path, args: &[&str]) -> String {
        let out = std::process::Command::new("git")
            .arg(format!("--git-dir={}", git_dir.display()))
            .args(args)
            .env("GIT_AUTHOR_NAME", "t")
            .env("GIT_AUTHOR_EMAIL", "t@example.com")
            .env("GIT_COMMITTER_NAME", "t")
            .env("GIT_COMMITTER_EMAIL", "t@example.com")
            .output()
            .expect("git runs");
        assert!(out.status.success(), "git {args:?}: {out:?}");
        String::from_utf8_lossy(&out.stdout).trim().to_string()
    }

    /// Points `refname` at a new commit with `message` in `git_dir`.
    fn commit_to(git_dir: &Path, refname: &str, message: &str) {
        let tree = plumbing(git_dir, &["hash-object", "-t", "tree", "-w", "--stdin"]);
        let commit = plumbing(git_dir, &["commit-tree", &tree, "-m", message]);
        plumbing(git_dir, &["update-ref", refname, &commit]);
    }

    struct Fixture {
        _dir: tempfile::TempDir,
        ctx: Ctx<LocalTransport>,
        standby: std::path::PathBuf,
        repo_id: String,
        local_repo: std::path::PathBuf,
    }

    async fn fixture() -> Fixture {
        let dir = tempfile::tempdir().expect("tempdir");
        let data_dir = dir.path().join("data");
        std::fs::create_dir_all(&data_dir).expect("data dir");
        let store = Store::open(&dir.path().join("forge.db"))
            .await
            .expect("store");
        let alice = store
            .create_user("Alice", None, "hash", false)
            .await
            .expect("user");
        let repo = store
            .create_repo(&alice.id, "demo", "", Visibility::Private, "main")
            .await
            .expect("repo");
        let git = GitEnv {
            git_bin: "git".into(),
            data_dir: data_dir.clone(),
        };
        let local_repo = forge_git::repo::repo_path(&data_dir, &repo.id).expect("path");
        forge_git::repo::init_bare(&git, &local_repo, "main")
            .await
            .expect("init bare");
        commit_to(&local_repo, "refs/heads/main", "first");

        let standby = dir.path().join("standby");
        std::fs::create_dir_all(&standby).expect("standby");
        let standby_str = standby.to_str().expect("utf8 path");
        let ctx = Ctx {
            store,
            git,
            data_dir,
            layout: Layout::new(standby_str).expect("layout"),
            transport: LocalTransport,
        };
        Fixture {
            _dir: dir,
            ctx,
            standby,
            repo_id: repo.id,
            local_repo,
        }
    }

    #[tokio::test]
    async fn push_records_matching_checksum_and_follows_new_refs() {
        let f = fixture().await;
        replicate_repo(&f.ctx, &f.repo_id)
            .await
            .expect("first push");

        let remote = f.ctx.layout.repo_dir(&f.repo_id).expect("remote dir");
        assert!(
            Path::new(&remote).join("HEAD").is_file(),
            "bare repo created"
        );
        let local_sum = refs_checksum(&f.ctx.git, &f.local_repo).await.expect("sum");
        let remote_sum = remote_refs_checksum(&f.ctx.transport, &remote)
            .await
            .expect("remote sum");
        assert_eq!(local_sum, remote_sum);
        let status = f
            .ctx
            .store
            .replication_status(&f.repo_id)
            .await
            .expect("status")
            .expect("row");
        assert!(status.last_ok_at.is_some());
        assert_eq!(status.refs_sha256.as_deref(), Some(local_sum.as_str()));
        assert_eq!(status.last_error, None);

        commit_to(&f.local_repo, "refs/heads/main", "second");
        commit_to(&f.local_repo, "refs/tags/v1", "tagged");
        replicate_repo(&f.ctx, &f.repo_id)
            .await
            .expect("second push");
        let local_sum = refs_checksum(&f.ctx.git, &f.local_repo).await.expect("sum");
        let remote_sum = remote_refs_checksum(&f.ctx.transport, &remote)
            .await
            .expect("remote sum");
        assert_eq!(
            local_sum, remote_sum,
            "tags and the new head reached the standby"
        );
    }

    #[tokio::test]
    async fn a_failed_push_is_recorded_and_keeps_the_last_success() {
        let f = fixture().await;
        replicate_repo(&f.ctx, &f.repo_id).await.expect("push");
        let ok = f
            .ctx
            .store
            .replication_status(&f.repo_id)
            .await
            .expect("status")
            .expect("row");

        // Break the local repo so the next run cannot read it.
        std::fs::remove_dir_all(&f.local_repo).expect("remove local repo");
        assert!(replicate_repo(&f.ctx, &f.repo_id).await.is_err());
        let failed = f
            .ctx
            .store
            .replication_status(&f.repo_id)
            .await
            .expect("status")
            .expect("row");
        assert!(failed.last_error.is_some());
        assert_eq!(failed.last_ok_at, ok.last_ok_at);
    }

    #[tokio::test]
    async fn deleted_repo_is_removed_from_the_standby_at_reconcile() {
        let f = fixture().await;
        let report = reconcile_once(&f.ctx).await;
        assert_eq!(
            report,
            ReconcileReport {
                replicated: 1,
                failures: vec![]
            }
        );

        let remote = f.ctx.layout.repo_dir(&f.repo_id).expect("remote dir");
        assert!(Path::new(&remote).is_dir());

        f.ctx
            .store
            .delete_repo(&f.repo_id)
            .await
            .expect("delete repo");
        let report = reconcile_once(&f.ctx).await;
        assert!(report.failures.is_empty(), "{:?}", report.failures);
        assert!(!Path::new(&remote).exists(), "standby copy is gone");
        assert!(
            f.ctx
                .store
                .replication_status(&f.repo_id)
                .await
                .expect("status")
                .is_none(),
            "status row is gone"
        );
        assert!(
            f.standby.join("repos").is_dir(),
            "only the repo dir is removed"
        );
    }

    #[tokio::test]
    async fn database_snapshot_ships_with_latest_link_and_prunes_to_24() {
        let f = fixture().await;
        // Older snapshots already on the standby: 30 names that sort before any real stamp.
        let db_dir = f.standby.join("db");
        std::fs::create_dir_all(&db_dir).expect("db dir");
        for i in 0..30 {
            std::fs::write(db_dir.join(format!("forge-19990101T{i:06}Z.db")), b"old")
                .expect("old snapshot");
        }
        std::fs::write(db_dir.join("forge-19990101T000000Z.db.tmp"), b"x").expect("stray");

        ship_db(&f.ctx).await.expect("ship");

        let mut names: Vec<String> = std::fs::read_dir(&db_dir)
            .expect("list")
            .map(|e| e.expect("entry").file_name().to_string_lossy().into_owned())
            .filter(|n| n.starts_with("forge-") && n.ends_with(".db"))
            .collect();
        names.sort();
        assert_eq!(names.len(), 24, "pruned to the newest 24: {names:?}");
        assert!(!db_dir.join("forge-19990101T000000Z.db").exists());
        assert!(db_dir.join("forge-19990101T000029Z.db").exists());
        assert!(
            db_dir.join("forge-19990101T000000Z.db.tmp").exists(),
            "foreign files stay"
        );

        let newest = names.last().expect("newest").clone();
        let link = std::fs::read_link(db_dir.join("latest")).expect("latest is a symlink");
        assert_eq!(
            link.to_string_lossy(),
            newest,
            "latest points at the new snapshot"
        );
        let info = verify_snapshot(&db_dir.join(&newest))
            .await
            .expect("shipped copy verifies");
        assert!(info.repos.iter().any(|r| r.0 == f.repo_id));

        let status = f
            .ctx
            .store
            .replication_status(DB_SUBJECT)
            .await
            .expect("status")
            .expect("row");
        assert!(status.last_ok_at.is_some());
        assert_eq!(status.last_error, None);
        assert!(
            !f.ctx.data_dir.join("run").join("replica-snap.db").exists(),
            "the local staging copy is removed"
        );
    }

    #[tokio::test]
    async fn reconcile_reports_failures_without_stopping_other_repos() {
        let f = fixture().await;
        std::fs::remove_dir_all(&f.local_repo).expect("remove local repo");
        let report = reconcile_once(&f.ctx).await;
        assert_eq!(report.replicated, 0);
        assert_eq!(report.failures.len(), 1, "{:?}", report.failures);
    }
}