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

    /// The hub's key, which countersigns when it accepted an update a key

    /// it certified signed (oxygenhub#241 step 4c, Oxygen spec §14.7).

    acceptor: Option<Arc<(oxy_core::identity::Identity, oxy_core::identity::SecretKey)>>,

}

/// 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)), acceptor: None }

    }

    /// Countersign accepted updates by keys this hub certified with `identity`.

    pub fn with_acceptor(mut self, identity: oxy_core::identity::Identity, secret: oxy_core::identity::SecretKey) -> Self {

        self.acceptor = Some(Arc::new((identity, secret)));

        self

    }

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

        let repo = store.open(&name).map_err(|e| match e {

            crate::repos::RepoError::NotFound => not_found(),

            other => Failure(StatusCode::INTERNAL_SERVER_ERROR, other.to_string()),

        })?;

        Ok(match &self.acceptor {

            Some(a) => repo.with_acceptor(a.0.clone(), a.1.clone()),

            None => repo,

        })

    }

    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());

        }

        // A wiki is made, empty and unbound, when someone who may write to

        // its repository first reaches it, so the first page can be pushed

        // as well as written on the web (§6.11); the push gives it its id.

        let (_, is_wiki) = crate::wiki::parent_of(&parsed);

        let may_write = match &caller {

            Caller::Site => true,

            Caller::Token(_, token) => token.has_scope("repo:write") || token.has_scope("repo:admin"),

            _ => false,

        };

        if is_wiki && may_write && auth::roles::allows(role, Access::Write) && !row.archived {

            if let Some(name) = RepoName::new(&row.owner, &row.name) {

                if !self.wikis.exists(&name) {

                    self.wikis

                        .create(&name, None)

                        .map_err(|e| Failure(StatusCode::INTERNAL_SERVER_ERROR, format!("make the wiki: {e}")))?;

                }

            }

        }

        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, via) = match self.caller(headers) {

            Ok(Caller::Token(account, token)) => {

                (Some((account.id, account.username)), if token.agent { token.name.clone() } else { String::new() })

            }

            Ok(Caller::Site) => (Some((0, "site token".to_string())), String::new()),

            _ => (None, String::new()),

        };

        let pool = self.pool.clone();

        // A part of a long push (objects alone) moves nothing: the push it

        // belongs to is recorded when its last part names the references.

        let results: Vec<db::repos::PushOutcome> = report

            .results

            .iter()

            .filter(|r| r.name != oxy_core::repo::OBJECTS_RESULT)

            .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_via(&pool, row.id, who, &via, &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 {

                // Issues close from every change newly on the default

                // bookmark, not only those this push imported: a client's

                // landing moves main to changes the hub already held.

                if let Some(r) = results.iter().find(|r| r.ok && r.reference == row.default_bookmark) {

                    let new_heads = moved.iter().find(|(b, _, _)| *b == row.default_bookmark).map(|(_, _, h)| h.clone()).unwrap_or_default();

                    let old = r.old_state.clone();

                    let old_heads = crate::repos::RepoName::new(&row.owner, &row.name);

                    let old_heads = match (old_heads, oxy_core::hash::ObjectRef::from_hex(&old)) {

                        (Some(n), Some(s)) => app.repos.read(&n, move |repo, _| Ok(repo.store().get_state(&s).map(|st| st.heads).unwrap_or_default())).await.unwrap_or_default(),

                        _ => Vec::new(),

                    };

                    crate::web::issues::close_from_arrived(&app, &row, old_heads, new_heads, None).await;

                } else {

                    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();