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

    }

}

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

            if !s.prefix.is_empty() {

                root = root.join(s.prefix.trim_matches('/'));

            }

            Ok(Box::new(Directory { root }))

        }

        "gcs" => bail!("GCS destinations come with oxygenhub#169's next step"),

        other => bail!("unknown destination {other:?}"),

    }

}

// ---------------------------------------------------------------- what

/// One file of a repository's store, in a manifest.

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]

pub struct File {

    pub path: String,

    pub size: u64,

    /// Its modification time in nanoseconds: unchanged size and time mean

    /// unchanged content, so it is not hashed again.

    pub mtime: i128,

    pub hash: String,

}

/// A repository at one run.

#[derive(Clone, Debug, Serialize, Deserialize)]

pub struct Manifest {

    pub owner: String,

    pub name: String,

    /// `repos` or `wikis`.

    pub kind: String,

    pub files: Vec<File>,

}

fn blob_key(hash: &str) -> String {

    format!("blobs/{}/{hash}", &hash[..2])

}

/// The large files that never change once written come first.

fn immutable(path: &str) -> bool {

    path.starts_with(".oxy/objects/") && (path.contains("/pack-") || path.contains("/loose/"))

}

#[derive(Default)]

struct Sent {

    files: i64,

    sent: i64,

    bytes: i64,

}

/// Walk `dir`, sending each file the destination lacks; `known` says which

/// paths are unchanged since the last run (and their hashes).

fn copy_dir(dest: &dyn Destination, dir: &Path, known: &HashMap<String, File>, only_immutable: bool, sent: &mut Sent) -> Result<Vec<File>> {

    let mut out = Vec::new();

    let mut stack = vec![dir.to_path_buf()];

    while let Some(d) = stack.pop() {

        let entries = match std::fs::read_dir(&d) {

            Ok(e) => e,

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

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

        };

        for e in entries.flatten() {

            let p = e.path();

            let ft = e.file_type()?;

            if ft.is_dir() {

                stack.push(p);

                continue;

            }

            if !ft.is_file() {

                continue;

            }

            let rel = p.strip_prefix(dir).unwrap_or(&p).to_string_lossy().replace('\\', "/");

            if only_immutable && !immutable(&rel) {

                continue;

            }

            // A file packed away while the walk ran is simply gone.

            let Ok(md) = std::fs::metadata(&p) else { continue };

            let mtime =

                md.modified().ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()).map(|d| d.as_nanos() as i128).unwrap_or(0);

            let size = md.len();

            sent.files += 1;

            if let Some(k) = known.get(&rel).filter(|k| k.size == size && k.mtime == mtime) {

                out.push(k.clone());

                continue;

            }

            let Ok(bytes) = std::fs::read(&p) else { continue };

            let hash = blake3::hash(&bytes).to_hex().to_string();

            let key = blob_key(&hash);

            if !dest.exists(&key)? {

                dest.put(&key, &bytes)?;

                sent.sent += 1;

                sent.bytes += bytes.len() as i64;

            }

            out.push(File { path: rel, size, mtime, hash });

        }

    }

    out.sort_by(|a, b| a.path.cmp(&b.path));

    Ok(out)

}

// ---------------------------------------------------------------- records

/// Tables and columns never written: secrets, credentials, sessions.

fn private_table(t: &str) -> bool {

    ["secret", "token", "session", "password", "totp", "recovery", "sqlx"].iter().any(|w| t.contains(w))

}

fn private_column(c: &str) -> bool {

    ["secret", "token", "password", "salt", "totp", "sealed", "hash_key"].iter().any(|w| c.contains(w))

}

async fn tables(pool: &Pool) -> Result<Vec<String>> {

    let mut t: Vec<String> =

        sqlx::query_scalar("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'").fetch_all(pool).await?;

    t.retain(|t| !private_table(t));

    t.sort();

    Ok(t)

}

async fn columns(pool: &Pool, table: &str) -> Result<Vec<String>> {

    Ok(sqlx::query_scalar(&format!("SELECT name FROM pragma_table_info('{table}')")).fetch_all(pool).await?)

}

/// The rows of `table` matching `cond`, as a JSON array, private columns

/// left out and blobs as hex.

async fn rows(pool: &Pool, table: &str, cond: &str) -> Result<serde_json::Value> {

    let cols = columns(pool, table).await?;

    let fields: Vec<String> = cols

        .iter()

        .filter(|c| !private_column(c))

        .map(|c| format!("'{c}', CASE WHEN typeof(\"{c}\") = 'blob' THEN hex(\"{c}\") ELSE \"{c}\" END"))

        .collect();

    if fields.is_empty() {

        return Ok(serde_json::json!([]));

    }

    let sql = format!("SELECT json_group_array(json_object({})) FROM \"{table}\" WHERE {cond}", fields.join(", "));

    let text: String = sqlx::query_scalar(&sql).fetch_one(pool).await?;

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

}

/// Every row about something: the tables with column `key` equal to `id`,

/// then, three levels deep, the tables whose foreign keys point at rows

/// already taken (an issue's comments, a proposal's reviews and their

/// threads).

