Backups: a restore brings back the records, and release assets travel 5ef6cf72cd19

/// The key of a run's records naming the accounts, teams and organisations

/// their rows point at, by id: `{"accounts": {"7": "andy"}, …}`.

const NAMES: &str = "_names";

/// Tables a restore leaves out: the search index (rebuilt from what is

/// restored), and what is passing or personal rather than the

/// repository's: inboxes, webhook deliveries, runners' declines.

const NOT_RESTORED: [&str; 5] = ["search", "notifications", "deliveries", "flow_declines", NAMES];

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

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

}

/// The names behind the account, team and organisation ids that `rec`'s

/// rows point at, so a restore on another hub finds them again by name.

async fn names_of(pool: &Pool, rec: &serde_json::Value) -> Result<serde_json::Value> {

    let mut ids: HashMap<String, std::collections::BTreeSet<i64>> = HashMap::new();

    for (table, rows) in rec.as_object().into_iter().flatten() {

        let Some(rows) = rows.as_array() else { continue };

        for (parent, from, _) in fk_list(pool, table).await? {

            if ["accounts", "teams", "organisations"].contains(&parent.as_str()) {

                ids.entry(parent).or_default().extend(rows.iter().filter_map(|r| r[&from].as_i64()));

            }

        }

    }

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

    for (kind, sql) in [

        ("accounts", "SELECT id, username FROM accounts WHERE id = ?"),

        ("organisations", "SELECT id, name FROM organisations WHERE id = ?"),

        ("teams", "SELECT t.id, o.name || '/' || t.name FROM teams t JOIN organisations o ON o.id = t.org_id WHERE t.id = ?"),

    ] {

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

        for id in ids.get(kind).into_iter().flatten() {

            if let Some((id, name)) = sqlx::query_as::<_, (i64, String)>(sql).bind(id).fetch_optional(pool).await? {

                m.insert(id.to_string(), name.into());

            }

        }

        out.insert(kind.into(), m.into());

    }

    Ok(out.into())

}

/// The id `name` has on this hub as one of `_names`' kinds.

async fn id_by_name(pool: &Pool, kind: &str, name: &str) -> Result<Option<i64>> {

    Ok(match kind {

        "accounts" => sqlx::query_scalar("SELECT id FROM accounts WHERE username = ?").bind(name).fetch_optional(pool).await?,

        "organisations" => sqlx::query_scalar("SELECT id FROM organisations WHERE name = ?").bind(name).fetch_optional(pool).await?,

        "teams" => match name.split_once('/') {

            Some((org, team)) => {

                sqlx::query_scalar("SELECT t.id FROM teams t JOIN organisations o ON o.id = t.org_id WHERE o.name = ? AND t.name = ?")

                    .bind(org)

                    .bind(team)

                    .fetch_optional(pool)

                    .await?

            }

            None => None,

        },

        _ => None,

    })

}

/// One column of a table on this hub.

struct Col {

    name: String,

    ty: String,

    notnull: bool,

    default: Option<String>,

    pk: i64,

}

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

    let rows: Vec<(String, String, i64, Option<String>, i64)> =

        sqlx::query_as(&format!("SELECT name, type, \"notnull\", dflt_value, pk FROM pragma_table_info('{table}')"))

            .fetch_all(pool)

            .await?;

    Ok(rows.into_iter().map(|(name, ty, notnull, default, pk)| Col { name, ty, notnull: notnull != 0, default, pk }).collect())

}

/// Bind one value of a backed-up row: blobs were written as hex.

fn bind_json<'q>(

    q: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>,

    v: serde_json::Value,

    blob: bool,

) -> sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>> {

    match v {

        serde_json::Value::Null => q.bind(None::<i64>),

        serde_json::Value::Bool(b) => q.bind(b as i64),

        serde_json::Value::Number(n) => match n.as_i64() {

            Some(i) => q.bind(i),

            None => q.bind(n.as_f64().unwrap_or(0.0)),

        },

        serde_json::Value::String(s) if blob => q.bind(hex::decode(&s).unwrap_or_else(|_| s.into_bytes())),

        serde_json::Value::String(s) => q.bind(s),

        other => q.bind(other.to_string()),

    }

}

