//! Backups (docs/backups.md, oxygenhub#169): each user and organisation

//! copies its repositories, on a schedule, to a destination it controls.

//!

//! A repository is its store directory (`.oxy/`): packs and loose objects

//! that never change once written, and small files that do (refs, the

//! operation log, policies, signature indexes). A run copies each file

//! once, named by its BLAKE3 (`blobs/`), and writes a manifest of path to

//! hash per repository (`runs/<at>/repos/`), so only what changed travels

//! and any kept run can be restored byte for byte, signatures and all.

//! The large files are copied first; the walk is then repeated under the

//! repository's read lock, which pushes wait on, so each manifest is one

//! consistent moment. Beside each manifest go the hub's records for the

//! repository as JSON (never a secret value), the owner's records, and

//! last a `DONE` marker: a run without one was cut short and is ignored.

use crate::db::{self, Pool};

use crate::repos::RepoName;

use crate::Shared;

use anyhow::{bail, Context, Result};

use serde::{Deserialize, Serialize};

use std::collections::HashMap;

use std::path::{Path, PathBuf};

/// How often, as the settings offer it.

pub const EVERY: &[(&str, &str)] = &[("off", "Off"), ("hourly", "Every hour"), ("daily", "Every day")];

/// How long runs are kept, in days.

pub const KEEP: &[i64] = &[7, 30, 90];

/// What an owner backs up, where and how often.

#[derive(Clone, Debug, sqlx::FromRow, Serialize)]

pub struct Setting {

    pub owner: String,

    pub kind: String,

    pub every: String,

    pub destination: String,

    pub target: String,

    pub prefix: String,

    pub provider: String,

    pub keep_days: i64,

    pub updated_ms: i64,

}

impl Setting {

    pub fn off(owner: &str, kind: &str) -> Setting {

        Setting {

            owner: owner.into(),

            kind: kind.into(),

            every: "off".into(),

            destination: "dir".into(),

            target: String::new(),

            prefix: String::new(),

            provider: String::new(),

            keep_days: 30,

            updated_ms: 0,

        }

    }

    /// Seconds between runs; none when off.

    pub fn period_s(&self) -> Option<i64> {

        match self.every.as_str() {

            "hourly" => Some(3600),

            "daily" => Some(86_400),

            _ => None,

        }

    }

}

/// One run, as the settings page and the API show it.

#[derive(Clone, Debug, sqlx::FromRow, Serialize)]

pub struct Run {

    pub id: i64,

    pub owner: String,

    pub started_ms: i64,

    pub finished_ms: Option<i64>,

    pub status: String,

    pub at: String,

    pub repos: i64,

    pub files: i64,

    pub sent: i64,

    pub bytes: i64,

    pub detail: String,

    pub trigger: String,

}

pub async fn setting(pool: &Pool, owner: &str) -> sqlx::Result<Option<Setting>> {

    sqlx::query_as("SELECT * FROM backup_settings WHERE owner = ?").bind(owner).fetch_optional(pool).await

}

pub async fn save(pool: &Pool, s: &Setting) -> sqlx::Result<()> {

    sqlx::query(

        "INSERT INTO backup_settings (owner, kind, every, destination, target, prefix, provider, keep_days, updated_ms) \

         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(owner) DO UPDATE SET kind = excluded.kind, every = excluded.every, \

         destination = excluded.destination, target = excluded.target, prefix = excluded.prefix, provider = excluded.provider, \

         keep_days = excluded.keep_days, updated_ms = excluded.updated_ms",

    )

    .bind(&s.owner)

    .bind(&s.kind)

    .bind(&s.every)

    .bind(&s.destination)

    .bind(&s.target)

    .bind(&s.prefix)

    .bind(&s.provider)

    .bind(s.keep_days)

    .bind(db::now_ms())

    .execute(pool)

    .await?;

    Ok(())

}

pub async fn runs(pool: &Pool, owner: &str, limit: i64) -> sqlx::Result<Vec<Run>> {

    sqlx::query_as("SELECT * FROM backup_runs WHERE owner = ? ORDER BY id DESC LIMIT ?").bind(owner).bind(limit).fetch_all(pool).await

}

