//! Page: standby replication (`/admin/replication`). Instance admins only; anyone else gets 404.
//!
//! Shows the `[replication]` target, the last database snapshot shipped, and one row per repo
//! with its health: ok, lagging (no success within twice the interval), error, or never replicated.
//! "Reconcile now" queues a full pass for the worker. It is a POST with a CSRF token, and it is
//! audited as `replication_reconcile_started`.

use std::collections::HashMap;
use std::time::Duration;

use forge_core::store::replication::{DB_SUBJECT, ReplicationStatus};
use forge_core::time::stamp_ago;
use forge_core::tz::format_in_tz;
use forge_core::{AuditAction, Error};
use serde::Deserialize;
use topcoat::Result;
use topcoat::context::{Cx, app_context};
use topcoat::router::content::Form;
use topcoat::router::error::{forbidden, see_other};
use topcoat::router::response::Response;
use topcoat::router::{page, query_params, route};
use topcoat::view::{View, component, view};

use crate::audit::{self, Actor, Target};
use crate::auth::check_csrf;
use crate::i18n::admin::{self as g, K};
use crate::i18n::{self, Key, Lang};
use crate::pages::admin_users::admin_only;
use crate::pages::issues_list::stamp_unix;
use crate::state::AppState;
use crate::views::{enc, shell};

#[query_params(error = bad_request)]
pub(crate) struct NoticeQuery {
    notice: Option<String>,
}

/// How a subject reads on the page.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Health {
    Ok,
    Lagging,
    Error,
    Never,
}

impl Health {
    fn label(self, lang: Lang) -> &'static str {
        match self {
            Health::Ok => g::t(lang, K::HealthOk),
            Health::Lagging => g::t(lang, K::HealthLagging),
            Health::Error => g::t(lang, K::HealthError),
            Health::Never => g::t(lang, K::HealthNever),
        }
    }
}

/// Health of one subject. `cutoff` is the oldest success that still counts as current; stamps are
/// fixed-width UTC, so a text comparison orders them.
fn health(status: Option<&ReplicationStatus>, cutoff: &str) -> Health {
    match status {
        None => Health::Never,
        Some(s) if s.last_error.is_some() => Health::Error,
        Some(s) => match s.last_ok_at.as_deref() {
            None => Health::Never,
            Some(ok) if ok >= cutoff => Health::Ok,
            Some(_) => Health::Lagging,
        },
    }
}

/// One line of the repos table, already formatted.
struct RepoRow {
    name: String,
    health: &'static str,
    last_ok: String,
    error: String,
}

/// Everything the page shows, already formatted.
struct PageData {
    enabled: bool,
    target: String,
    db_health: &'static str,
    db_last_ok: String,
    db_last_attempt: String,
    db_error: String,
    interval: String,
    healthcheck: &'static str,
    repos: Vec<RepoRow>,
}

fn ok_text(stamp: Option<&str>, tz: &str, lang: Lang) -> String {
    match stamp {
        Some(stamp) => format_in_tz(stamp_unix(stamp), tz),
        None => i18n::t(lang, Key::Never).to_owned(),
    }
}

/// Builds the page data from the config, the status rows and the repo list.
async fn page_data(state: &AppState, tz: &str, lang: Lang) -> forge_core::Result<PageData> {
    let config = &state.config.replication;
    let db_cutoff = stamp_ago(Duration::from_secs(config.db_snapshot_minutes * 60 * 2));
    let repo_cutoff = stamp_ago(Duration::from_secs(config.reconcile_minutes * 60 * 2));

    let statuses: HashMap<String, ReplicationStatus> = state
        .store
        .list_replication_status()
        .await?
        .into_iter()
        .map(|s| (s.subject.clone(), s))
        .collect();

    let db = statuses.get(DB_SUBJECT);
    let mut repos = Vec::new();
    let ids = state.store.repo_ids().await?;
    for id in &ids {
        let status = statuses.get(id);
        let name = match state.store.repo_by_id(id).await? {
            Some(repo) => format!("{}/{}", repo.owner_name, repo.name),
            None => id.clone(),
        };
        repos.push(RepoRow {
            name,
            health: health(status, &repo_cutoff).label(lang),
            last_ok: ok_text(status.and_then(|s| s.last_ok_at.as_deref()), tz, lang),
            error: status
                .and_then(|s| s.last_error.clone())
                .unwrap_or_default(),
        });
    }
    // Repos deleted since their last push: their standby copy is removed at the next reconcile.
    for status in statuses.values() {
        if status.subject != DB_SUBJECT && !ids.contains(&status.subject) {
            repos.push(RepoRow {
                name: g::deleted_repo_name(lang, &status.subject),
                health: health(Some(status), &repo_cutoff).label(lang),
                last_ok: ok_text(status.last_ok_at.as_deref(), tz, lang),
                error: status.last_error.clone().unwrap_or_default(),
            });
        }
    }

    Ok(PageData {
        enabled: config.enabled(),
        target: config.target.clone(),
        db_health: health(db, &db_cutoff).label(lang),
        db_last_ok: ok_text(db.and_then(|s| s.last_ok_at.as_deref()), tz, lang),
        db_last_attempt: ok_text(db.and_then(|s| s.last_attempt_at.as_deref()), tz, lang),
        db_error: db.and_then(|s| s.last_error.clone()).unwrap_or_default(),
        interval: g::interval_line(lang, config.db_snapshot_minutes, config.reconcile_minutes),
        healthcheck: if config.healthcheck_url.is_empty() {
            g::t(lang, K::HealthcheckNotSet)
        } else {
            g::t(lang, K::HealthcheckSet)
        },
        repos,
    })
}

