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

    /// Push protection's scan budget (#268, `[limits]`).

    scan_budget: crate::push_protection::Budget,

    /// Repositories whose push protection scan stopped early, by id: the

    /// note goes on the push log entry of the push that moves references

    /// (the parts of a long push before it move none).

    scan_notes: Arc<std::sync::Mutex<std::collections::HashMap<i64, String>>>,

    /// The keys that signed what a push to each repository (by name)

    /// brought, noted when it is vetted and recorded once it is accepted

    /// (#286, Settings › Keys' "last used").

    signers: Arc<std::sync::Mutex<std::collections::HashMap<String, std::collections::BTreeSet<String>>>>,

}

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

            scan_budget: crate::push_protection::Budget::default(),

            scan_notes: Arc::default(),

            signers: Arc::default(),

        }

    }

    /// Scan pushes for secrets within `budget` (#268).

    pub fn with_scan_budget(mut self, budget: crate::push_protection::Budget) -> Self {

        self.scan_budget = budget;

        self

    }

    /// Who pushes, for the push log: the account (id 0 for a site token)

    /// and, for an agent's token, the agent's name.

    fn pusher(&self, headers: &HeaderMap) -> (Option<(i64, String)>, String) {

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

        }

    }

    /// 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 add_log(&self, name: &str, refs: &mut oxy_core::protocol::RefsResponse) {

        let Some(app) = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade()) else { return };

        let row = RepoName::parse(name).and_then(|p| self.row(&crate::wiki::parent_of(&p).0).ok()).map(|r| r.id);

        let backfill = crate::tlog::add_log(&app, row, refs);

        if !backfill.is_empty() {

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

        }

    }

    /// Tailor the refs answer to the client asking (oxygen#131): one that

    /// does not declare `certified-tags` in `Oxy-Features` (oxy 0.5.0 and

    /// older) is not sent a tag that only a certified key's certificate

    /// authorizes, which it would refuse and, with it, the whole clone.

    fn adjust_refs(&self, _name: &str, repo: &Repo, refs: &mut oxy_core::protocol::RefsResponse, headers: &HeaderMap) {

        oxy_server::tailor_refs(repo, refs, headers);

    }

    /// Push protection (hub spec §15.4, #268): a push adding what looks

    /// like a secret is refused, saying where, unless the repository turned

    /// it off or a maintainer allowed that finding's fingerprint.

    fn vet_push(

        &self,

        name: &str,

        repo: &Repo,

        bundle: &oxy_core::protocol::Bundle,

        updates: &[oxy_core::protocol::RefUpdate],

        headers: &HeaderMap,

    ) -> Result<(), Failure> {

        let keys: std::collections::BTreeSet<String> = bundle

            .changes

            .iter()

            .flat_map(|(_, sigs)| sigs.iter())

            .chain(bundle.attests.iter().flat_map(|a| a.sigs.iter()))

            .map(|sig| sig.key.to_hex())

            .collect();

        if !keys.is_empty() {

            self.signers.lock().unwrap().entry(name.to_string()).or_default().extend(keys);

        }

        if bundle.changes.is_empty() {

            return Ok(());

        }

        let parsed = RepoName::parse(name).ok_or_else(not_found)?;

        let row = self.row(&crate::wiki::parent_of(&parsed).0)?;

        if !row.push_protection {

            return Ok(());

        }

        let scan = crate::push_protection::scan(bundle, self.scan_budget);

        if let Some(note) = scan.note() {

            tracing::info!("push to {name}: {note}");

            self.scan_notes.lock().unwrap().insert(row.id, note);

        }

        if scan.findings.is_empty() {

            return Ok(());

        }

        let allowed = self.block(db::push_protection::allowed(&self.pool, row.id)).map_err(internal)?;

        let mut findings: Vec<_> = scan.findings.into_iter().filter(|f| !allowed.contains(&f.fingerprint)).collect();

        if findings.is_empty() {

            return Ok(());

        }

        crate::push_protection::locate(repo, bundle, &mut findings);

        let (pusher, via) = self.pusher(headers);

        let base = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade()).map(|a| a.config.base_url().to_string());

        let settings =

            if row.is_drop() { String::new() } else { format!("{}/{}/{}/settings", base.unwrap_or_default(), row.owner, row.name) };

        let text = crate::push_protection::refusal(&findings, &settings, &via);

        // Recorded for the maintainers (Settings › Push protection) and in

        // the push log, as every refused push is.

        let records: Vec<db::push_protection::NewFinding> = findings

            .iter()

            .map(|f| db::push_protection::NewFinding {

                rule: &f.rule,

                description: &f.description,

                path: &f.path,

                line: f.line as i64,

                fingerprint: &f.fingerprint,

                excerpt: &f.excerpt,

            })

            .collect();

        let who = pusher.as_ref().map(|(id, n)| (*id, n.as_str()));

        if let Err(e) = self.block(db::push_protection::record(&self.pool, row.id, who.map(|w| w.1).unwrap_or(""), &via, &records)) {

            tracing::warn!("recording push protection findings for {name}: {e}");

        }

        let reason = format!(

            "push protection: {} possible secret{} ({}){}",

            findings.len(),

            if findings.len() == 1 { "" } else { "s" },