/// Restore one repository's records from a run into this hub, the

/// repository being `repo` here and `old` in the backup (docs/backups.md

/// §7): its issues, proposals, labels, milestones, releases and their

/// assets, checks, Flow runs and the rest, each table after the tables it

/// points at, every id mapped to its new one and every account, team and

/// organisation found again by name. A row pointing at an account this

/// hub lacks keeps a blank reference where it may, and is left out where

/// it may not. What needs re-entering (webhook secrets, never backed up)

/// and what was left out come back as notes.

async fn restore_records(

    app: &Shared,

    dest: &std::sync::Arc<dyn Destination>,

    rec: &serde_json::Value,

    old: i64,

    repo: i64,

) -> Result<Vec<String>> {

    let pool = &app.db;

    let obj = rec.as_object().context("the records are not an object")?;

    let mut by_name: HashMap<(String, i64), Option<i64>> = HashMap::new();

    for (kind, m) in obj.get(NAMES).and_then(|n| n.as_object()).into_iter().flatten() {

        for (id, name) in m.as_object().into_iter().flatten() {

            let (Ok(id), Some(name)) = (id.parse::<i64>(), name.as_str()) else { continue };

            by_name.insert((kind.clone(), id), id_by_name(pool, kind, name).await?);

        }

    }

    let here: std::collections::HashSet<String> = tables(pool).await?.into_iter().collect();

    let mut notes = Vec::new();

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

    for t in obj.keys().filter(|t| t.as_str() != "repositories" && !NOT_RESTORED.contains(&t.as_str())) {

        if here.contains(t) {

            todo.push(t.clone());

        } else {

            notes.push(format!("{t}: not a table on this hub, left out"));

        }

    }

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

    for t in &todo {

        fks.insert(t.clone(), fk_list(pool, t).await?);

    }

    // Each table after the tables of the records it points at.

    let mut order = Vec::new();

    while !todo.is_empty() {

        let mut ready: Vec<String> =

            todo.iter().filter(|t| fks[*t].iter().all(|(p, _, _)| p == *t || !todo.contains(p))).cloned().collect();

        if ready.is_empty() {

            ready.push(todo[0].clone());

        }

        todo.retain(|t| !ready.contains(t));

        order.extend(ready);

    }

    let mut maps: HashMap<String, HashMap<i64, i64>> = HashMap::new();

    maps.insert("repositories".into(), [(old, repo)].into());

    let mut left_out: std::collections::BTreeMap<String, usize> = Default::default();

    let mut tx = pool.begin().await?;

    for table in &order {

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

        let keys: Vec<&str> = cols.iter().filter(|c| c.pk > 0).map(|c| c.name.as_str()).collect();

        let by_id = keys == ["id"] && cols.iter().any(|c| c.name == "id" && c.ty.eq_ignore_ascii_case("INTEGER"));

        let verb = if keys == ["repo_id"] {

            "INSERT OR REPLACE"

        } else if by_id {

            "INSERT"

        } else {

            "INSERT OR IGNORE"

        };

        let mut rows: Vec<&serde_json::Map<String, serde_json::Value>> =

            obj[table].as_array().into_iter().flatten().filter_map(|r| r.as_object()).collect();

        rows.sort_by_key(|r| r.get("id").and_then(|v| v.as_i64()).unwrap_or(0));

        'row: for row in rows {

            let (mut names, mut values) = (Vec::new(), Vec::new());

            for c in &cols {

                if by_id && c.name == "id" {

                    continue;

                }

                let mut v = row.get(&c.name).cloned().filter(|v| !v.is_null());

                if let Some((parent, _, _)) = fks[table].iter().find(|(_, from, _)| *from == c.name) {

                    if let Some(was) = v.as_ref().and_then(|v| v.as_i64()) {

                        let now = match parent.as_str() {

                            "accounts" | "teams" | "organisations" => by_name.get(&(parent.clone(), was)).copied().flatten(),

                            p => maps.get(p).and_then(|m| m.get(&was)).copied(),

                        };

                        if now.is_none() && c.notnull {

                            *left_out.entry(table.clone()).or_default() += 1;

                            continue 'row;

                        }

                        v = now.map(serde_json::Value::from);

                    }

                }

                let blob = c.ty.to_ascii_uppercase().contains("BLOB");

                match v {

                    Some(v) => {

                        names.push(c.name.clone());

                        values.push((v, blob));

                    }

                    // Never backed up (a secret): blank here, to enter again.

                    None if c.notnull && c.default.is_none() => {

                        names.push(c.name.clone());

                        let blank = if c.ty.to_ascii_uppercase().contains("INT") { serde_json::json!(0) } else { serde_json::json!("") };

                        values.push((blank, false));

                    }

                    None => {}

                }

            }

            let sql = format!(

                "{verb} INTO \"{table}\" ({}) VALUES ({})",

                names.iter().map(|n| format!("\"{n}\"")).collect::<Vec<_>>().join(", "),

                vec!["?"; names.len()].join(", ")

            );

            let mut q = sqlx::query(&sql);

            for (v, blob) in values {

                q = bind_json(q, v, blob);

            }

            let done = q.execute(&mut *tx).await.with_context(|| format!("restore a row of {table}"))?;

            if let (true, Some(was)) = (by_id, row.get("id").and_then(|v| v.as_i64())) {

                maps.entry(table.clone()).or_default().insert(was, done.last_insert_rowid());

            }

        }

    }

    tx.commit().await?;

    for (table, n) in left_out {

        notes.push(format!("{table}: {n} row(s) left out: they name accounts or rows this hub lacks"));

    }

    // Search finds what came back.

    for (kind, table) in [("issue", "issues"), ("proposal", "proposals")] {

        let found: Vec<(i64, i64, String, String)> =

            sqlx::query_as(&format!("SELECT id, number, title, body FROM {table} WHERE repo_id = ?")).bind(repo).fetch_all(pool).await?;

        for (id, number, title, body) in found {

            db::notify::index(pool, kind, repo, id, number, &title, &body).await?;

        }

    }

    // Webhook secrets were never written: they are entered again.

    let hooks: Vec<String> =

        sqlx::query_scalar("SELECT url FROM hooks WHERE repo_id = ? AND secret = ''").bind(repo).fetch_all(pool).await?;

    for url in hooks {

        notes.push(format!("webhook {url}: enter its secret again (Settings › Webhooks)"));

    }

    // Release assets, each checked against its name.

    for hash in obj.get("release_assets").and_then(|a| a.as_array()).into_iter().flatten().filter_map(|a| a["hash"].as_str()) {

        if hash.len() != 64 || !hash.bytes().all(|b| b.is_ascii_hexdigit()) {

            continue;

        }

        let path = crate::web::releases::blob_path(app, hash);

        if path.exists() {

            continue;

        }

        let got = tokio::task::block_in_place(|| dest.get(&blob_key(hash)));

        match got {

            Ok(bytes) if blake3::hash(&bytes).to_hex().as_str() == hash => {

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

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

            }

            Ok(_) => notes.push(format!("release asset {}: the copy does not match its hash", &hash[..12])),

            Err(_) => notes.push(format!("release asset {}: not in the backup", &hash[..12])),

        }

    }

    Ok(notes)

}

            // The repository's own row, which restore recreates first, and

            // the names behind the accounts, teams and organisations its

            // rows point at, which another hub finds them by.

            let names = names_of(&app.db, &r).await?;

                m.insert(NAMES.into(), names);

            Ok(r) => {

                tokio::task::block_in_place(|| dest.put(&format!("runs/{at}/records/{}.json", row.name), &serde_json::to_vec(&r)?))?;

                // Release assets live beside the stores, named by their

                // BLAKE3: each sent once, as the stores' files are.

                for hash in r["release_assets"].as_array().into_iter().flatten().filter_map(|a| a["hash"].as_str()) {

                    if hash.len() != 64 || !hash.bytes().all(|b| b.is_ascii_hexdigit()) {

                        continue;

                    }

                    let path = crate::web::releases::blob_path(app, hash);

                    let key = blob_key(hash);

                    let put = tokio::task::block_in_place(|| -> Result<Option<i64>> {

                        if dest.exists(&key)? {

                            return Ok(None);

                        }

                        let bytes = std::fs::read(&path).with_context(|| format!("read release asset {}", &hash[..12]))?;

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

                        Ok(Some(bytes.len() as i64))

                    });

                    match put {

                        Ok(Some(n)) => {

                            sent.files += 1;

                            sent.sent += 1;

                            sent.bytes += n;

                        }

                        Ok(None) => sent.files += 1,

                        Err(e) => problems.push(format!("{}/{} release asset: {e:#}", row.owner, row.name)),

                    }

                }

            }