// ---------------------------------------------------------------- where

/// Where backups are written: keys are `/`-separated paths.

pub trait Destination: Send + Sync {

    fn exists(&self, key: &str) -> Result<bool>;

    /// Write `bytes` at `key`, whole or not at all.

    fn put(&self, key: &str, bytes: &[u8]) -> Result<()>;

    fn get(&self, key: &str) -> Result<Vec<u8>>;

    /// The names directly under `prefix` (files and "directories").

    fn list(&self, prefix: &str) -> Result<Vec<String>>;

    /// Remove `key` and everything under it.

    fn remove(&self, key: &str) -> Result<()>;

}

/// A directory on the hub's host.

pub struct Directory {

    pub root: PathBuf,

}

impl Directory {

    fn path(&self, key: &str) -> Result<PathBuf> {

        if key.split('/').any(|p| p == ".." || p == ".") || key.starts_with('/') {

            bail!("not a clean key: {key}");

        }

        Ok(self.root.join(key))

    }

}

impl Destination for Directory {

    fn exists(&self, key: &str) -> Result<bool> {

        Ok(self.path(key)?.exists())

    }

    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {

        let p = self.path(key)?;

        std::fs::create_dir_all(p.parent().context("no parent")?)?;

        let tmp = p.with_extension(format!("part-{}", std::process::id()));

        std::fs::write(&tmp, bytes)?;

        std::fs::rename(&tmp, &p)?;

        Ok(())

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        let p = self.path(key)?;

        std::fs::read(&p).with_context(|| format!("read {}", p.display()))

    }

    fn list(&self, prefix: &str) -> Result<Vec<String>> {

        let p = self.path(prefix)?;

        let mut out: Vec<String> = match std::fs::read_dir(&p) {

            Ok(d) => d.flatten().map(|e| e.file_name().to_string_lossy().into_owned()).collect(),

            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),

            Err(e) => return Err(e.into()),

        };

        out.sort();

        Ok(out)

    }

    fn remove(&self, key: &str) -> Result<()> {

        let p = self.path(key)?;

        if p.is_dir() {

            std::fs::remove_dir_all(p)?;

        } else if p.exists() {

            std::fs::remove_file(p)?;

        }

        Ok(())

    }

}

/// A Cloud Storage bucket, reached keylessly (docs/backups.md §3, §6):

/// the hub mints a short token as `backup:OWNER`, Google's STS exchanges

/// it for an access token through the owner's workload identity

/// provider, and that writes the bucket. No key exists anywhere; the

/// access token lasts an hour and is fetched again before it runs out.

pub struct Gcs {

    bucket: String,

    /// Keys go under this, when set.

    prefix: String,

    storage: String,

    sts: String,

    /// The provider's resource name, `//iam.googleapis.com/projects/…`.

    provider: String,

    subject: String,

    app: Shared,

    client: reqwest::Client,

    rt: tokio::runtime::Handle,

    token: std::sync::Mutex<Option<(String, i64)>>,

}

impl Gcs {

    pub fn new(app: &Shared, s: &Setting) -> Gcs {

        Gcs {

            bucket: s.target.clone(),

            prefix: s.prefix.trim_matches('/').to_string(),

            storage: app.config.backups.storage_url.trim_end_matches('/').to_string(),

            sts: app.config.backups.sts_url.trim_end_matches('/').to_string(),

            provider: s.provider.clone(),

            subject: format!("backup:{}", s.owner),

            app: app.clone(),

            client: reqwest::Client::builder().timeout(std::time::Duration::from_secs(300)).build().unwrap_or_default(),

            rt: tokio::runtime::Handle::current(),

            token: std::sync::Mutex::new(None),

        }

    }

    fn name(&self, key: &str) -> String {

        if self.prefix.is_empty() {

            key.to_string()

        } else {

            format!("{}/{key}", self.prefix)

        }

    }

    /// An access token, exchanged for a fresh one of the hub's when the