#[component]
async fn replication_view(
    lang: Lang,
    csrf: String,
    data: PageData,
    notice: Option<String>,
) -> Result<impl View> {
    Ok(view! {
        <section class="page-head">
            <h1>(g::t(lang, K::Replication))</h1>
            <p class="muted">(g::t(lang, K::ReplicationIntro))</p>
        </section>
        if let Some(message) = notice {
            <section class="card notice" role="status">
                <p>(message.as_str())</p>
            </section>
        }
        <section class="card">
            <h2>(g::t(lang, K::Standby))</h2>
            if data.enabled {
                <dl class="details">
                    <dt>(g::t(lang, K::ReplicationTarget))</dt><dd><code>(data.target.as_str())</code></dd>
                    <dt>(g::t(lang, K::Interval))</dt><dd>(data.interval.as_str())</dd>
                    <dt>(g::t(lang, K::Healthcheck))</dt><dd>(data.healthcheck)</dd>
                    <dt>(g::t(lang, K::Database))</dt><dd>(data.db_health)</dd>
                    <dt>(g::t(lang, K::LastDbSnapshotShipped))</dt><dd class="small">(data.db_last_ok.as_str())</dd>
                    <dt>(g::t(lang, K::LastDbAttempt))</dt><dd class="small">(data.db_last_attempt.as_str())</dd>
                </dl>
                if !data.db_error.is_empty() {
                    <p class="error">(g::t(lang, K::DatabaseErrorPrefix))(data.db_error.as_str())</p>
                }
            } else {
                <p>(g::t(lang, K::ReplicationOff))</p>
            }
            <div class="actions">
                <form method="post" action="/admin/replication/reconcile">
                    <input type="hidden" name="csrf" value=(csrf.as_str())>
                    <button type="submit" class="primary">(g::t(lang, K::ReconcileNow))</button>
                </form>
            </div>
        </section>
        <section class="card">
            <h2>(g::t(lang, K::Repos))</h2>
            if data.repos.is_empty() {
                <p class="muted">(g::t(lang, K::NoReposYet))</p>
            } else {
                <div class="table-wrap">
                    <table class="table">
                        <thead>
                            <tr>
                                <th>(g::t(lang, K::TableRepo))</th>
                                <th>(g::t(lang, K::TableStatus))</th>
                                <th>(g::t(lang, K::LastSuccess))</th>
                                <th>(g::t(lang, K::TableError))</th>
                            </tr>
                        </thead>
                        <tbody>
                            for row in data.repos.into_iter() {
                                <tr>
                                    <td>(row.name.as_str())</td>
                                    <td>(row.health)</td>
                                    <td class="small">(row.last_ok.as_str())</td>
                                    <td class="small">(row.error.as_str())</td>
                                </tr>
                            }
                        </tbody>
                    </table>
                </div>
            }
        </section>
    })
}

#[page("/admin/replication")]
async fn replication_page(cx: &Cx) -> Result<impl View> {
    let viewer = admin_only(cx).await?;
    let query = query_params::<NoticeQuery>(cx)?;
    let lang = i18n::lang(cx, Some(&viewer.0));
    let state = app_context::<AppState>(cx);
    let data = page_data(state, viewer.0.timezone.as_str(), lang).await?;
    let csrf = viewer.1.csrf_token.clone();
    let title = g::t(lang, K::Replication);
    Ok(view! {
        shell(title: title, viewer: Some(&viewer), replication_view(lang: lang, csrf: csrf, data: data, notice: query.notice.clone()))
    })
}