async fn records(pool: &Pool, key: &str, id: i64) -> Result<serde_json::Value> {

    let all = tables(pool).await?;

    let mut conds: Vec<(String, String)> = Vec::new();

    for t in &all {

        if columns(pool, t).await?.iter().any(|c| c == key) {

            conds.push((t.clone(), format!("\"{key}\" = {id}")));

        }

    }

    for _ in 0..3 {

        let mut more = Vec::new();

        for t in &all {

            if conds.iter().any(|(c, _)| c == t) || more.iter().any(|(c, _): &(String, String)| c == t) {

                continue;

            }

            let fks: Vec<(String, String, String)> =

                sqlx::query_as(&format!("SELECT \"table\", \"from\", \"to\" FROM pragma_foreign_key_list('{t}')")).fetch_all(pool).await?;

            if let Some((parent, from, to, pcond)) = fks

                .iter()

                .find_map(|(p, f, to)| conds.iter().find(|(c, _)| c == p).map(|(_, pc)| (p.clone(), f.clone(), to.clone(), pc.clone())))

            {

                more.push((t.clone(), format!("\"{from}\" IN (SELECT \"{to}\" FROM \"{parent}\" WHERE {pcond})")));

            }

        }

        if more.is_empty() {

            break;

        }

        conds.extend(more);

    }

    let mut out = serde_json::Map::new();

    for (t, cond) in conds {

        let r = rows(pool, &t, &cond).await?;

        if r.as_array().is_some_and(|a| !a.is_empty()) {

            out.insert(t, r);

        }

    }

    Ok(serde_json::Value::Object(out))

}

// ---------------------------------------------------------------- a run

fn run_name(ms: i64) -> String {

    let t = time::OffsetDateTime::from_unix_timestamp_nanos(ms as i128 * 1_000_000).unwrap_or(time::OffsetDateTime::UNIX_EPOCH);

    format!("{:04}{:02}{:02}T{:02}{:02}{:02}Z", t.year(), u8::from(t.month()), t.day(), t.hour(), t.minute(), t.second())

}

/// The last whole run's manifests, by `kind/name`.

fn last_manifests(dest: &dyn Destination) -> HashMap<String, HashMap<String, File>> {

    let mut out = HashMap::new();

    let Ok(at) = dest.get("LATEST").map(|b| String::from_utf8_lossy(&b).trim().to_string()) else { return out };

    for name in dest.list(&format!("runs/{at}/repos")).unwrap_or_default() {

        if let Ok(m) = dest.get(&format!("runs/{at}/repos/{name}")).and_then(|b| Ok(serde_json::from_slice::<Manifest>(&b)?)) {

            out.insert(format!("{}/{}", m.kind, m.name), m.files.into_iter().map(|f| (f.path.clone(), f)).collect());

        }

    }

    out

}

/// Back up `owner` now, as its setting says. The run is recorded whatever

/// happens; a failure notifies the owner once until a run succeeds.

pub async fn run(app: &Shared, owner: &str, trigger: &str) -> Result<Run> {

    let s = setting(&app.db, owner).await?.with_context(|| format!("{owner} has no backup set up"))?;

    let started = db::now_ms();

    let at = run_name(started);

    let id: i64 =

        sqlx::query_scalar("INSERT INTO backup_runs (owner, started_ms, status, at, trigger) VALUES (?, ?, 'running', ?, ?) RETURNING id")

            .bind(owner)

            .bind(started)

            .bind(&at)

            .bind(trigger)

            .fetch_one(&app.db)

            .await?;

    let outcome = run_into(app, &s, &at).await;

    let (status, detail, sent) = match outcome {

        Ok((sent, problems)) if problems.is_empty() => ("ok", String::new(), sent),

        Ok((sent, problems)) => ("partial", problems.join("; "), sent),

        Err(e) => ("failed", format!("{e:#}"), (0, Sent::default())),

    };

    sqlx::query("UPDATE backup_runs SET finished_ms = ?, status = ?, detail = ?, repos = ?, files = ?, sent = ?, bytes = ? WHERE id = ?")

        .bind(db::now_ms())

        .bind(status)

        .bind(&detail)

        .bind(sent.0)

        .bind(sent.1.files)

        .bind(sent.1.sent)

        .bind(sent.1.bytes)

        .bind(id)

        .execute(&app.db)

        .await?;

    if status != "ok" {

        notify_failure(app, &s, status, &detail).await;

    }

    let r: Run = sqlx::query_as("SELECT * FROM backup_runs WHERE id = ?").bind(id).fetch_one(&app.db).await?;

    Ok(r)

}

async fn run_into(app: &Shared, s: &Setting, at: &str) -> Result<((i64, Sent), Vec<String>)> {

    let dest: std::sync::Arc<dyn Destination> = destination(app, s)?.into();

    let known = {

        let d = dest.clone();

        tokio::task::spawn_blocking(move || last_manifests(d.as_ref())).await?

    };

    let mine: Vec<db::repos::RepoRow> = db::repos::list(&app.db).await?.into_iter().filter(|r| r.owner == s.owner).collect();

    let mut problems = Vec::new();

    let mut sent = Sent::default();

    let mut repos = 0i64;

    for row in &mine {

        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };

        for (kind, store) in [("repos", &app.repos), ("wikis", &app.wikis)] {

            if !store.exists(&name) {

                continue;

            }

            let dir = store.dir(&name);

            let k = known.get(&format!("{kind}/{}", row.name)).cloned().unwrap_or_default();

            // The large, unchanging files without the lock, then the whole

            // walk again under the read lock: pushes wait for that moment.

            let (d1, dir1, k1) = (dest.clone(), dir.clone(), k.clone());

            let first = tokio::task::spawn_blocking(move || {

                let mut s = Sent::default();

                copy_dir(d1.as_ref(), &dir1, &k1, true, &mut s).map(|f| (f, s))

            })

            .await?;

            let mut k2 = k;

            if let Ok((files, _)) = &first {

                for f in files {

                    k2.insert(f.path.clone(), f.clone());

                }