    /// last is within five minutes of running out.

    fn access(&self) -> Result<String> {

        let now = db::now_ms();

        if let Some((t, until)) = self.token.lock().unwrap().as_ref() {

            if *until - now > 300_000 {

                return Ok(t.clone());

            }

        }

        let (token, expires_in) = self.rt.block_on(async {

            let issuer = self.app.oidc().await?;

            // Google's default audience for a provider: its name over https.

            let audience = format!("https:{}", self.provider);

            let jwt = issuer.mint_for(self.app.config.base_url(), &self.subject, &audience, db::now_ms(), 600);

            let res = self

                .client

                .post(format!("{}/v1/token", self.sts))

                .json(&serde_json::json!({

                    "grantType": "urn:ietf:params:oauth:grant-type:token-exchange",

                    "audience": self.provider,

                    "scope": "https://www.googleapis.com/auth/devstorage.read_write",

                    "requestedTokenType": "urn:ietf:params:oauth:token-type:access_token",

                    "subjectToken": jwt,

                    "subjectTokenType": "urn:ietf:params:oauth:token-type:jwt",

                }))

                .send()

                .await?;

            let status = res.status();

            let body: serde_json::Value = res.json().await.unwrap_or_default();

            if !status.is_success() {

                bail!(

                    "Google's token exchange refused {}: {} (check the provider's issuer is this hub's address and its attribute condition admits backup:)",

                    self.subject,

                    body["error_description"].as_str().or(body["error"].as_str()).unwrap_or("no reason given")

                );

            }

            let token = body["access_token"].as_str().context("the token exchange gave no access token")?.to_string();

            anyhow::Ok((token, body["expires_in"].as_i64().unwrap_or(3600)))

        })?;

        *self.token.lock().unwrap() = Some((token.clone(), now + expires_in * 1000));

        Ok(token)

    }

    fn object_url(&self, key: &str) -> Result<reqwest::Url> {

        let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

        u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&self.name(key));

        Ok(u)

    }

    fn refused(&self, what: &str, status: reqwest::StatusCode, body: &str) -> anyhow::Error {

        let hint = if status == reqwest::StatusCode::FORBIDDEN || status == reqwest::StatusCode::UNAUTHORIZED {

            format!(" (grant roles/storage.objectAdmin on gs://{} to the principal …/subject/{})", self.bucket, self.subject)

        } else {

            String::new()

        };

        anyhow::anyhow!("GCS {what}: {status}{hint}: {}", body.chars().take(200).collect::<String>())

    }

    /// Every object name under `full` (no delimiter), for removal.

    fn all_under(&self, full: &str) -> Result<Vec<String>> {

        let mut out = Vec::new();

        let mut page: Option<String> = None;

        loop {

            let token = self.access()?;

            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

            u.query_pairs_mut().append_pair("prefix", full).append_pair("fields", "items(name),nextPageToken");

            if let Some(p) = &page {

                u.query_pairs_mut().append_pair("pageToken", p);

            }

            let body: serde_json::Value = self.rt.block_on(async {

                let res = self.client.get(u).bearer_auth(&token).send().await?;

                let status = res.status();

                let text = res.text().await?;

                if !status.is_success() {

                    return Err(self.refused("list", status, &text));

                }

                anyhow::Ok(serde_json::from_str(&text)?)

            })?;

            out.extend(body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str().map(String::from)));

            match body["nextPageToken"].as_str() {

                Some(p) => page = Some(p.to_string()),

                None => return Ok(out),

            }

        }

    }

}

impl Destination for Gcs {