#[derive(Deserialize)]
struct ReconcileForm {
    #[serde(default)]
    csrf: String,
}

#[route(POST "/admin/replication/reconcile")]
async fn replication_reconcile(cx: &Cx, Form(form): Form<ReconcileForm>) -> Result<Response> {
    let viewer = admin_only(cx).await?;
    if !check_csrf(&viewer.1, &form.csrf) {
        return Err(forbidden().into());
    }
    let lang = i18n::lang(cx, Some(&viewer.0));
    let state = app_context::<AppState>(cx);
    let notice = match reconcile_now(cx, state, &viewer.0, lang).await {
        Ok(()) => g::t(lang, K::ReconcileStarted).to_owned(),
        Err(Error::Config(message)) => message,
        Err(other) => other.to_string(),
    };
    let location = format!("/admin/replication?notice={}", enc(&notice));
    Err(see_other(&location).into())
}

/// Queues a reconcile pass and audits the request. Refused when replication is not configured.
async fn reconcile_now(
    cx: &Cx,
    state: &AppState,
    user: &forge_core::User,
    lang: Lang,
) -> forge_core::Result<()> {
    if !state.config.replication.enabled() {
        return Err(Error::Config(
            g::t(lang, K::ErrReplicationNotConfigured).to_owned(),
        ));
    }
    audit::record(
        &state.store,
        Some(cx),
        Actor::User(user),
        AuditAction::ReplicationReconcileStarted,
        Target {
            kind: "replication",
            id: "",
            name: "",
        },
        serde_json::json!({ "trigger": "manual" }),
    )
    .await;
    state.replication.request_reconcile();
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use forge_core::model::Visibility;
    use topcoat::router::request::Request;
    use topcoat::router::{Body, Router, to_bytes};

    use crate::auth::SESSION_COOKIE;
    use crate::test_support::{app_router, create_session, test_config, test_state};

    const ADMIN_TOKEN: &str = "replication-admin-session";
    const USER_TOKEN: &str = "replication-user-session";
    const CSRF: &str = "replication-csrf";
    const TARGET: &str = "sb@standby.example:/srv/forge-standby";

    struct Fixture {
        _dir: tempfile::TempDir,
        router: Router,
        state: AppState,
        repo_id: String,
    }

    /// An instance with an admin, a plain user and one repo. Replication is configured when asked.
    async fn fixture(configured: bool) -> 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 mut state = test_state(&data_dir).await;
        let mut config = test_config(&data_dir);
        if configured {
            config.replication.target = TARGET.into();
        }
        state.config = std::sync::Arc::new(config);
        let admin = state
            .store
            .create_user("root", None, "h", true)
            .await
            .expect("admin");
        let user = state
            .store
            .create_user("plain", None, "h", false)
            .await
            .expect("user");
        create_session(&state.store, &admin.id, ADMIN_TOKEN, CSRF).await;
        create_session(&state.store, &user.id, USER_TOKEN, CSRF).await;
        let repo = state
            .store
            .create_repo(&admin.id, "demo", "", Visibility::Private, "main")
            .await
            .expect("repo");
        let router = app_router(state.clone());
        Fixture {
            _dir: dir,
            router,
            state,
            repo_id: repo.id,
        }
    }

    async fn send(
        router: &Router,
        method: &str,
        uri: &str,
        token: Option<&str>,
        form: Option<String>,
    ) -> (u16, String) {
        let mut builder = Request::builder().method(method).uri(uri);
        if let Some(token) = token {
            builder = builder.header("cookie", format!("__Host-{SESSION_COOKIE}={token}"));
        }
        let body = match form {
            Some(form) => {
                builder = builder.header("content-type", "application/x-www-form-urlencoded");
                Body::from(form)
            }
            None => Body::empty(),
        };
        let res = router.handle(builder.body(body).expect("request")).await;
        let status = res.status().as_u16();
        let bytes = to_bytes(res.into_body(), usize::MAX).await.expect("body");
        (status, String::from_utf8_lossy(&bytes).into_owned())
    }

    #[tokio::test]
    async fn page_and_reconcile_are_404_for_anonymous_and_non_admins() {
        let fx = fixture(true).await;
        let (anon, _) = send(&fx.router, "GET", "/admin/replication", None, None).await;
        assert_eq!(anon, 404);
        let (user, _) = send(
            &fx.router,
            "GET",
            "/admin/replication",
            Some(USER_TOKEN),
            None,
        )
        .await;
        assert_eq!(user, 404);
        let (posted, _) = send(
            &fx.router,
            "POST",
            "/admin/replication/reconcile",
            Some(USER_TOKEN),
            Some(format!("csrf={CSRF}")),
        )
        .await;
        assert_eq!(posted, 404);
        assert!(!fx.state.replication.take_reconcile_request());
    }

    #[tokio::test]
    async fn page_shows_target_db_and_repo_health_without_secrets() {
        let fx = fixture(true).await;
        let store = &fx.state.store;
        store
            .record_replication_ok(DB_SUBJECT, None)
            .await
            .expect("db ok");
        store
            .record_replication_ok(&fx.repo_id, Some("abc"))
            .await
            .expect("repo ok");
        store
            .record_replication_error("gone-repo-id", "remove failed")
            .await
            .expect("stale");

        let (status, body) = send(
            &fx.router,
            "GET",
            "/admin/replication",
            Some(ADMIN_TOKEN),
            None,
        )
        .await;
        assert_eq!(status, 200);
        assert!(body.contains("sb@standby.example:/srv/forge-standby"));
        assert!(body.contains("root/demo"), "repo name is shown");
        assert!(body.contains("ok"));
        assert!(
            body.contains("gone-repo-id"),
            "stale rows are shown until cleaned"
        );
        assert!(body.contains("pending cleanup"));
        assert!(body.contains("Reconcile now"));
    }

    #[tokio::test]
    async fn unconfigured_page_says_so_and_reconcile_is_refused() {
        let fx = fixture(false).await;
        let (status, body) = send(
            &fx.router,
            "GET",
            "/admin/replication",
            Some(ADMIN_TOKEN),
            None,
        )
        .await;
        assert_eq!(status, 200);
        assert!(body.contains("Replication is off"));

        let (status, _) = send(
            &fx.router,
            "POST",
            "/admin/replication/reconcile",
            Some(ADMIN_TOKEN),
            Some(format!("csrf={CSRF}")),
        )
        .await;
        assert_eq!(status, 303);
        assert!(!fx.state.replication.take_reconcile_request());
        let audit = fx
            .state
            .store
            .list_audit(&Default::default(), None, 10)
            .await
            .expect("audit");
        assert!(
            !audit
                .iter()
                .any(|e| e.action == AuditAction::ReplicationReconcileStarted)
        );
    }

    #[tokio::test]
    async fn reconcile_needs_csrf_and_is_audited() {
        let fx = fixture(true).await;
        let (bad, _) = send(
            &fx.router,
            "POST",
            "/admin/replication/reconcile",
            Some(ADMIN_TOKEN),
            Some("csrf=wrong".into()),
        )
        .await;
        assert_eq!(bad, 403);
        assert!(!fx.state.replication.take_reconcile_request());

        let (status, _) = send(
            &fx.router,
            "POST",
            "/admin/replication/reconcile",
            Some(ADMIN_TOKEN),
            Some(format!("csrf={CSRF}")),
        )
        .await;
        assert_eq!(status, 303);
        assert!(fx.state.replication.take_reconcile_request());
        let audit = fx
            .state
            .store
            .list_audit(&Default::default(), None, 10)
            .await
            .expect("audit");
        assert!(
            audit
                .iter()
                .any(|e| e.action == AuditAction::ReplicationReconcileStarted),
            "the request is audited"
        );
    }

    #[test]
    fn health_reads_errors_before_age() {
        let cutoff = "2026-01-01T00:00:00.000000000Z";
        let fresh = ReplicationStatus {
            subject: "r".into(),
            last_ok_at: Some("2026-06-01T00:00:00.000000000Z".into()),
            last_error: None,
            last_attempt_at: None,
            refs_sha256: None,
        };
        assert_eq!(health(Some(&fresh), cutoff), Health::Ok);
        let old = ReplicationStatus {
            last_ok_at: Some("2025-06-01T00:00:00.000000000Z".into()),
            ..fresh.clone()
        };
        assert_eq!(health(Some(&old), cutoff), Health::Lagging);
        let failing = ReplicationStatus {
            last_error: Some("boom".into()),
            ..fresh.clone()
        };
        assert_eq!(health(Some(&failing), cutoff), Health::Error);
        assert_eq!(health(None, cutoff), Health::Never);
    }
}