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

        }

        // An agent's grant token (keyless agents, #241): its owner's

        // account, as an agent's token, writing when the grant lets its

        // keys move references. Revoking the grant stops it at once.

        let grant_token = presented.starts_with(crate::keyless::GRANT_TOKEN_PREFIX);

        let token = if grant_token {

            let grant = self.block(db::accounts::agent_grant_by_token(&pool, &hash)).map_err(internal)?;

            let Some(g) = grant.filter(|g| g.live(db::now_ms())) else {

                return Err(Failure(StatusCode::UNAUTHORIZED, "this agent grant is not valid (revoked, ended or unknown)".into()));

            };

            let _ = self.block(db::accounts::touch_agent_grant(&pool, g.id));

            Some(db::accounts::Token {

                id: 0,

                account_id: g.account_id,

                name: g.agent.clone(),

                hash: String::new(),

                scopes: if g.scope.split(',').any(|s| s.trim() == "refs") { "repo:read,repo:write" } else { "repo:read" }.into(),

                repos: String::new(),

                created_ms: g.created_ms,

                expires_ms: Some(g.expires_ms),

                last_used_ms: g.last_used_ms,

                renew_ms: None,

                agent: true,

            })

        } else {

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

        }

        if !grant_token {

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

        // The transparency log (#238): what the push brought, logged before

        // the push is answered.

        let app = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade());

        if let Some(app) = app {

            let to_log = self.repos.open(&parsed).map(|r| crate::tlog::push_entries(&r, Some(row.id), report)).unwrap_or_default();

            if !to_log.is_empty() {

                self.block(crate::tlog::log(&app, to_log));

            }

        }

        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) {