/// repository's and wiki's store byte for byte, then its row and settings,

/// then its records (issues, proposals, releases and the rest), so it shows

/// on the hub as it was (docs/backups.md §7). What was restored, then any

/// notes (`note: …`) on what needs entering again or was left out.

    let mut notes = Vec::new();

        // Its settings as they were: every column the row kept but its

        // name, its id and a fork's parent (another hub's id).

        let cols = table_cols(&app.db, "repositories").await?;

        let set: Vec<(&Col, serde_json::Value)> = cols

            .iter()

            .filter(|c| !["id", "owner", "name", "forked_from"].contains(&c.name.as_str()))

            .filter_map(|c| row.get(&c.name).filter(|v| !v.is_null()).map(|v| (c, v.clone())))

            .collect();

        if !set.is_empty() {

            let sql = format!(

                "UPDATE repositories SET {} WHERE id = ?",

                set.iter().map(|(c, _)| format!("\"{}\" = ?", c.name)).collect::<Vec<_>>().join(", ")

            );

            let mut q = sqlx::query(&sql);

            for (c, v) in set {

                q = bind_json(q, v, c.ty.to_ascii_uppercase().contains("BLOB"));

            }

            q.bind(made.id).execute(&app.db).await?;

        }

        let old = row["id"].as_i64().unwrap_or(-1);

        for n in restore_records(app, &dest, &rec, old, made.id).await.with_context(|| format!("restore {owner}/{repo}'s records"))? {

            notes.push(format!("note: {owner}/{repo}: {n}"));

    let mut done = done;

    done.extend(notes);

/// docs/backups.md §7 (BP1): a restore brings back the records as well as

/// the stores: issues with their comments, labels, milestone and

/// assignees, proposals with their reviews and events, releases with their

/// asset files, and webhooks without their secrets, every id mapped and

/// every account found again by name; what this hub lacks is said.

#[tokio::test(flavor = "multi_thread")]

async fn a_restore_brings_back_issues_proposals_and_releases() {

    let store = tempfile::tempdir().unwrap();

    let h = hub(store.path()).await;

    let andy = db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", false).await.unwrap();

    let bob = db::accounts::create_account(&h.app.db, "bob", "[email protected]", "x", "salt", false).await.unwrap();

    create(&h.app, "andy", "demo", hub_api::Visibility::Public).await;

    let work = tempfile::tempdir().unwrap();

    push(&format!("{}/andy/demo", h.base), work.path(), &[("README.md", "hello\n")]);

    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();

    db::repos::set_diffusion(&h.app.db, row.id, "green").await.unwrap();

    // Ids on the first hub start further on than on the fresh one.

    for n in 0..3 {

        db::issues::upsert_label(&h.app.db, row.id, &format!("spare{n}"), "#000000", "").await.unwrap();

    }

    let bug = db::issues::upsert_label(&h.app.db, row.id, "bug", "#d73a4a", "Something is wrong").await.unwrap();

    let m = db::issues::upsert_milestone(&h.app.db, row.id, "1.0", "the first", None).await.unwrap();

    let issue = db::issues::create(

        &h.app.db,

        db::issues::NewIssue { repo_id: row.id, title: "Sprites flicker", body: "on level two", author_id: Some(andy.id), author: "andy" },

    )

    .await

    .unwrap();

    db::issues::add_comment(&h.app.db, issue.id, Some(bob.id), "bob", "seen it too").await.unwrap();

    db::issues::set_issue_label(&h.app.db, issue.id, bug.id, true).await.unwrap();

    db::issues::set_milestone(&h.app.db, issue.id, Some(m.id)).await.unwrap();

    db::issues::set_assignee(&h.app.db, issue.id, andy.id, true).await.unwrap();

    db::issues::set_assignee(&h.app.db, issue.id, bob.id, true).await.unwrap();

    let p = db::proposals::create(

        &h.app.db,

        db::proposals::NewProposal {

            repo_id: row.id,

            title: "Double-buffer the sprites",

            body: "fixes the flicker",

            author_id: Some(bob.id),

            author: "bob",

            draft: false,

            target: "main",

            source_kind: "bookmark",

            source_repo_id: Some(row.id),

            source_bookmark: "fix",

            source_changes: "",

        },

    )

    .await

    .unwrap();

    db::proposals::add_event(&h.app.db, p.id, "bob", "opened", "").await.unwrap();

    db::proposals::add_review(&h.app.db, p.id, andy.id, "andy", "approve", "nice").await.unwrap();

    let release = db::releases::create(&h.app.db, row.id, "v1.0", "One", "notes", false, Some((andy.id, "andy"))).await.unwrap();

    let asset = b"a game image".to_vec();

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

    let path = h.app.config.data.join("assets").join(&hash[..2]).join(&hash);

    std::fs::create_dir_all(path.parent().unwrap()).unwrap();

    std::fs::write(&path, &asset).unwrap();

    db::releases::add_asset(&h.app.db, release.id, "game.prg", asset.len() as i64, &hash, "application/octet-stream").await.unwrap();

    db::hooks::create_hook(&h.app.db, Some(row.id), None, "https://example.com/hook", "s3cret-value", "push").await.unwrap();

    let mut s = Setting::off("andy", "user");

    s.every = "hourly".into();

    s.target = store.path().display().to_string();

    backups::save(&h.app.db, &s).await.unwrap();

    let run = backups::run(&h.app, "andy", "now").await.unwrap();

    assert_eq!(run.status, "ok", "{}", run.detail);

    let dest = Directory { root: store.path().join("andy") };

    let text = String::from_utf8_lossy(&dest.get(&format!("runs/{}/records/demo.json", run.at)).unwrap()).into_owned();

    assert!(!text.contains("s3cret-value"), "never a secret");

    assert!(text.contains("\"_names\"") && text.contains("\"bob\""), "accounts by name");

    // A fresh hub that knows andy but not bob.

    let empty = tempfile::tempdir().unwrap();

    let fresh = hub(empty.path()).await;

    let andy2 = db::accounts::create_account(&fresh.app.db, "andy", "[email protected]", "x", "salt", false).await.unwrap();

    let dest: Arc<dyn Destination> = Arc::new(Directory { root: store.path().join("andy") });

    let done = backups::restore_into_hub(&fresh.app, dest, None).await.unwrap();

    assert_eq!(done[0], "andy/demo (repos)");

    let notes = done[1..].join("\n");

    assert!(notes.contains("webhook https://example.com/hook: enter its secret again"), "{notes}");

    assert!(notes.contains("issue_assignees: 1 row(s) left out"), "bob is not on this hub: {notes}");

    let r2 = db::repos::get(&fresh.app.db, "andy", "demo").await.unwrap().unwrap();

    assert_eq!(r2.diffusion, "green", "settings come back");

    // The issue, its comment, label, milestone and the assignee this hub has.

    let issues: Vec<(i64, i64, String, Option<i64>)> = sqlx::query_as("SELECT id, number, title, author_id FROM issues WHERE repo_id = ?")

        .bind(r2.id)

        .fetch_all(&fresh.app.db)

        .await

        .unwrap();

    assert_eq!(issues.len(), 1);

    let (iid, number, title, author) = issues[0].clone();

    assert_eq!((number, title.as_str(), author), (issue.number, "Sprites flicker", Some(andy2.id)));

    let comments = db::issues::comments(&fresh.app.db, iid).await.unwrap();

    assert_eq!(comments.len(), 1);

    assert_eq!((comments[0].author.as_str(), comments[0].body.as_str()), ("bob", "seen it too"));

    let labels: Vec<String> =

        sqlx::query_scalar("SELECT l.name FROM issue_labels il JOIN labels l ON l.id = il.label_id WHERE il.issue_id = ?")

            .bind(iid)

            .fetch_all(&fresh.app.db)

            .await

            .unwrap();

    assert_eq!(labels, vec!["bug"], "the label, its id mapped");

    let milestone: Option<String> =

        sqlx::query_scalar("SELECT m.title FROM issues i JOIN milestones m ON m.id = i.milestone_id WHERE i.id = ?")

            .bind(iid)

            .fetch_one(&fresh.app.db)

            .await

            .unwrap();

    assert_eq!(milestone.as_deref(), Some("1.0"));

    assert_eq!(db::issues::assignees(&fresh.app.db, iid).await.unwrap(), vec!["andy"]);

    // The proposal, its review and its timeline.

    let p2 = db::proposals::get(&fresh.app.db, r2.id, p.number).await.unwrap().expect("the proposal");

    assert_eq!((p2.title.as_str(), p2.author.as_str(), p2.author_id), ("Double-buffer the sprites", "bob", None));

    let reviews: Vec<(String, String)> = sqlx::query_as("SELECT author, verdict FROM proposal_reviews WHERE proposal_id = ?")

        .bind(p2.id)

        .fetch_all(&fresh.app.db)

        .await

        .unwrap();

    assert_eq!(reviews, vec![("andy".to_string(), "approve".to_string())]);

    assert!(db::proposals::events(&fresh.app.db, p2.id).await.unwrap().iter().any(|e| e.event == "opened"));

    // The release and its asset's bytes.

    let rel = db::releases::get(&fresh.app.db, r2.id, "v1.0").await.unwrap().expect("the release");

    let assets = db::releases::assets(&fresh.app.db, rel.id).await.unwrap();

    assert_eq!(assets.len(), 1);

    assert_eq!(std::fs::read(fresh.app.config.data.join("assets").join(&hash[..2]).join(&hash)).unwrap(), asset);

    // Search finds what came back.

    let hits = db::notify::search(&fresh.app.db, "flicker", Some(r2.id), 10).await.unwrap();

    assert!(hits.iter().any(|h| h.kind == "issue" && h.target_id == iid), "{:?}", hits.iter().map(|h| &h.title).collect::<Vec<_>>());

    // New issues number on from where the old hub was.

    let next = db::issues::create(

        &fresh.app.db,

        db::issues::NewIssue { repo_id: r2.id, title: "After", body: "", author_id: Some(andy2.id), author: "andy" },

    )

    .await

    .unwrap();

    assert!(next.number > p.number, "{} after {}", next.number, p.number);

}

notices, and restore (`oxygenhub admin backup run|restore`). Since

2026-10-14 (oxygenhub#175) a restore brings back the records too (§7) and

release assets travel with the stores. Next: Admin › Backups,

per-repository settings, S3.

**As built (oxygenhub#175).** `oxygenhub admin backup restore --from DIR

[--run AT]` into a hub that lacks the repositories: each store and wiki

byte for byte (every file checked against its hash), then the

repository's row with its settings as they were (visibility, default

bookmark, Flow and Diffusion settings and the rest), then its records.

Each run's records carry `_names`, the username, organisation or

`org/team` behind every account, team and organisation id their rows

point at, so the restore finds them again by name on any hub. Tables go

in parents first; every row gets a new id and every reference follows

it. A reference to an account this hub lacks is left blank where the

column allows it (the author's name stays in the row) and the row left

out where it does not (an assignment, a review request); the restore

says how many. Webhooks come back without their secrets, each named to be

entered again. Release assets, which live beside the stores rather than

in them, are sent once each as blobs and written back checked against

their hash. Search is rebuilt for what came back. Left out on purpose:

inboxes (notifications), webhook deliveries, runners' declines.

The design as first written: