//! 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>,

    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>, pool: db::Pool, site_tokens: Vec<String>, runtime: tokio::runtime::Handle) -> HubBackend {

        HubBackend { repos, 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),

            Ok(None) => Err(not_found()),

            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 name = RepoName::parse(name).ok_or_else(not_found)?;

        self.repos.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 row = self.row(&parsed)?;

        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"),

                };

                (role, needed && token.covers(&parsed.to_string()))

            }

        };

        // 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 !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)?;

        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 };

        let Ok(row) = self.row(&parsed) 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;

                }

            }

        });

    }

}

/// `/{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<()>),

}