//! The wire protocol, embedded (hub spec §3.3, §5.4): `oxy-server`'s
//! `Server` over a backend that resolves `owner/name`, authorises reads
//! and pushes by account role and token scope, and records who pushed.

use crate::auth::{self, Access, Role};
use crate::db::{self, accounts::Account};
use crate::repos::{RepoName, Repos};
use crate::Shared;
use axum::body::Bytes;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, Method, StatusCode};
use axum::response::{IntoResponse, Response};
use oxy_core::repo::{PushReport, Repo};
use oxy_server::{Backend, Endpoint, Failure};
use std::sync::Arc;

pub struct HubBackend {
    repos: Arc<Repos>,
    /// Wikis, reached as `OWNER/NAME.wiki` (hub spec §6.11).
    wikis: Arc<Repos>,
    pool: db::Pool,
    /// Site tokens from `hub.toml`: full access, for bootstrapping.
    site_tokens: Vec<String>,
    runtime: tokio::runtime::Handle,
    /// The application, set once it exists (the backend is built first).
    app_hint: Arc<std::sync::Mutex<Option<std::sync::Weak<crate::App>>>>,
}

/// Who a request's credential resolved to.
enum Caller {
    Anonymous,
    /// A site token from the configuration.
    Site,
    Token(Box<Account>, Box<db::accounts::Token>),
    /// A running OxygenFlow job: reads its own repository and the subrepos
    /// its state pins, nothing else (hub spec §13.2).
    Job {
        repo_id: i64,
        also: Vec<i64>,
    },
}

impl HubBackend {
    pub fn new(
        repos: Arc<Repos>,
        wikis: Arc<Repos>,
        pool: db::Pool,
        site_tokens: Vec<String>,
        runtime: tokio::runtime::Handle,
    ) -> HubBackend {
        HubBackend { repos, wikis, pool, site_tokens, runtime, app_hint: Arc::new(std::sync::Mutex::new(None)) }
    }

    /// The slot the application registers itself in after construction.
    pub fn app_slot(&self) -> Arc<std::sync::Mutex<Option<std::sync::Weak<crate::App>>>> {
        self.app_hint.clone()
    }

    fn block<T>(&self, f: impl std::future::Future<Output = T>) -> T {
        tokio::task::block_in_place(|| self.runtime.block_on(f))
    }

    fn row(&self, name: &RepoName) -> Result<db::repos::RepoRow, Failure> {
        let pool = self.pool.clone();
        match self.block(db::repos::get(&pool, &name.owner, &name.name)) {
            Ok(Some(r)) => Ok(r),
            // A transferred repository still answers to its old name (#108).
            Ok(None) => match self.block(db::repos::moved_to(&pool, &name.owner, &name.name)) {
                Ok(Some(r)) => Ok(r),
                Ok(None) => Err(not_found()),
                Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
            },
            Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
        }
    }

    fn caller(&self, headers: &HeaderMap) -> Result<Caller, Failure> {
        let Some(presented) = oxy_server::bearer_token(headers) else { return Ok(Caller::Anonymous) };
        if self.site_tokens.iter().any(|t| auth::mac_equal(t, presented)) {
            return Ok(Caller::Site);
        }
        let hash = auth::token_hash(presented);
        let pool = self.pool.clone();
        if presented.starts_with(crate::flow::JOB_TOKEN_PREFIX) {
            let job = self.block(db::flow::job_by_token(&pool, &hash, crate::flow::JOB_TOKEN_MAX_AGE_MS)).map_err(internal)?;
            let Some(job) = job else { return Err(Failure(StatusCode::UNAUTHORIZED, "this job token is not valid".into())) };
            let run = self.block(db::flow::run_by_id(&pool, job.run_id)).map_err(internal)?;
            let Some(run) = run else { return Err(Failure(StatusCode::UNAUTHORIZED, "this job token is not valid".into())) };
            return Ok(Caller::Job { repo_id: run.repo_id, also: job.also_read_ids() });
        }
        let token = self.block(db::accounts::token_by_hash(&pool, &hash)).map_err(internal)?;
        let Some(token) = token else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
        if token.expired(db::now_ms()) {
            return Err(Failure(StatusCode::UNAUTHORIZED, "this token has expired".into()));
        }
        let account = self.block(db::accounts::account_by_id(&pool, token.account_id)).map_err(internal)?;
        let Some(account) = account else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
        if account.banned {
            return Err(Failure(StatusCode::UNAUTHORIZED, "this account is banned".into()));
        }
        if account.disabled {
            return Err(Failure(StatusCode::UNAUTHORIZED, "this account is disabled; contact the hub's administrator".into()));
        }
        let _ = self.block(db::accounts::touch_token(&pool, token.id));
        Ok(Caller::Token(Box::new(account), Box::new(token)))
    }
}

fn internal(e: sqlx::Error) -> Failure {
    Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
}

fn not_found() -> Failure {
    Failure(StatusCode::NOT_FOUND, "no such repository".into())
}

impl Backend for HubBackend {
    fn open(&self, name: &str) -> Result<Repo, Failure> {
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
        // `NAME.wiki` is the repository's wiki (§6.11); the row knows the
        // current name, so a transferred repository still answers to its
        // old one (#108).
        let (parent, is_wiki) = crate::wiki::parent_of(&parsed);
        let row = self.row(&parent)?;
        let name = RepoName::new(&row.owner, &row.name).ok_or_else(not_found)?;
        let store = if is_wiki { &self.wikis } else { &self.repos };
        store.open(&name).map_err(|e| match e {
            crate::repos::RepoError::NotFound => not_found(),
            other => Failure(StatusCode::INTERNAL_SERVER_ERROR, other.to_string()),
        })
    }

    fn authorize(&self, name: &str, access: oxy_server::Access, headers: &HeaderMap) -> Result<(), Failure> {
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
        let (parent, _) = crate::wiki::parent_of(&parsed);
        let row = self.row(&parent)?;
        let caller = self.caller(headers)?;
        if let Caller::Job { repo_id, also } = &caller {
            // Only reads: of the job's own repository, and of the subrepos its
            // state pins that its pusher could read when the job was claimed.
            return match access {
                oxy_server::Access::Read if *repo_id == row.id || also.contains(&row.id) => Ok(()),
                oxy_server::Access::Read => Err(not_found()),
                oxy_server::Access::Push => Err(Failure(StatusCode::FORBIDDEN, "a job token cannot push".into())),
            };
        }
        let (role, scope_ok) = match &caller {
            Caller::Job { .. } => unreachable!(),
            Caller::Site => (Some(Role::Admin), true),
            Caller::Anonymous => (self.block(auth::roles::role_for(&self.pool, None, &row)).map_err(internal)?, true),
            Caller::Token(account, token) => {
                let role = self.block(auth::roles::role_for(&self.pool, Some(account), &row)).map_err(internal)?;
                let needed = match access {
                    oxy_server::Access::Read => {
                        token.has_scope("repo:read") || token.has_scope("repo:write") || token.has_scope("repo:admin")
                    }
                    oxy_server::Access::Push => token.has_scope("repo:write") || token.has_scope("repo:admin"),
                };
                let current = format!("{}/{}", row.owner, row.name);
                let covered = token.covers(&parsed.to_string()) || token.covers(&current) || token.covers(&parent.to_string());
                (role, needed && covered)
            }
        };
        // An unreadable repository does not exist for the caller (hub spec §4.4).
        if !auth::roles::allows(role, Access::Read) {
            return Err(not_found());
        }
        match access {
            oxy_server::Access::Read if scope_ok => Ok(()),
            oxy_server::Access::Read => Err(Failure(StatusCode::FORBIDDEN, "this token cannot read repositories".into())),
            oxy_server::Access::Push => {
                if matches!(caller, Caller::Anonymous) {
                    return Err(Failure(StatusCode::UNAUTHORIZED, "a personal access token is required to push".into()));
                }
                if !auth::roles::allows(role, Access::Write) {
                    return Err(Failure(StatusCode::FORBIDDEN, "you do not have write access to this repository".into()));
                }
                if row.archived {
                    return Err(Failure(StatusCode::FORBIDDEN, "this repository is archived and read-only".into()));
                }
                if !scope_ok {
                    return Err(Failure(StatusCode::FORBIDDEN, "this token lacks the repo:write scope for this repository".into()));
                }
                Ok(())
            }
        }
    }

