//! 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, nothing else (hub spec §13.2).
    Job {
        repo_id: 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 });
        }
        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()));
        }
        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 } = caller {
            // Only reads, only of the job's own repository.
            return match access {
                oxy_server::Access::Read if repo_id == 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 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, for `closes #N` (hub spec §8), when
        // the default bookmark moved.
        let messages: Vec<(String, String)> = if report.updated.iter().any(|u| u.name == row.default_bookmark) {
            self.repos
                .open(&parsed)
                .map(|r| report.imported.iter().filter_map(|h| r.store().get_change(h).ok().map(|c| (h.to_hex(), c.message))).collect())
                .unwrap_or_default()
        } else {
            vec![]
        };
        // Bookmarks that moved, with their new heads, for OxygenFlow runs.
        let moved: Vec<(String, String, Vec<oxy_core::hash::ChangeHash>)> = report
            .updated
            .iter()
            .filter(|u| 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);
            }
            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;
                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<()>),
}