    fn exists(&self, key: &str) -> Result<bool> {

        let token = self.access()?;

        let mut u = self.object_url(key)?;

        u.query_pairs_mut().append_pair("fields", "name");

        self.rt.block_on(async {

            let res = self.client.get(u).bearer_auth(&token).send().await?;

            match res.status() {

                s if s.is_success() => Ok(true),

                reqwest::StatusCode::NOT_FOUND => Ok(false),

                s => Err(self.refused("read", s, &res.text().await.unwrap_or_default())),

            }

        })

    }

    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {

        let token = self.access()?;

        let mut u = reqwest::Url::parse(&format!("{}/upload/storage/v1/b/{}/o", self.storage, self.bucket))?;

        u.query_pairs_mut().append_pair("uploadType", "media").append_pair("name", &self.name(key));

        let body = bytes.to_vec();

        self.rt.block_on(async {

            let res = self.client.post(u).bearer_auth(&token).header("content-type", "application/octet-stream").body(body).send().await?;

            let status = res.status();

            if !status.is_success() {

                return Err(self.refused("write", status, &res.text().await.unwrap_or_default()));

            }

            Ok(())

        })

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        let token = self.access()?;

        let mut u = self.object_url(key)?;

        u.query_pairs_mut().append_pair("alt", "media");

        self.rt.block_on(async {

            let res = self.client.get(u).bearer_auth(&token).send().await?;

            let status = res.status();

            if !status.is_success() {

                return Err(self.refused(&format!("read {key}"), status, &res.text().await.unwrap_or_default()));

            }

            Ok(res.bytes().await?.to_vec())

        })

    }

    fn list(&self, prefix: &str) -> Result<Vec<String>> {

        let full = format!("{}/", self.name(prefix.trim_end_matches('/')));

        let mut out = std::collections::BTreeSet::new();

        let mut page: Option<String> = None;

        loop {

            let token = self.access()?;

            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

            u.query_pairs_mut()

                .append_pair("prefix", &full)

                .append_pair("delimiter", "/")

                .append_pair("fields", "items(name),prefixes,nextPageToken");

            if let Some(p) = &page {

                u.query_pairs_mut().append_pair("pageToken", p);

            }

            let body: serde_json::Value = self.rt.block_on(async {

                let res = self.client.get(u).bearer_auth(&token).send().await?;

                let status = res.status();

                let text = res.text().await?;

                if !status.is_success() {

                    return Err(self.refused("list", status, &text));

                }

                anyhow::Ok(serde_json::from_str(&text)?)

            })?;

            for n in body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str()) {

                out.insert(n.trim_start_matches(&full).to_string());

            }

            for p in body["prefixes"].as_array().into_iter().flatten().filter_map(|p| p.as_str()) {

                out.insert(p.trim_start_matches(&full).trim_end_matches('/').to_string());

            }

            match body["nextPageToken"].as_str() {

                Some(p) => page = Some(p.to_string()),

                None => return Ok(out.into_iter().filter(|n| !n.is_empty()).collect()),

            }

        }

    }

    fn remove(&self, key: &str) -> Result<()> {

        let mut names = self.all_under(&format!("{}/", self.name(key)))?;

        names.push(self.name(key));

        for n in names {

            let token = self.access()?;

            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

            u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&n);

            self.rt.block_on(async {

                let res = self.client.delete(u).bearer_auth(&token).send().await?;

                match res.status() {

                    s if s.is_success() || s == reqwest::StatusCode::NOT_FOUND => Ok(()),

                    s => Err(self.refused("delete", s, &res.text().await.unwrap_or_default())),

                }

            })?;

        }

        Ok(())

    }

}

/// Check a destination end to end: write a probe, read it back, remove it.

pub fn probe(dest: &dyn Destination) -> Result<()> {

    let marker = format!("probe {}", db::now_ms());

    dest.put(".oxyhub-probe", marker.as_bytes())?;

    if dest.get(".oxyhub-probe")? != marker.as_bytes() {

        bail!("the probe read back differently");

    }

    dest.remove(".oxyhub-probe")

}

/// The owner's destination, from its setting and the site's rules.

pub fn destination(app: &Shared, s: &Setting) -> Result<Box<dyn Destination>> {

    match s.destination.as_str() {

        "dir" => {

            let allowed = &app.config.backups.directories;

            let target = PathBuf::from(&s.target);

            if !allowed.contains(&target) {

                bail!("{} is not a backup directory this hub allows (hub.toml [backups] directories)", s.target);

            }

            let mut root = target.join(&s.owner);