    fn default_bookmark(&self, name: &str) -> Option<String> {
        let parsed = RepoName::parse(name)?;
        if crate::wiki::parent_of(&parsed).1 {
            return Some(crate::wiki::BOOKMARK.into());
        }
        self.row(&parsed).ok().map(|r| r.default_bookmark)
    }

    fn after_push(&self, name: &str, report: &PushReport, headers: &HeaderMap) {
        let Some(parsed) = RepoName::parse(name) else { return };
        // A wiki push starts no run and rings no webhook.
        if crate::wiki::parent_of(&parsed).1 {
            return;
        }
        let Ok(row) = self.row(&parsed) else { return };
        let Some(parsed) = RepoName::new(&row.owner, &row.name) else { return };
        let pusher = match self.caller(headers) {
            Ok(Caller::Token(account, _)) => Some((account.id, account.username)),
            Ok(Caller::Site) => Some((0, "site token".to_string())),
            _ => None,
        };
        let pool = self.pool.clone();
        let results: Vec<db::repos::PushOutcome> = report
            .results
            .iter()
            .map(|r| {
                let updated = report.updated.iter().find(|u| u.name == r.name);
                db::repos::PushOutcome {
                    reference: r.name.clone(),
                    ok: r.ok,
                    reason: r.reason.clone(),
                    old_state: updated.map(|u| u.old_state.to_hex()).unwrap_or_default(),
                    new_state: updated.map(|u| u.new_state.to_hex()).unwrap_or_default(),
                    heads: updated.map(|u| u.heads.len() as i64).unwrap_or(0),
                    changes: if r.ok { report.imported.len() as i64 } else { 0 },
                }
            })
            .collect();
        // Messages of the changes stored: `closes #N` when the default
        // bookmark moved, and `#N` references on any bookmark (hub spec §8).
        let bookmarks: Vec<String> = report
            .updated
            .iter()
            .filter(|u| !u.name.starts_with("tags/") && report.results.iter().any(|r| r.ok && r.name == u.name))
            .map(|u| u.name.clone())
            .collect();
        let imported: Vec<(String, String)> = if bookmarks.is_empty() {
            vec![]
        } else {
            self.repos
                .open(&parsed)
                .map(|r| {
                    report.imported.iter().filter_map(|h| r.store().get_change_meta(h).ok().map(|c| (h.to_hex(), c.message))).collect()
                })
                .unwrap_or_default()
        };
        let messages: Vec<(String, String)> = if bookmarks.contains(&row.default_bookmark) { imported.clone() } else { vec![] };
        // Authors whose identity is a successor (a rotation, a renewal, a
        // delegation) of an earlier one: each step kept the key pair, so
        // if the chain reaches a linked key it is the same person, and the
        // successor is linked too, without re-uploading it (hub spec §5.3).
        let successors: Vec<(String, String, String, Vec<u8>, String)> = self
            .repos
            .open(&parsed)
            .map(|r| {
                let mut seen = std::collections::HashSet::new();
                let mut out = Vec::new();
                for h in &report.imported {
                    let Ok(meta) = r.store().get_change_meta(h) else { continue };
                    let mut key = meta.author;
                    while !key.is_zero() && seen.insert(key) {
                        let Ok(id) = r.store().get_identity(&key) else { break };
                        let Some(prev) = id.previous else { break };
                        let Ok(before) = r.store().get_identity(&prev) else { break };
                        if before.ed25519_pk != id.ed25519_pk || before.mldsa65_pk != id.mldsa65_pk {
                            break;
                        }
                        out.push((key.to_hex(), id.name.clone(), id.email.clone(), id.encode(), prev.to_hex()));
                        key = prev;
                    }
                }
                out
            })
            .unwrap_or_default();
        let on = if bookmarks.contains(&row.default_bookmark) {
            row.default_bookmark.clone()
        } else {
            bookmarks.first().cloned().unwrap_or_default()
        };
        // Bookmarks that moved, with their new heads, for OxygenFlow runs.
        let moved: Vec<(String, String, Vec<oxy_core::hash::ChangeHash>)> = report
            .updated
            .iter()
            // A bookmark sent again unchanged starts nothing.
            .filter(|u| u.old_state != u.new_state && report.results.iter().any(|r| r.ok && r.name == u.name))
            .map(|u| (u.name.clone(), u.new_state.to_hex(), u.heads.clone()))
            .collect();
        let app_hint = self.app_hint.clone();
        self.runtime.spawn(async move {
            let who = pusher.as_ref().map(|(id, name)| (*id, name.as_str()));
            if let Err(e) = db::repos::record_push(&pool, row.id, who, &results).await {
                tracing::warn!("recording push to {}: {e}", parsed);
            }
            // Oldest first: a link made for one step lets the next step link.
            for (key, name, email, object, prev) in successors.iter().rev() {
                let Ok(None) = db::accounts::identity_by_key(&pool, key).await else { continue };
                let Ok(Some(earlier)) = db::accounts::identity_by_key(&pool, prev).await else { continue };
                match db::accounts::link_identity(&pool, earlier.account_id, key, name, email, object).await {
                    Ok(()) => {
                        tracing::info!("linked identity {} to account {} as a successor of {}", &key[..12], earlier.account_id, &prev[..12]);
                        db::accounts::audit(&pool, None, "identity.rotated", key, &format!("successor of {}", &prev[..12])).await.ok();
                    }
                    Err(e) => tracing::warn!("linking successor identity {}: {e}", &key[..12]),
                }
            }
            let app = app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade());
            if let Some(app) = app {
                crate::web::issues::close_from_messages(&app, &row, &messages).await;
                let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
                crate::web::issues::references_from_messages(&app, &row, &imported, &on, &actor).await;
                if results.iter().any(|r| r.ok && r.reference == row.default_bookmark) {
                    crate::mirror::refresh(&app, &row).await;
                }
                let accepted: Vec<serde_json::Value> = results
                    .iter()
                    .filter(|r| r.ok)
                    .map(|r| serde_json::json!({ "reference": r.reference, "old_state": r.old_state, "new_state": r.new_state, "heads": r.heads, "changes": r.changes }))
                    .collect();
                if !accepted.is_empty() {
                    let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
                    crate::hooks::emit(&app, &row, "push", &actor, serde_json::json!({ "updates": accepted })).await;
                }
                let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
                for (bookmark, state, heads) in &moved {
                    crate::flow::on_push(&app, &row, bookmark, state, heads, &actor).await;
                    crate::web::proposals::settle_after_push(&app, &row, bookmark, &actor).await;
                }
            }
        });
    }
}

/// `/{owner}/{name}/oxy/v1/{endpoint}`.
pub async fn dispatch(
    State(app): State<Shared>,
    Path((owner, name, endpoint)): Path<(String, String, String)>,
    method: Method,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let Some(endpoint) = Endpoint::parse(&endpoint) else {
        return Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response();
    };
    let Some(repo) = RepoName::new(&owner, &name) else {
        return not_found().into_response();
    };
    let _guard = match endpoint {
        Endpoint::Push | Endpoint::Checkpoint => Guard::Write(app.repos.write_guard(&repo).await),
        _ => Guard::Read(app.repos.read_guard(&repo).await),
    };
    app.server.handle(repo.to_string(), endpoint, &method, &headers, body).await
}

enum Guard {
    Read(#[allow(dead_code)] tokio::sync::OwnedRwLockReadGuard<()>),
    Write(#[allow(dead_code)] tokio::sync::OwnedRwLockWriteGuard<()>),
}