//! The wire protocol, embedded (hub spec §3.3, §5.4): `oxy-server`'s

//! `Server` over a backend that resolves `owner/name`, applies visibility

//! and push tokens, and records pushes.

use crate::repos::{RepoName, Repos};

use crate::{db, Shared};

use axum::body::Bytes;

use axum::extract::{Path, State};

use axum::http::{HeaderMap, Method, StatusCode};

use axum::response::{IntoResponse, Response};

use hub_api::Visibility;

use oxy_core::repo::{PushReport, Repo};

use oxy_server::{Access, Backend, Endpoint, Failure};

use std::sync::Arc;

pub struct HubBackend {

    repos: Arc<Repos>,

    pool: db::Pool,

    push_tokens: Vec<String>,

    runtime: tokio::runtime::Handle,

}

impl HubBackend {

    pub fn new(repos: Arc<Repos>, pool: db::Pool, push_tokens: Vec<String>, runtime: tokio::runtime::Handle) -> HubBackend {

        HubBackend { repos, pool, push_tokens, runtime }

    }

    fn row(&self, name: &RepoName) -> Result<db::repos::RepoRow, Failure> {

        let pool = self.pool.clone();

        let (owner, repo) = (name.owner.clone(), name.name.clone());

        let row = tokio::task::block_in_place(|| self.runtime.block_on(db::repos::get(&pool, &owner, &repo)));

        match row {

            Ok(Some(r)) => Ok(r),

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

            Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),

        }

    }

    fn has_push_token(&self, headers: &HeaderMap) -> bool {

        match oxy_server::bearer_token(headers) {

            Some(t) => self.push_tokens.iter().any(|x| x == t),

            None => false,

        }

    }

}

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: Access, headers: &HeaderMap) -> Result<(), Failure> {

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

        let row = self.row(&parsed)?;

        let token = self.has_push_token(headers);

        // Until accounts exist (hub 0.2) a push token is the only credential:

        // it reads everything and may push everywhere.

        if row.visibility() != Visibility::Public && !token {

            return Err(not_found());

        }

        match access {

            Access::Read => Ok(()),

            Access::Push if self.push_tokens.is_empty() || token => Ok(()),

            Access::Push => Err(Failure(StatusCode::UNAUTHORIZED, "a valid bearer token is required to push".into())),

        }

    }

    fn after_push(&self, name: &str, report: &PushReport) {

        let Some(parsed) = RepoName::parse(name) else { return };

        let Ok(row) = self.row(&parsed) else { return };

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

        self.runtime.spawn(async move {

            if let Err(e) = db::repos::record_push(&pool, row.id, &results).await {

                tracing::warn!("recording push to {}: {e}", parsed);

            }

        });

    }

}

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

    };

    // Pushes and checkpoints write; everything else reads.

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

}