OXYGENHUB
# d3b048b27277 oxygenhub / crates / hub / src / backups.rs
59.0 KiB · Rust · last changed Backups per repository: leave one out, or give it its own schedule 23 days ago source blame history raw
1 
//! Backups (docs/backups.md, oxygenhub#169): each user and organisation
2 
//! copies its repositories, on a schedule, to a destination it controls.
3 
//!
4 
//! A repository is its store directory (`.oxy/`): packs and loose objects
5 
//! that never change once written, and small files that do (refs, the
6 
//! operation log, policies, signature indexes). A run copies each file
7 
//! once, named by its BLAKE3 (`blobs/`), and writes a manifest of path to
8 
//! hash per repository (`runs/<at>/repos/`), so only what changed travels
9 
//! and any kept run can be restored byte for byte, signatures and all.
10 
//! The large files are copied first; the walk is then repeated under the
11 
//! repository's read lock, which pushes wait on, so each manifest is one
12 
//! consistent moment. Beside each manifest go the hub's records for the
13 
//! repository as JSON (never a secret value), the owner's records, and
14 
//! last a `DONE` marker: a run without one was cut short and is ignored.
15
16 
use crate::db::{self, Pool};
17 
use crate::repos::RepoName;
18 
use crate::Shared;
19 
use anyhow::{bail, Context, Result};
20 
use serde::{Deserialize, Serialize};
21 
use std::collections::HashMap;
22 
use std::path::{Path, PathBuf};
23
24 
/// How often, as the settings offer it.
25 
pub const EVERY: &[(&str, &str)] = &[("off", "Off"), ("hourly", "Every hour"), ("daily", "Every day")];
26 
/// How long runs are kept, in days.
27 
pub const KEEP: &[i64] = &[7, 30, 90];
28
29 
/// What an owner backs up, where and how often.
30 
#[derive(Clone, Debug, sqlx::FromRow, Serialize)]
31 
pub struct Setting {
32 
    pub owner: String,
33 
    pub kind: String,
34 
    pub every: String,
35 
    pub destination: String,
36 
    pub target: String,
37 
    pub prefix: String,
38 
    pub provider: String,
39 
    pub keep_days: i64,
40 
    pub updated_ms: i64,
41 
}
42
43 
impl Setting {
44 
    pub fn off(owner: &str, kind: &str) -> Setting {
45 
        Setting {
46 
            owner: owner.into(),
47 
            kind: kind.into(),
48 
            every: "off".into(),
49 
            destination: "dir".into(),
50 
            target: String::new(),
51 
            prefix: String::new(),
52 
            provider: String::new(),
53 
            keep_days: 30,
54 
            updated_ms: 0,
55 
        }
56 
    }
57
58 
    /// Seconds between runs; none when off.
59 
    pub fn period_s(&self) -> Option<i64> {
60 
        match self.every.as_str() {
61 
            "hourly" => Some(3600),
62 
            "daily" => Some(86_400),
63 
            _ => None,
64 
        }
65 
    }
66 
}
67
68 
/// One run, as the settings page and the API show it.
69 
#[derive(Clone, Debug, sqlx::FromRow, Serialize)]
70 
pub struct Run {
71 
    pub id: i64,
72 
    pub owner: String,
73 
    pub started_ms: i64,
74 
    pub finished_ms: Option<i64>,
75 
    pub status: String,
76 
    pub at: String,
77 
    pub repos: i64,
78 
    pub files: i64,
79 
    pub sent: i64,
80 
    pub bytes: i64,
81 
    pub detail: String,
82 
    pub trigger: String,
83 
}
84
85 
pub async fn setting(pool: &Pool, owner: &str) -> sqlx::Result<Option<Setting>> {
86 
    sqlx::query_as("SELECT * FROM backup_settings WHERE owner = ?").bind(owner).fetch_optional(pool).await
87 
}
88
89 
pub async fn save(pool: &Pool, s: &Setting) -> sqlx::Result<()> {
90 
    sqlx::query(
91 
        "INSERT INTO backup_settings (owner, kind, every, destination, target, prefix, provider, keep_days, updated_ms) \
92 
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(owner) DO UPDATE SET kind = excluded.kind, every = excluded.every, \
93 
         destination = excluded.destination, target = excluded.target, prefix = excluded.prefix, provider = excluded.provider, \
94 
         keep_days = excluded.keep_days, updated_ms = excluded.updated_ms",
95 
    )
96 
    .bind(&s.owner)
97 
    .bind(&s.kind)
98 
    .bind(&s.every)
99 
    .bind(&s.destination)
100 
    .bind(&s.target)
101 
    .bind(&s.prefix)
102 
    .bind(&s.provider)
103 
    .bind(s.keep_days)
104 
    .bind(db::now_ms())
105 
    .execute(pool)
106 
    .await?;
107 
    Ok(())
108 
}
109
110 
pub async fn runs(pool: &Pool, owner: &str, limit: i64) -> sqlx::Result<Vec<Run>> {
111 
    sqlx::query_as("SELECT * FROM backup_runs WHERE owner = ? ORDER BY id DESC LIMIT ?").bind(owner).bind(limit).fetch_all(pool).await
112 
}
113
114 
/// What a repository may choose about its backup (§5): with its owner's,
115 
/// left out, or at least hourly or daily whatever its owner's schedule.
116 
pub const REPO_CHOICES: &[(&str, &str)] =
117 
    &[("", "With its owner's backup"), ("hourly", "At least every hour"), ("daily", "At least every day"), ("off", "Left out")];
118
119 
/// Whether `row` is in its owner's run: never when it is left out, always
120 
/// when it has its own schedule, else as its owner's schedule says (a run
121 
/// asked for by hand takes every repository not left out).
122 
pub fn included(s: &Setting, row: &db::repos::RepoRow, by_hand: bool) -> bool {
123 
    match row.backup.as_str() {
124 
        "off" => false,
125 
        "hourly" | "daily" => true,
126 
        _ => by_hand || s.every != "off",
127 
    }
128 
}
129
130 
/// Seconds between an owner's runs: the shortest of its own schedule and
131 
/// its repositories' own; none when nothing is scheduled.
132 
pub fn period_with(s: &Setting, repos: &[db::repos::RepoRow]) -> Option<i64> {
133 
    let own = repos.iter().filter_map(|r| match r.backup.as_str() {
134 
        "hourly" => Some(3600),
135 
        "daily" => Some(86_400),
136 
        _ => None,
137 
    });
138 
    s.period_s().into_iter().chain(own).min()
139 
}
140
141 
/// One owner's backup as Admin › Backups shows it (§5): its setting, its
142 
/// newest run, and how many runs failed in the last day. An administrator
143 
/// sees that it ran, never what it wrote.
144 
#[derive(Clone, Debug, Serialize)]
145 
pub struct Overview {
146 
    pub setting: Setting,
147 
    pub last: Option<Run>,
148 
    pub failed_day: i64,
149 
    /// The repositories it covers now.
150 
    pub repos: i64,
151 
}
152
153 
impl Overview {
154 
    /// Where its runs go, as a person reads it.
155 
    pub fn place(&self) -> String {
156 
        let s = &self.setting;
157 
        match s.destination.as_str() {
158 
            "gcs" if s.prefix.is_empty() => format!("gs://{}", s.target),
159 
            "gcs" => format!("gs://{}/{}", s.target, s.prefix),
160 
            _ => format!("{}/{}", s.target, s.owner),
161 
        }
162 
    }
163 
}
164
165 
/// Every owner's backup (Admin › Backups), and the owners of repositories
166 
/// that none covers, with how many each has.
167 
pub async fn overview(pool: &Pool) -> sqlx::Result<(Vec<Overview>, Vec<(String, i64)>)> {
168 
    let settings: Vec<Setting> = sqlx::query_as("SELECT * FROM backup_settings ORDER BY owner").fetch_all(pool).await?;
169 
    let counts: Vec<(String, i64)> =
170 
        sqlx::query_as("SELECT owner, COUNT(*) FROM repositories GROUP BY owner ORDER BY owner").fetch_all(pool).await?;
171 
    let day_ago = db::now_ms() - 86_400_000;
172 
    let mut out = Vec::new();
173 
    for s in settings {
174 
        let last = runs(pool, &s.owner, 1).await?.into_iter().next();
175 
        let failed_day: i64 =
176 
            sqlx::query_scalar("SELECT COUNT(*) FROM backup_runs WHERE owner = ? AND status = 'failed' AND started_ms >= ?")
177 
                .bind(&s.owner)
178 
                .bind(day_ago)
179 
                .fetch_one(pool)
180 
                .await?;
181 
        let repos = db::repos::list(pool).await?.iter().filter(|r| r.owner == s.owner && included(&s, r, false)).count() as i64;
182 
        out.push(Overview { setting: s, last, failed_day, repos });
183 
    }
184 
    let uncovered = counts.into_iter().filter(|(owner, _)| !out.iter().any(|o| o.setting.owner == *owner && o.repos > 0)).collect();
185 
    Ok((out, uncovered))
186 
}
187
188 
// ---------------------------------------------------------------- where
189
190 
/// Where backups are written: keys are `/`-separated paths.
191 
pub trait Destination: Send + Sync {
192 
    fn exists(&self, key: &str) -> Result<bool>;
193 
    /// Write `bytes` at `key`, whole or not at all.
194 
    fn put(&self, key: &str, bytes: &[u8]) -> Result<()>;
195 
    fn get(&self, key: &str) -> Result<Vec<u8>>;
196 
    /// The names directly under `prefix` (files and "directories").
197 
    fn list(&self, prefix: &str) -> Result<Vec<String>>;
198 
    /// Remove `key` and everything under it.
199 
    fn remove(&self, key: &str) -> Result<()>;
200 
}
201
202 
/// A directory on the hub's host.
203 
pub struct Directory {
204 
    pub root: PathBuf,
205 
}
206
207 
impl Directory {
208 
    fn path(&self, key: &str) -> Result<PathBuf> {
209 
        if key.split('/').any(|p| p == ".." || p == ".") || key.starts_with('/') {
210 
            bail!("not a clean key: {key}");
211 
        }
212 
        Ok(self.root.join(key))
213 
    }
214 
}
215
216 
impl Destination for Directory {
217 
    fn exists(&self, key: &str) -> Result<bool> {
218 
        Ok(self.path(key)?.exists())
219 
    }
220 
    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {
221 
        let p = self.path(key)?;
222 
        std::fs::create_dir_all(p.parent().context("no parent")?)?;
223 
        let tmp = p.with_extension(format!("part-{}", std::process::id()));
224 
        std::fs::write(&tmp, bytes)?;
225 
        std::fs::rename(&tmp, &p)?;
226 
        Ok(())
227 
    }
228 
    fn get(&self, key: &str) -> Result<Vec<u8>> {
229 
        let p = self.path(key)?;
230 
        std::fs::read(&p).with_context(|| format!("read {}", p.display()))
231 
    }
232 
    fn list(&self, prefix: &str) -> Result<Vec<String>> {
233 
        let p = self.path(prefix)?;
234 
        let mut out: Vec<String> = match std::fs::read_dir(&p) {
235 
            Ok(d) => d.flatten().map(|e| e.file_name().to_string_lossy().into_owned()).collect(),
236 
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
237 
            Err(e) => return Err(e.into()),
238 
        };
239 
        out.sort();
240 
        Ok(out)
241 
    }
242 
    fn remove(&self, key: &str) -> Result<()> {
243 
        let p = self.path(key)?;
244 
        if p.is_dir() {
245 
            std::fs::remove_dir_all(p)?;
246 
        } else if p.exists() {
247 
            std::fs::remove_file(p)?;
248 
        }
249 
        Ok(())
250 
    }
251 
}
252
253 
/// A Cloud Storage bucket, reached keylessly (docs/backups.md §3, §6):
254 
/// the hub mints a short token as `backup:OWNER`, Google's STS exchanges
255 
/// it for an access token through the owner's workload identity
256 
/// provider, and that writes the bucket. No key exists anywhere; the
257 
/// access token lasts an hour and is fetched again before it runs out.
258 
pub struct Gcs {
259 
    bucket: String,
260 
    /// Keys go under this, when set.
261 
    prefix: String,
262 
    storage: String,
263 
    sts: String,
264 
    /// The provider's resource name, `//iam.googleapis.com/projects/…`.
265 
    provider: String,
266 
    subject: String,
267 
    app: Shared,
268 
    client: reqwest::Client,
269 
    rt: tokio::runtime::Handle,
270 
    token: std::sync::Mutex<Option<(String, i64)>>,
271 
}
272
273 
impl Gcs {
274 
    pub fn new(app: &Shared, s: &Setting) -> Gcs {
275 
        Gcs {
276 
            bucket: s.target.clone(),
277 
            prefix: s.prefix.trim_matches('/').to_string(),
278 
            storage: app.config.backups.storage_url.trim_end_matches('/').to_string(),
279 
            sts: app.config.backups.sts_url.trim_end_matches('/').to_string(),
280 
            provider: s.provider.clone(),
281 
            subject: format!("backup:{}", s.owner),
282 
            app: app.clone(),
283 
            client: reqwest::Client::builder().timeout(std::time::Duration::from_secs(300)).build().unwrap_or_default(),
284 
            rt: tokio::runtime::Handle::current(),
285 
            token: std::sync::Mutex::new(None),
286 
        }
287 
    }
288
289 
    fn name(&self, key: &str) -> String {
290 
        if self.prefix.is_empty() {
291 
            key.to_string()
292 
        } else {
293 
            format!("{}/{key}", self.prefix)
294 
        }
295 
    }
296
297 
    /// An access token, exchanged for a fresh one of the hub's when the
298 
    /// last is within five minutes of running out.
299 
    fn access(&self) -> Result<String> {
300 
        let now = db::now_ms();
301 
        if let Some((t, until)) = self.token.lock().unwrap().as_ref() {
302 
            if *until - now > 300_000 {
303 
                return Ok(t.clone());
304 
            }
305 
        }
306 
        let (token, expires_in) = self.rt.block_on(async {
307 
            let issuer = self.app.oidc().await?;
308 
            // Google's default audience for a provider: its name over https.
309 
            let audience = format!("https:{}", self.provider);
310 
            let jwt = issuer.mint_for(self.app.config.base_url(), &self.subject, &audience, db::now_ms(), 600);
311 
            let res = self
312 
                .client
313 
                .post(format!("{}/v1/token", self.sts))
314 
                .json(&serde_json::json!({
315 
                    "grantType": "urn:ietf:params:oauth:grant-type:token-exchange",
316 
                    "audience": self.provider,
317 
                    "scope": "https://www.googleapis.com/auth/devstorage.read_write",
318 
                    "requestedTokenType": "urn:ietf:params:oauth:token-type:access_token",
319 
                    "subjectToken": jwt,
320 
                    "subjectTokenType": "urn:ietf:params:oauth:token-type:jwt",
321 
                }))
322 
                .send()
323 
                .await?;
324 
            let status = res.status();
325 
            let body: serde_json::Value = res.json().await.unwrap_or_default();
326 
            if !status.is_success() {
327 
                bail!(
328 
                    "Google's token exchange refused {}: {} (check the provider's issuer is this hub's address and its attribute condition admits backup:)",
329 
                    self.subject,
330 
                    body["error_description"].as_str().or(body["error"].as_str()).unwrap_or("no reason given")
331 
                );
332 
            }
333 
            let token = body["access_token"].as_str().context("the token exchange gave no access token")?.to_string();
334 
            anyhow::Ok((token, body["expires_in"].as_i64().unwrap_or(3600)))
335 
        })?;
336 
        *self.token.lock().unwrap() = Some((token.clone(), now + expires_in * 1000));
337 
        Ok(token)
338 
    }
339
340 
    fn object_url(&self, key: &str) -> Result<reqwest::Url> {
341 
        let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
342 
        u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&self.name(key));
343 
        Ok(u)
344 
    }
345
346 
    fn refused(&self, what: &str, status: reqwest::StatusCode, body: &str) -> anyhow::Error {
347 
        let hint = if status == reqwest::StatusCode::FORBIDDEN || status == reqwest::StatusCode::UNAUTHORIZED {
348 
            format!(" (grant roles/storage.objectAdmin on gs://{} to the principal …/subject/{})", self.bucket, self.subject)
349 
        } else {
350 
            String::new()
351 
        };
352 
        anyhow::anyhow!("GCS {what}: {status}{hint}: {}", body.chars().take(200).collect::<String>())
353 
    }
354
355 
    /// Every object name under `full` (no delimiter), for removal.
356 
    fn all_under(&self, full: &str) -> Result<Vec<String>> {
357 
        let mut out = Vec::new();
358 
        let mut page: Option<String> = None;
359 
        loop {
360 
            let token = self.access()?;
361 
            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
362 
            u.query_pairs_mut().append_pair("prefix", full).append_pair("fields", "items(name),nextPageToken");
363 
            if let Some(p) = &page {
364 
                u.query_pairs_mut().append_pair("pageToken", p);
365 
            }
366 
            let body: serde_json::Value = self.rt.block_on(async {
367 
                let res = self.client.get(u).bearer_auth(&token).send().await?;
368 
                let status = res.status();
369 
                let text = res.text().await?;
370 
                if !status.is_success() {
371 
                    return Err(self.refused("list", status, &text));
372 
                }
373 
                anyhow::Ok(serde_json::from_str(&text)?)
374 
            })?;
375 
            out.extend(body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str().map(String::from)));
376 
            match body["nextPageToken"].as_str() {
377 
                Some(p) => page = Some(p.to_string()),
378 
                None => return Ok(out),
379 
            }
380 
        }
381 
    }
382 
}
383
384 
impl Destination for Gcs {
385 
    fn exists(&self, key: &str) -> Result<bool> {
386 
        let token = self.access()?;
387 
        let mut u = self.object_url(key)?;
388 
        u.query_pairs_mut().append_pair("fields", "name");
389 
        self.rt.block_on(async {
390 
            let res = self.client.get(u).bearer_auth(&token).send().await?;
391 
            match res.status() {
392 
                s if s.is_success() => Ok(true),
393 
                reqwest::StatusCode::NOT_FOUND => Ok(false),
394 
                s => Err(self.refused("read", s, &res.text().await.unwrap_or_default())),
395 
            }
396 
        })
397 
    }
398 
    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {
399 
        let token = self.access()?;
400 
        let mut u = reqwest::Url::parse(&format!("{}/upload/storage/v1/b/{}/o", self.storage, self.bucket))?;
401 
        u.query_pairs_mut().append_pair("uploadType", "media").append_pair("name", &self.name(key));
402 
        let body = bytes.to_vec();
403 
        self.rt.block_on(async {
404 
            let res = self.client.post(u).bearer_auth(&token).header("content-type", "application/octet-stream").body(body).send().await?;
405 
            let status = res.status();
406 
            if !status.is_success() {
407 
                return Err(self.refused("write", status, &res.text().await.unwrap_or_default()));
408 
            }
409 
            Ok(())
410 
        })
411 
    }
412 
    fn get(&self, key: &str) -> Result<Vec<u8>> {
413 
        let token = self.access()?;
414 
        let mut u = self.object_url(key)?;
415 
        u.query_pairs_mut().append_pair("alt", "media");
416 
        self.rt.block_on(async {
417 
            let res = self.client.get(u).bearer_auth(&token).send().await?;
418 
            let status = res.status();
419 
            if !status.is_success() {
420 
                return Err(self.refused(&format!("read {key}"), status, &res.text().await.unwrap_or_default()));
421 
            }
422 
            Ok(res.bytes().await?.to_vec())
423 
        })
424 
    }
425 
    fn list(&self, prefix: &str) -> Result<Vec<String>> {
426 
        let full = format!("{}/", self.name(prefix.trim_end_matches('/')));
427 
        let mut out = std::collections::BTreeSet::new();
428 
        let mut page: Option<String> = None;
429 
        loop {
430 
            let token = self.access()?;
431 
            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
432 
            u.query_pairs_mut()
433 
                .append_pair("prefix", &full)
434 
                .append_pair("delimiter", "/")
435 
                .append_pair("fields", "items(name),prefixes,nextPageToken");
436 
            if let Some(p) = &page {
437 
                u.query_pairs_mut().append_pair("pageToken", p);
438 
            }
439 
            let body: serde_json::Value = self.rt.block_on(async {
440 
                let res = self.client.get(u).bearer_auth(&token).send().await?;
441 
                let status = res.status();
442 
                let text = res.text().await?;
443 
                if !status.is_success() {
444 
                    return Err(self.refused("list", status, &text));
445 
                }
446 
                anyhow::Ok(serde_json::from_str(&text)?)
447 
            })?;
448 
            for n in body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str()) {
449 
                out.insert(n.trim_start_matches(&full).to_string());
450 
            }
451 
            for p in body["prefixes"].as_array().into_iter().flatten().filter_map(|p| p.as_str()) {
452 
                out.insert(p.trim_start_matches(&full).trim_end_matches('/').to_string());
453 
            }
454 
            match body["nextPageToken"].as_str() {
455 
                Some(p) => page = Some(p.to_string()),
456 
                None => return Ok(out.into_iter().filter(|n| !n.is_empty()).collect()),
457 
            }
458 
        }
459 
    }
460 
    fn remove(&self, key: &str) -> Result<()> {
461 
        let mut names = self.all_under(&format!("{}/", self.name(key)))?;
462 
        names.push(self.name(key));
463 
        for n in names {
464 
            let token = self.access()?;
465 
            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
466 
            u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&n);
467 
            self.rt.block_on(async {
468 
                let res = self.client.delete(u).bearer_auth(&token).send().await?;
469 
                match res.status() {
470 
                    s if s.is_success() || s == reqwest::StatusCode::NOT_FOUND => Ok(()),
471 
                    s => Err(self.refused("delete", s, &res.text().await.unwrap_or_default())),
472 
                }
473 
            })?;
474 
        }
475 
        Ok(())
476 
    }
477 
}
478
479 
/// Check a destination end to end: write a probe, read it back, remove it.
480 
pub fn probe(dest: &dyn Destination) -> Result<()> {
481 
    let marker = format!("probe {}", db::now_ms());
482 
    dest.put(".oxyhub-probe", marker.as_bytes())?;
483 
    if dest.get(".oxyhub-probe")? != marker.as_bytes() {
484 
        bail!("the probe read back differently");
485 
    }
486 
    dest.remove(".oxyhub-probe")
487 
}
488
489 
/// The owner's destination, from its setting and the site's rules.
490 
pub fn destination(app: &Shared, s: &Setting) -> Result<Box<dyn Destination>> {
491 
    match s.destination.as_str() {
492 
        "dir" => {
493 
            let allowed = &app.config.backups.directories;
494 
            let target = PathBuf::from(&s.target);
495 
            if !allowed.contains(&target) {
496 
                bail!("{} is not a backup directory this hub allows (hub.toml [backups] directories)", s.target);
497 
            }
498 
            let mut root = target.join(&s.owner);
499 
            if !s.prefix.is_empty() {
500 
                root = root.join(s.prefix.trim_matches('/'));
501 
            }
502 
            Ok(Box::new(Directory { root }))
503 
        }
504 
        "gcs" => {
505 
            if s.target.is_empty() || s.provider.is_empty() {
506 
                bail!("a GCS destination needs its bucket and its workload identity provider");
507 
            }
508 
            Ok(Box::new(Gcs::new(app, s)))
509 
        }
510 
        other => bail!("unknown destination {other:?}"),
511 
    }
512 
}
513
514 
// ---------------------------------------------------------------- what
515
516 
/// One file of a repository's store, in a manifest.
517 
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
518 
pub struct File {
519 
    pub path: String,
520 
    pub size: u64,
521 
    /// Its modification time in nanoseconds: unchanged size and time mean
522 
    /// unchanged content, so it is not hashed again.
523 
    pub mtime: i128,
524 
    pub hash: String,
525 
}
526
527 
/// A repository at one run.
528 
#[derive(Clone, Debug, Serialize, Deserialize)]
529 
pub struct Manifest {
530 
    pub owner: String,
531 
    pub name: String,
532 
    /// `repos` or `wikis`.
533 
    pub kind: String,
534 
    pub files: Vec<File>,
535 
}
536
537 
fn blob_key(hash: &str) -> String {
538 
    format!("blobs/{}/{hash}", &hash[..2])
539 
}
540
541 
/// The large files that never change once written come first.
542 
fn immutable(path: &str) -> bool {
543 
    path.starts_with(".oxy/objects/") && (path.contains("/pack-") || path.contains("/loose/"))
544 
}
545
546 
#[derive(Default)]
547 
struct Sent {
548 
    files: i64,
549 
    sent: i64,
550 
    bytes: i64,
551 
}
552
553 
/// Walk `dir`, sending each file the destination lacks; `known` says which
554 
/// paths are unchanged since the last run (and their hashes).
555 
fn copy_dir(dest: &dyn Destination, dir: &Path, known: &HashMap<String, File>, only_immutable: bool, sent: &mut Sent) -> Result<Vec<File>> {
556 
    let mut out = Vec::new();
557 
    let mut stack = vec![dir.to_path_buf()];
558 
    while let Some(d) = stack.pop() {
559 
        let entries = match std::fs::read_dir(&d) {
560 
            Ok(e) => e,
561 
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
562 
            Err(e) => return Err(e.into()),
563 
        };
564 
        for e in entries.flatten() {
565 
            let p = e.path();
566 
            let ft = e.file_type()?;
567 
            if ft.is_dir() {
568 
                stack.push(p);
569 
                continue;
570 
            }
571 
            if !ft.is_file() {
572 
                continue;
573 
            }
574 
            let rel = p.strip_prefix(dir).unwrap_or(&p).to_string_lossy().replace('\\', "/");
575 
            if only_immutable && !immutable(&rel) {
576 
                continue;
577 
            }
578 
            // A file packed away while the walk ran is simply gone.
579 
            let Ok(md) = std::fs::metadata(&p) else { continue };
580 
            let mtime =
581 
                md.modified().ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()).map(|d| d.as_nanos() as i128).unwrap_or(0);
582 
            let size = md.len();
583 
            sent.files += 1;
584 
            if let Some(k) = known.get(&rel).filter(|k| k.size == size && k.mtime == mtime) {
585 
                out.push(k.clone());
586 
                continue;
587 
            }
588 
            let Ok(bytes) = std::fs::read(&p) else { continue };
589 
            let hash = blake3::hash(&bytes).to_hex().to_string();
590 
            let key = blob_key(&hash);
591 
            if !dest.exists(&key)? {
592 
                dest.put(&key, &bytes)?;
593 
                sent.sent += 1;
594 
                sent.bytes += bytes.len() as i64;
595 
            }
596 
            out.push(File { path: rel, size, mtime, hash });
597 
        }
598 
    }
599 
    out.sort_by(|a, b| a.path.cmp(&b.path));
600 
    Ok(out)
601 
}
602
603 
// ---------------------------------------------------------------- records
604
605 
/// Tables and columns never written: secrets, credentials, sessions.
606 
fn private_table(t: &str) -> bool {
607 
    ["secret", "token", "session", "password", "totp", "recovery", "sqlx"].iter().any(|w| t.contains(w))
608 
}
609
610 
fn private_column(c: &str) -> bool {
611 
    ["secret", "token", "password", "salt", "totp", "sealed", "hash_key"].iter().any(|w| c.contains(w))
612 
}
613
614 
async fn tables(pool: &Pool) -> Result<Vec<String>> {
615 
    let mut t: Vec<String> =
616 
        sqlx::query_scalar("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'").fetch_all(pool).await?;
617 
    t.retain(|t| !private_table(t));
618 
    t.sort();
619 
    Ok(t)
620 
}
621
622 
async fn columns(pool: &Pool, table: &str) -> Result<Vec<String>> {
623 
    Ok(sqlx::query_scalar(&format!("SELECT name FROM pragma_table_info('{table}')")).fetch_all(pool).await?)
624 
}
625
626 
/// The rows of `table` matching `cond`, as a JSON array, private columns
627 
/// left out and blobs as hex.
628 
async fn rows(pool: &Pool, table: &str, cond: &str) -> Result<serde_json::Value> {
629 
    let cols = columns(pool, table).await?;
630 
    let fields: Vec<String> = cols
631 
        .iter()
632 
        .filter(|c| !private_column(c))
633 
        .map(|c| format!("'{c}', CASE WHEN typeof(\"{c}\") = 'blob' THEN hex(\"{c}\") ELSE \"{c}\" END"))
634 
        .collect();
635 
    if fields.is_empty() {
636 
        return Ok(serde_json::json!([]));
637 
    }
638 
    let sql = format!("SELECT json_group_array(json_object({})) FROM \"{table}\" WHERE {cond}", fields.join(", "));
639 
    let text: String = sqlx::query_scalar(&sql).fetch_one(pool).await?;
640 
    Ok(serde_json::from_str(&text)?)
641 
}
642
643 
/// Every row about something: the tables with column `key` equal to `id`,
644 
/// then, three levels deep, the tables whose foreign keys point at rows
645 
/// already taken (an issue's comments, a proposal's reviews and their
646 
/// threads).
647 
async fn records(pool: &Pool, key: &str, id: i64) -> Result<serde_json::Value> {
648 
    let all = tables(pool).await?;
649 
    let mut conds: Vec<(String, String)> = Vec::new();
650 
    for t in &all {
651 
        if columns(pool, t).await?.iter().any(|c| c == key) {
652 
            conds.push((t.clone(), format!("\"{key}\" = {id}")));
653 
        }
654 
    }
655 
    for _ in 0..3 {
656 
        let mut more = Vec::new();
657 
        for t in &all {
658 
            if conds.iter().any(|(c, _)| c == t) || more.iter().any(|(c, _): &(String, String)| c == t) {
659 
                continue;
660 
            }
661 
            let fks: Vec<(String, String, String)> =
662 
                sqlx::query_as(&format!("SELECT \"table\", \"from\", \"to\" FROM pragma_foreign_key_list('{t}')")).fetch_all(pool).await?;
663 
            if let Some((parent, from, to, pcond)) = fks
664 
                .iter()
665 
                .find_map(|(p, f, to)| conds.iter().find(|(c, _)| c == p).map(|(_, pc)| (p.clone(), f.clone(), to.clone(), pc.clone())))
666 
            {
667 
                more.push((t.clone(), format!("\"{from}\" IN (SELECT \"{to}\" FROM \"{parent}\" WHERE {pcond})")));
668 
            }
669 
        }
670 
        if more.is_empty() {
671 
            break;
672 
        }
673 
        conds.extend(more);
674 
    }
675 
    let mut out = serde_json::Map::new();
676 
    for (t, cond) in conds {
677 
        let r = rows(pool, &t, &cond).await?;
678 
        if r.as_array().is_some_and(|a| !a.is_empty()) {
679 
            out.insert(t, r);
680 
        }
681 
    }
682 
    Ok(serde_json::Value::Object(out))
683 
}
684
685 
/// The key of a run's records naming the accounts, teams and organisations
686 
/// their rows point at, by id: `{"accounts": {"7": "andy"}, …}`.
687 
const NAMES: &str = "_names";
688
689 
/// Tables a restore leaves out: the search index (rebuilt from what is
690 
/// restored), and what is passing or personal rather than the
691 
/// repository's: inboxes, webhook deliveries, runners' declines.
692 
const NOT_RESTORED: [&str; 5] = ["search", "notifications", "deliveries", "flow_declines", NAMES];
693
694 
async fn fk_list(pool: &Pool, table: &str) -> Result<Vec<(String, String, String)>> {
695 
    Ok(sqlx::query_as(&format!("SELECT \"table\", \"from\", \"to\" FROM pragma_foreign_key_list('{table}')")).fetch_all(pool).await?)
696 
}
697
698 
/// The names behind the account, team and organisation ids that `rec`'s
699 
/// rows point at, so a restore on another hub finds them again by name.
700 
async fn names_of(pool: &Pool, rec: &serde_json::Value) -> Result<serde_json::Value> {
701 
    let mut ids: HashMap<String, std::collections::BTreeSet<i64>> = HashMap::new();
702 
    for (table, rows) in rec.as_object().into_iter().flatten() {
703 
        let Some(rows) = rows.as_array() else { continue };
704 
        for (parent, from, _) in fk_list(pool, table).await? {
705 
            if ["accounts", "teams", "organisations"].contains(&parent.as_str()) {
706 
                ids.entry(parent).or_default().extend(rows.iter().filter_map(|r| r[&from].as_i64()));
707 
            }
708 
        }
709 
    }
710 
    let mut out = serde_json::Map::new();
711 
    for (kind, sql) in [
712 
        ("accounts", "SELECT id, username FROM accounts WHERE id = ?"),
713 
        ("organisations", "SELECT id, name FROM organisations WHERE id = ?"),
714 
        ("teams", "SELECT t.id, o.name || '/' || t.name FROM teams t JOIN organisations o ON o.id = t.org_id WHERE t.id = ?"),
715 
    ] {
716 
        let mut m = serde_json::Map::new();
717 
        for id in ids.get(kind).into_iter().flatten() {
718 
            if let Some((id, name)) = sqlx::query_as::<_, (i64, String)>(sql).bind(id).fetch_optional(pool).await? {
719 
                m.insert(id.to_string(), name.into());
720 
            }
721 
        }
722 
        out.insert(kind.into(), m.into());
723 
    }
724 
    Ok(out.into())
725 
}
726
727 
/// The id `name` has on this hub as one of `_names`' kinds.
728 
async fn id_by_name(pool: &Pool, kind: &str, name: &str) -> Result<Option<i64>> {
729 
    Ok(match kind {
730 
        "accounts" => sqlx::query_scalar("SELECT id FROM accounts WHERE username = ?").bind(name).fetch_optional(pool).await?,
731 
        "organisations" => sqlx::query_scalar("SELECT id FROM organisations WHERE name = ?").bind(name).fetch_optional(pool).await?,
732 
        "teams" => match name.split_once('/') {
733 
            Some((org, team)) => {
734 
                sqlx::query_scalar("SELECT t.id FROM teams t JOIN organisations o ON o.id = t.org_id WHERE o.name = ? AND t.name = ?")
735 
                    .bind(org)
736 
                    .bind(team)
737 
                    .fetch_optional(pool)
738 
                    .await?
739 
            }
740 
            None => None,
741 
        },
742 
        _ => None,
743 
    })
744 
}
745
746 
/// One column of a table on this hub.
747 
struct Col {
748 
    name: String,
749 
    ty: String,
750 
    notnull: bool,
751 
    default: Option<String>,
752 
    pk: i64,
753 
}
754
755 
async fn table_cols(pool: &Pool, table: &str) -> Result<Vec<Col>> {
756 
    let rows: Vec<(String, String, i64, Option<String>, i64)> =
757 
        sqlx::query_as(&format!("SELECT name, type, \"notnull\", dflt_value, pk FROM pragma_table_info('{table}')"))
758 
            .fetch_all(pool)
759 
            .await?;
760 
    Ok(rows.into_iter().map(|(name, ty, notnull, default, pk)| Col { name, ty, notnull: notnull != 0, default, pk }).collect())
761 
}
762
763 
/// Bind one value of a backed-up row: blobs were written as hex.
764 
fn bind_json<'q>(
765 
    q: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>,
766 
    v: serde_json::Value,
767 
    blob: bool,
768 
) -> sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>> {
769 
    match v {
770 
        serde_json::Value::Null => q.bind(None::<i64>),
771 
        serde_json::Value::Bool(b) => q.bind(b as i64),
772 
        serde_json::Value::Number(n) => match n.as_i64() {
773 
            Some(i) => q.bind(i),
774 
            None => q.bind(n.as_f64().unwrap_or(0.0)),
775 
        },
776 
        serde_json::Value::String(s) if blob => q.bind(hex::decode(&s).unwrap_or_else(|_| s.into_bytes())),
777 
        serde_json::Value::String(s) => q.bind(s),
778 
        other => q.bind(other.to_string()),
779 
    }
780 
}
781
782 
/// Restore one repository's records from a run into this hub, the
783 
/// repository being `repo` here and `old` in the backup (docs/backups.md
784 
/// §7): its issues, proposals, labels, milestones, releases and their
785 
/// assets, checks, Flow runs and the rest, each table after the tables it
786 
/// points at, every id mapped to its new one and every account, team and
787 
/// organisation found again by name. A row pointing at an account this
788 
/// hub lacks keeps a blank reference where it may, and is left out where
789 
/// it may not. What needs re-entering (webhook secrets, never backed up)
790 
/// and what was left out come back as notes.
791 
async fn restore_records(
792 
    app: &Shared,
793 
    dest: &std::sync::Arc<dyn Destination>,
794 
    rec: &serde_json::Value,
795 
    old: i64,
796 
    repo: i64,
797 
) -> Result<Vec<String>> {
798 
    let pool = &app.db;
799 
    let obj = rec.as_object().context("the records are not an object")?;
800 
    let mut by_name: HashMap<(String, i64), Option<i64>> = HashMap::new();
801 
    for (kind, m) in obj.get(NAMES).and_then(|n| n.as_object()).into_iter().flatten() {
802 
        for (id, name) in m.as_object().into_iter().flatten() {
803 
            let (Ok(id), Some(name)) = (id.parse::<i64>(), name.as_str()) else { continue };
804 
            by_name.insert((kind.clone(), id), id_by_name(pool, kind, name).await?);
805 
        }
806 
    }
807 
    let here: std::collections::HashSet<String> = tables(pool).await?.into_iter().collect();
808 
    let mut notes = Vec::new();
809 
    let mut todo: Vec<String> = Vec::new();
810 
    for t in obj.keys().filter(|t| t.as_str() != "repositories" && !NOT_RESTORED.contains(&t.as_str())) {
811 
        if here.contains(t) {
812 
            todo.push(t.clone());
813 
        } else {
814 
            notes.push(format!("{t}: not a table on this hub, left out"));
815 
        }
816 
    }
817 
    let mut fks: HashMap<String, Vec<(String, String, String)>> = HashMap::new();
818 
    for t in &todo {
819 
        fks.insert(t.clone(), fk_list(pool, t).await?);
820 
    }
821 
    // Each table after the tables of the records it points at.
822 
    let mut order = Vec::new();
823 
    while !todo.is_empty() {
824 
        let mut ready: Vec<String> =
825 
            todo.iter().filter(|t| fks[*t].iter().all(|(p, _, _)| p == *t || !todo.contains(p))).cloned().collect();
826 
        if ready.is_empty() {
827 
            ready.push(todo[0].clone());
828 
        }
829 
        todo.retain(|t| !ready.contains(t));
830 
        order.extend(ready);
831 
    }
832 
    let mut maps: HashMap<String, HashMap<i64, i64>> = HashMap::new();
833 
    maps.insert("repositories".into(), [(old, repo)].into());
834 
    let mut left_out: std::collections::BTreeMap<String, usize> = Default::default();
835 
    let mut tx = pool.begin().await?;
836 
    for table in &order {
837 
        let cols = table_cols(pool, table).await?;
838 
        let keys: Vec<&str> = cols.iter().filter(|c| c.pk > 0).map(|c| c.name.as_str()).collect();
839 
        let by_id = keys == ["id"] && cols.iter().any(|c| c.name == "id" && c.ty.eq_ignore_ascii_case("INTEGER"));
840 
        let verb = if keys == ["repo_id"] {
841 
            "INSERT OR REPLACE"
842 
        } else if by_id {
843 
            "INSERT"
844 
        } else {
845 
            "INSERT OR IGNORE"
846 
        };
847 
        let mut rows: Vec<&serde_json::Map<String, serde_json::Value>> =
848 
            obj[table].as_array().into_iter().flatten().filter_map(|r| r.as_object()).collect();
849 
        rows.sort_by_key(|r| r.get("id").and_then(|v| v.as_i64()).unwrap_or(0));
850 
        'row: for row in rows {
851 
            let (mut names, mut values) = (Vec::new(), Vec::new());
852 
            for c in &cols {
853 
                if by_id && c.name == "id" {
854 
                    continue;
855 
                }
856 
                let mut v = row.get(&c.name).cloned().filter(|v| !v.is_null());
857 
                if let Some((parent, _, _)) = fks[table].iter().find(|(_, from, _)| *from == c.name) {
858 
                    if let Some(was) = v.as_ref().and_then(|v| v.as_i64()) {
859 
                        let now = match parent.as_str() {
860 
                            "accounts" | "teams" | "organisations" => by_name.get(&(parent.clone(), was)).copied().flatten(),
861 
                            p => maps.get(p).and_then(|m| m.get(&was)).copied(),
862 
                        };
863 
                        if now.is_none() && c.notnull {
864 
                            *left_out.entry(table.clone()).or_default() += 1;
865 
                            continue 'row;
866 
                        }
867 
                        v = now.map(serde_json::Value::from);
868 
                    }
869 
                }
870 
                let blob = c.ty.to_ascii_uppercase().contains("BLOB");
871 
                match v {
872 
                    Some(v) => {
873 
                        names.push(c.name.clone());
874 
                        values.push((v, blob));
875 
                    }
876 
                    // Never backed up (a secret): blank here, to enter again.
877 
                    None if c.notnull && c.default.is_none() => {
878 
                        names.push(c.name.clone());
879 
                        let blank = if c.ty.to_ascii_uppercase().contains("INT") { serde_json::json!(0) } else { serde_json::json!("") };
880 
                        values.push((blank, false));
881 
                    }
882 
                    None => {}
883 
                }
884 
            }
885 
            let sql = format!(
886 
                "{verb} INTO \"{table}\" ({}) VALUES ({})",
887 
                names.iter().map(|n| format!("\"{n}\"")).collect::<Vec<_>>().join(", "),
888 
                vec!["?"; names.len()].join(", ")
889 
            );
890 
            let mut q = sqlx::query(&sql);
891 
            for (v, blob) in values {
892 
                q = bind_json(q, v, blob);
893 
            }
894 
            let done = q.execute(&mut *tx).await.with_context(|| format!("restore a row of {table}"))?;
895 
            if let (true, Some(was)) = (by_id, row.get("id").and_then(|v| v.as_i64())) {
896 
                maps.entry(table.clone()).or_default().insert(was, done.last_insert_rowid());
897 
            }
898 
        }
899 
    }
900 
    tx.commit().await?;
901 
    for (table, n) in left_out {
902 
        notes.push(format!("{table}: {n} row(s) left out: they name accounts or rows this hub lacks"));
903 
    }
904 
    // Search finds what came back.
905 
    for (kind, table) in [("issue", "issues"), ("proposal", "proposals")] {
906 
        let found: Vec<(i64, i64, String, String)> =
907 
            sqlx::query_as(&format!("SELECT id, number, title, body FROM {table} WHERE repo_id = ?")).bind(repo).fetch_all(pool).await?;
908 
        for (id, number, title, body) in found {
909 
            db::notify::index(pool, kind, repo, id, number, &title, &body).await?;
910 
        }
911 
    }
912 
    // Webhook secrets were never written: they are entered again.
913 
    let hooks: Vec<String> =
914 
        sqlx::query_scalar("SELECT url FROM hooks WHERE repo_id = ? AND secret = ''").bind(repo).fetch_all(pool).await?;
915 
    for url in hooks {
916 
        notes.push(format!("webhook {url}: enter its secret again (Settings › Webhooks)"));
917 
    }
918 
    // Release assets, each checked against its name.
919 
    for hash in obj.get("release_assets").and_then(|a| a.as_array()).into_iter().flatten().filter_map(|a| a["hash"].as_str()) {
920 
        if hash.len() != 64 || !hash.bytes().all(|b| b.is_ascii_hexdigit()) {
921 
            continue;
922 
        }
923 
        let path = crate::web::releases::blob_path(app, hash);
924 
        if path.exists() {
925 
            continue;
926 
        }
927 
        let got = tokio::task::block_in_place(|| dest.get(&blob_key(hash)));
928 
        match got {
929 
            Ok(bytes) if blake3::hash(&bytes).to_hex().as_str() == hash => {
930 
                std::fs::create_dir_all(path.parent().context("no parent")?)?;
931 
                std::fs::write(&path, &bytes)?;
932 
            }
933 
            Ok(_) => notes.push(format!("release asset {}: the copy does not match its hash", &hash[..12])),
934 
            Err(_) => notes.push(format!("release asset {}: not in the backup", &hash[..12])),
935 
        }
936 
    }
937 
    Ok(notes)
938 
}
939
940 
// ---------------------------------------------------------------- a run
941
942 
fn run_name(ms: i64) -> String {
943 
    let t = time::OffsetDateTime::from_unix_timestamp_nanos(ms as i128 * 1_000_000).unwrap_or(time::OffsetDateTime::UNIX_EPOCH);
944 
    format!("{:04}{:02}{:02}T{:02}{:02}{:02}Z", t.year(), u8::from(t.month()), t.day(), t.hour(), t.minute(), t.second())
945 
}
946
947 
/// The last whole run's manifests, by `kind/name`.
948 
fn last_manifests(dest: &dyn Destination) -> HashMap<String, HashMap<String, File>> {
949 
    let mut out = HashMap::new();
950 
    let Ok(at) = dest.get("LATEST").map(|b| String::from_utf8_lossy(&b).trim().to_string()) else { return out };
951 
    for name in dest.list(&format!("runs/{at}/repos")).unwrap_or_default() {
952 
        if let Ok(m) = dest.get(&format!("runs/{at}/repos/{name}")).and_then(|b| Ok(serde_json::from_slice::<Manifest>(&b)?)) {
953 
            out.insert(format!("{}/{}", m.kind, m.name), m.files.into_iter().map(|f| (f.path.clone(), f)).collect());
954 
        }
955 
    }
956 
    out
957 
}
958
959 
/// Back up `owner` now, as its setting says. The run is recorded whatever
960 
/// happens; a failure notifies the owner once until a run succeeds.
961 
pub async fn run(app: &Shared, owner: &str, trigger: &str) -> Result<Run> {
962 
    let s = setting(&app.db, owner).await?.with_context(|| format!("{owner} has no backup set up"))?;
963 
    let started = db::now_ms();
964 
    let at = run_name(started);
965 
    let id: i64 =
966 
        sqlx::query_scalar("INSERT INTO backup_runs (owner, started_ms, status, at, trigger) VALUES (?, ?, 'running', ?, ?) RETURNING id")
967 
            .bind(owner)
968 
            .bind(started)
969 
            .bind(&at)
970 
            .bind(trigger)
971 
            .fetch_one(&app.db)
972 
            .await?;
973 
    let outcome = run_into(app, &s, &at, trigger == "now").await;
974 
    let (status, detail, sent) = match outcome {
975 
        Ok((sent, problems)) if problems.is_empty() => ("ok", String::new(), sent),
976 
        Ok((sent, problems)) => ("partial", problems.join("; "), sent),
977 
        Err(e) => ("failed", format!("{e:#}"), (0, Sent::default())),
978 
    };
979 
    sqlx::query("UPDATE backup_runs SET finished_ms = ?, status = ?, detail = ?, repos = ?, files = ?, sent = ?, bytes = ? WHERE id = ?")
980 
        .bind(db::now_ms())
981 
        .bind(status)
982 
        .bind(&detail)
983 
        .bind(sent.0)
984 
        .bind(sent.1.files)
985 
        .bind(sent.1.sent)
986 
        .bind(sent.1.bytes)
987 
        .bind(id)
988 
        .execute(&app.db)
989 
        .await?;
990 
    if status != "ok" {
991 
        notify_failure(app, &s, status, &detail).await;
992 
    }
993 
    let r: Run = sqlx::query_as("SELECT * FROM backup_runs WHERE id = ?").bind(id).fetch_one(&app.db).await?;
994 
    Ok(r)
995 
}
996
997 
async fn run_into(app: &Shared, s: &Setting, at: &str, by_hand: bool) -> Result<((i64, Sent), Vec<String>)> {
998 
    let dest: std::sync::Arc<dyn Destination> = destination(app, s)?.into();
999 
    let known = {
1000 
        let d = dest.clone();
1001 
        tokio::task::spawn_blocking(move || last_manifests(d.as_ref())).await?
1002 
    };
1003 
    // Its repositories, but those left out (§5).
1004 
    let mine: Vec<db::repos::RepoRow> =
1005 
        db::repos::list(&app.db).await?.into_iter().filter(|r| r.owner == s.owner && included(s, r, by_hand)).collect();
1006 
    let mut problems = Vec::new();
1007 
    let mut sent = Sent::default();
1008 
    let mut repos = 0i64;
1009 
    for row in &mine {
1010 
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
1011 
        for (kind, store) in [("repos", &app.repos), ("wikis", &app.wikis)] {
1012 
            if !store.exists(&name) {
1013 
                continue;
1014 
            }
1015 
            let dir = store.dir(&name);
1016 
            let k = known.get(&format!("{kind}/{}", row.name)).cloned().unwrap_or_default();
1017 
            // The large, unchanging files without the lock, then the whole
1018 
            // walk again under the read lock: pushes wait for that moment.
1019 
            let (d1, dir1, k1) = (dest.clone(), dir.clone(), k.clone());
1020 
            let first = tokio::task::spawn_blocking(move || {
1021 
                let mut s = Sent::default();
1022 
                copy_dir(d1.as_ref(), &dir1, &k1, true, &mut s).map(|f| (f, s))
1023 
            })
1024 
            .await?;
1025 
            let mut k2 = k;
1026 
            if let Ok((files, _)) = &first {
1027 
                for f in files {
1028 
                    k2.insert(f.path.clone(), f.clone());
1029 
                }
1030 
            }
1031 
            let guard = store.read_guard(&name).await;
1032 
            let (d2, dir2) = (dest.clone(), dir.clone());
1033 
            let second = tokio::task::spawn_blocking(move || {
1034 
                let _guard = guard;
1035 
                let mut s = Sent::default();
1036 
                copy_dir(d2.as_ref(), &dir2, &k2, false, &mut s).map(|f| (f, s))
1037 
            })
1038 
            .await?;
1039 
            match second {
1040 
                Ok((files, s2)) => {
1041 
                    let s1 = first.map(|(_, s)| s).unwrap_or_default();
1042 
                    sent.files += s2.files;
1043 
                    sent.sent += s1.sent + s2.sent;
1044 
                    sent.bytes += s1.bytes + s2.bytes;
1045 
                    let m = Manifest { owner: row.owner.clone(), name: row.name.clone(), kind: kind.into(), files };
1046 
                    let key = format!("runs/{at}/repos/{kind}-{}.json", row.name);
1047 
                    tokio::task::block_in_place(|| dest.put(&key, &serde_json::to_vec(&m)?))?;
1048 
                    if kind == "repos" {
1049 
                        repos += 1;
1050 
                    }
1051 
                }
1052 
                Err(e) => problems.push(format!("{}/{} ({kind}): {e:#}", row.owner, row.name)),
1053 
            }
1054 
        }
1055 
        let rec = async {
1056 
            let mut r = records(&app.db, "repo_id", row.id).await?;
1057 
            // The repository's own row, which restore recreates first, and
1058 
            // the names behind the accounts, teams and organisations its
1059 
            // rows point at, which another hub finds them by.
1060 
            let own = rows(&app.db, "repositories", &format!("id = {}", row.id)).await?;
1061 
            let names = names_of(&app.db, &r).await?;
1062 
            if let Some(m) = r.as_object_mut() {
1063 
                m.insert("repositories".into(), own);
1064 
                m.insert(NAMES.into(), names);
1065 
            }
1066 
            anyhow::Ok(r)
1067 
        };
1068 
        match rec.await {
1069 
            Ok(r) => {
1070 
                tokio::task::block_in_place(|| dest.put(&format!("runs/{at}/records/{}.json", row.name), &serde_json::to_vec(&r)?))?;
1071 
                // Release assets live beside the stores, named by their
1072 
                // BLAKE3: each sent once, as the stores' files are.
1073 
                for hash in r["release_assets"].as_array().into_iter().flatten().filter_map(|a| a["hash"].as_str()) {
1074 
                    if hash.len() != 64 || !hash.bytes().all(|b| b.is_ascii_hexdigit()) {
1075 
                        continue;
1076 
                    }
1077 
                    let path = crate::web::releases::blob_path(app, hash);
1078 
                    let key = blob_key(hash);
1079 
                    let put = tokio::task::block_in_place(|| -> Result<Option<i64>> {
1080 
                        if dest.exists(&key)? {
1081 
                            return Ok(None);
1082 
                        }
1083 
                        let bytes = std::fs::read(&path).with_context(|| format!("read release asset {}", &hash[..12]))?;
1084 
                        dest.put(&key, &bytes)?;
1085 
                        Ok(Some(bytes.len() as i64))
1086 
                    });
1087 
                    match put {
1088 
                        Ok(Some(n)) => {
1089 
                            sent.files += 1;
1090 
                            sent.sent += 1;
1091 
                            sent.bytes += n;
1092 
                        }
1093 
                        Ok(None) => sent.files += 1,
1094 
                        Err(e) => problems.push(format!("{}/{} release asset: {e:#}", row.owner, row.name)),
1095 
                    }
1096 
                }
1097 
            }
1098 
            Err(e) => problems.push(format!("{}/{} records: {e:#}", row.owner, row.name)),
1099 
        }
1100 
    }
1101 
    // The owner itself.
1102 
    let owner_rows = if s.kind == "org" {
1103 
        match db::accounts::org_by_name(&app.db, &s.owner).await? {
1104 
            Some(o) => records(&app.db, "org_id", o.id).await?,
1105 
            None => serde_json::json!({}),
1106 
        }
1107 
    } else {
1108 
        match db::accounts::account_by_name(&app.db, &s.owner).await? {
1109 
            Some(a) => records(&app.db, "account_id", a.id).await?,
1110 
            None => serde_json::json!({}),
1111 
        }
1112 
    };
1113 
    // The blocking calls of a destination (GCS drives its client on this
1114 
    // runtime) run where blocking is allowed.
1115 
    tokio::task::block_in_place(|| -> Result<()> {
1116 
        dest.put(&format!("runs/{at}/owner.json"), &serde_json::to_vec(&owner_rows)?)?;
1117 
        dest.put(&format!("runs/{at}/DONE"), at.as_bytes())?;
1118 
        dest.put("LATEST", at.as_bytes())?;
1119 
        prune(dest.as_ref(), s.keep_days)
1120 
    })?;
1121 
    Ok(((repos, sent), problems))
1122 
}
1123
1124 
/// Drop whole runs older than `keep_days`, and the blobs no kept run
1125 
/// names. The newest whole run is always kept.
1126 
pub fn prune(dest: &dyn Destination, keep_days: i64) -> Result<()> {
1127 
    let cutoff = run_name(db::now_ms() - keep_days.max(1) * 86_400_000);
1128 
    let runs = dest.list("runs")?;
1129 
    let whole: Vec<&String> = runs.iter().filter(|r| dest.exists(&format!("runs/{r}/DONE")).unwrap_or(false)).collect();
1130 
    let newest = whole.last().map(|s| s.to_string());
1131 
    let mut removed = false;
1132 
    for r in &runs {
1133 
        if r.as_str() < cutoff.as_str() && Some(r.to_string()) != newest {
1134 
            dest.remove(&format!("runs/{r}"))?;
1135 
            removed = true;
1136 
        }
1137 
    }
1138 
    if !removed {
1139 
        return Ok(());
1140 
    }
1141 
    let mut wanted = std::collections::HashSet::new();
1142 
    for r in dest.list("runs")? {
1143 
        for m in dest.list(&format!("runs/{r}/repos"))? {
1144 
            if let Ok(m) = dest.get(&format!("runs/{r}/repos/{m}")).and_then(|b| Ok(serde_json::from_slice::<Manifest>(&b)?)) {
1145 
                wanted.extend(m.files.into_iter().map(|f| f.hash));
1146 
            }
1147 
        }
1148 
    }
1149 
    for shard in dest.list("blobs")? {
1150 
        for b in dest.list(&format!("blobs/{shard}"))? {
1151 
            if !wanted.contains(&b) {
1152 
                dest.remove(&format!("blobs/{shard}/{b}"))?;
1153 
            }
1154 
        }
1155 
    }
1156 
    Ok(())
1157 
}
1158
1159 
async fn notify_failure(app: &Shared, s: &Setting, status: &str, detail: &str) {
1160 
    // Once per spell of failures: not when the run before failed too.
1161 
    let before: Option<String> =
1162 
        sqlx::query_scalar("SELECT status FROM backup_runs WHERE owner = ? AND finished_ms IS NOT NULL ORDER BY id DESC LIMIT 1 OFFSET 1")
1163 
            .bind(&s.owner)
1164 
            .fetch_optional(&app.db)
1165 
            .await
1166 
            .ok()
1167 
            .flatten();
1168 
    if before.as_deref().is_some_and(|b| b != "ok") {
1169 
        return;
1170 
    }
1171 
    let (who, url): (Vec<i64>, String) = if s.kind == "org" {
1172 
        match db::accounts::org_by_name(&app.db, &s.owner).await.ok().flatten() {
1173 
            Some(o) => (
1174 
                db::accounts::org_members(&app.db, o.id)
1175 
                    .await
1176 
                    .unwrap_or_default()
1177 
                    .into_iter()
1178 
                    .filter(|(_, r)| r == "owner")
1179 
                    .map(|(a, _)| a.id)
1180 
                    .collect(),
1181 
                format!("/orgs/{}/settings#backups", s.owner),
1182 
            ),
1183 
            None => return,
1184 
        }
1185 
    } else {
1186 
        match db::accounts::account_by_name(&app.db, &s.owner).await.ok().flatten() {
1187 
            Some(a) => (vec![a.id], "/settings/backups".into()),
1188 
            None => return,
1189 
        }
1190 
    };
1191 
    let title = format!("Backup of {} {status}: {}", s.owner, detail.chars().take(160).collect::<String>());
1192 
    for id in who {
1193 
        let _ = db::notify::add(
1194 
            &app.db,
1195 
            db::notify::New {
1196 
                account_id: id,
1197 
                repo_id: None,
1198 
                reason: db::notify::Reason::Participating,
1199 
                kind: "backup",
1200 
                title: &title,
1201 
                url: &url,
1202 
                actor: "",
1203 
            },
1204 
        )
1205 
        .await;
1206 
    }
1207 
}
1208
1209 
/// Run each owner's backup when it is due, one at a time. A run left
1210 
/// "running" by a restart is marked failed first.
1211 
pub async fn scheduler(app: Shared) {
1212 
    let _ = sqlx::query(
1213 
        "UPDATE backup_runs SET status = 'failed', detail = 'the hub restarted during the run', finished_ms = ? WHERE status = 'running'",
1214 
    )
1215 
    .bind(db::now_ms())
1216 
    .execute(&app.db)
1217 
    .await;
1218 
    loop {
1219 
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
1220 
        for s in due(&app.db).await.unwrap_or_default() {
1221 
            if let Err(e) = run(&app, &s.owner, "schedule").await {
1222 
                tracing::warn!("backup of {}: {e:#}", s.owner);
1223 
            }
1224 
        }
1225 
    }
1226 
}
1227
1228 
/// The owners whose next run is due now: by their own schedule, or by
1229 
/// the shortest schedule one of their repositories chose (§5). A run takes
1230 
/// every included repository and sends only what changed, so each run
1231 
/// stays whole and a restore point has everything.
1232 
pub async fn due(pool: &Pool) -> sqlx::Result<Vec<Setting>> {
1233 
    let all: Vec<Setting> = sqlx::query_as("SELECT * FROM backup_settings WHERE updated_ms > 0").fetch_all(pool).await?;
1234 
    let repos = db::repos::list(pool).await?;
1235 
    let now = db::now_ms();
1236 
    let mut out = Vec::new();
1237 
    for s in all {
1238 
        let theirs: Vec<db::repos::RepoRow> = repos.iter().filter(|r| r.owner == s.owner).cloned().collect();
1239 
        let Some(period) = period_with(&s, &theirs) else { continue };
1240 
        let last: Option<i64> =
1241 
            sqlx::query_scalar("SELECT MAX(started_ms) FROM backup_runs WHERE owner = ?").bind(&s.owner).fetch_one(pool).await?;
1242 
        if last.is_none_or(|l| now - l >= period * 1000) {
1243 
            out.push(s);
1244 
        }
1245 
    }
1246 
    Ok(out)
1247 
}
1248
1249 
/// Restore every repository of the whole run `at` (the latest when none)
1250 
/// from `dest` into `repos_root/OWNER/NAME` and `wikis_root/…`, refusing to
1251 
/// overwrite one that exists. Each file is checked against its hash.
1252 
pub fn restore(dest: &dyn Destination, at: Option<&str>, repos_root: &Path, wikis_root: &Path) -> Result<Vec<String>> {
1253 
    let at = match at {
1254 
        Some(a) => a.to_string(),
1255 
        None => String::from_utf8_lossy(&dest.get("LATEST").context("no whole run at this destination")?).trim().to_string(),
1256 
    };
1257 
    if !dest.exists(&format!("runs/{at}/DONE"))? {
1258 
        bail!("run {at} is not whole (no DONE): pick another with --run");
1259 
    }
1260 
    let mut done = Vec::new();
1261 
    for name in dest.list(&format!("runs/{at}/repos"))? {
1262 
        let m: Manifest = serde_json::from_slice(&dest.get(&format!("runs/{at}/repos/{name}"))?)?;
1263 
        let root = if m.kind == "wikis" { wikis_root } else { repos_root };
1264 
        let dir = root.join(&m.owner).join(&m.name);
1265 
        if dir.exists() {
1266 
            bail!("{} exists: restore into an empty hub, or move it aside", dir.display());
1267 
        }
1268 
        for f in &m.files {
1269 
            let bytes = dest.get(&blob_key(&f.hash))?;
1270 
            if blake3::hash(&bytes).to_hex().as_str() != f.hash {
1271 
                bail!("{}/{} {}: the copy does not match its hash", m.owner, m.name, f.path);
1272 
            }
1273 
            let p = dir.join(&f.path);
1274 
            std::fs::create_dir_all(p.parent().context("no parent")?)?;
1275 
            std::fs::write(&p, &bytes)?;
1276 
        }
1277 
        done.push(format!("{}/{} ({})", m.owner, m.name, m.kind));
1278 
    }
1279 
    Ok(done)
1280 
}
1281
1282 
/// Restore the whole run `at` (the latest when none) into this hub: each
1283 
/// repository's and wiki's store byte for byte, then its row and settings,
1284 
/// then its records (issues, proposals, releases and the rest), so it shows
1285 
/// on the hub as it was (docs/backups.md §7). What was restored, then any
1286 
/// notes (`note: …`) on what needs entering again or was left out.
1287 
pub async fn restore_into_hub(app: &Shared, dest: std::sync::Arc<dyn Destination>, at: Option<String>) -> Result<Vec<String>> {
1288 
    let (repos_root, wikis_root) = (app.repos.root().to_path_buf(), app.wikis.root().to_path_buf());
1289 
    let (d, a) = (dest.clone(), at.clone());
1290 
    let done = tokio::task::spawn_blocking(move || restore(d.as_ref(), a.as_deref(), &repos_root, &wikis_root)).await??;
1291 
    let at = match at {
1292 
        Some(a) => a,
1293 
        None => String::from_utf8_lossy(&tokio::task::block_in_place(|| dest.get("LATEST"))?).trim().to_string(),
1294 
    };
1295 
    let mut notes = Vec::new();
1296 
    for name in tokio::task::block_in_place(|| dest.list(&format!("runs/{at}/records")))? {
1297 
        let rec: serde_json::Value =
1298 
            serde_json::from_slice(&tokio::task::block_in_place(|| dest.get(&format!("runs/{at}/records/{name}")))?)?;
1299 
        let Some(row) = rec["repositories"].as_array().and_then(|a| a.first()) else { continue };
1300 
        let (owner, repo) = (row["owner"].as_str().unwrap_or_default(), row["name"].as_str().unwrap_or_default());
1301 
        if owner.is_empty() || db::repos::get(&app.db, owner, repo).await?.is_some() {
1302 
            continue;
1303 
        }
1304 
        let visibility = match row["visibility"].as_str() {
1305 
            Some("private") => hub_api::Visibility::Private,
1306 
            Some("internal") => hub_api::Visibility::Internal,
1307 
            _ => hub_api::Visibility::Public,
1308 
        };
1309 
        let made = db::repos::create(&app.db, owner, repo, row["description"].as_str().unwrap_or_default(), visibility).await?;
1310 
        // Its settings as they were: every column the row kept but its
1311 
        // name, its id and a fork's parent (another hub's id).
1312 
        let cols = table_cols(&app.db, "repositories").await?;
1313 
        let set: Vec<(&Col, serde_json::Value)> = cols
1314 
            .iter()
1315 
            .filter(|c| !["id", "owner", "name", "forked_from"].contains(&c.name.as_str()))
1316 
            .filter_map(|c| row.get(&c.name).filter(|v| !v.is_null()).map(|v| (c, v.clone())))
1317 
            .collect();
1318 
        if !set.is_empty() {
1319 
            let sql = format!(
1320 
                "UPDATE repositories SET {} WHERE id = ?",
1321 
                set.iter().map(|(c, _)| format!("\"{}\" = ?", c.name)).collect::<Vec<_>>().join(", ")
1322 
            );
1323 
            let mut q = sqlx::query(&sql);
1324 
            for (c, v) in set {
1325 
                q = bind_json(q, v, c.ty.to_ascii_uppercase().contains("BLOB"));
1326 
            }
1327 
            q.bind(made.id).execute(&app.db).await?;
1328 
        }
1329 
        let old = row["id"].as_i64().unwrap_or(-1);
1330 
        for n in restore_records(app, &dest, &rec, old, made.id).await.with_context(|| format!("restore {owner}/{repo}'s records"))? {
1331 
            notes.push(format!("note: {owner}/{repo}: {n}"));
1332 
        }
1333 
    }
1334 
    let mut done = done;
1335 
    done.extend(notes);
1336 
    Ok(done)
1337 
}
1338
1339 
#[cfg(test)]
1340 
mod tests {
1341 
    use super::*;
1342
1343 
    #[test]
1344 
    fn a_directory_destination_keeps_keys_inside_its_root() {
1345 
        let d = tempfile::tempdir().unwrap();
1346 
        let dest = Directory { root: d.path().to_path_buf() };
1347 
        dest.put("runs/a/DONE", b"a").unwrap();
1348 
        assert_eq!(dest.get("runs/a/DONE").unwrap(), b"a");
1349 
        assert_eq!(dest.list("runs").unwrap(), vec!["a"]);
1350 
        assert!(dest.put("../escape", b"x").is_err() && dest.put("/abs", b"x").is_err());
1351 
        dest.remove("runs").unwrap();
1352 
        assert!(dest.list("runs").unwrap().is_empty());
1353 
    }
1354
1355 
    #[test]
1356 
    fn files_are_sent_once_and_unchanged_ones_not_read_again() {
1357 
        let src = tempfile::tempdir().unwrap();
1358 
        let out = tempfile::tempdir().unwrap();
1359 
        let dest = Directory { root: out.path().to_path_buf() };
1360 
        std::fs::create_dir_all(src.path().join(".oxy/objects/loose")).unwrap();
1361 
        std::fs::write(src.path().join(".oxy/objects/loose/aa"), b"object").unwrap();
1362 
        std::fs::write(src.path().join(".oxy/ops-head"), b"ref one").unwrap();
1363 
        let mut s = Sent::default();
1364 
        let files = copy_dir(&dest, src.path(), &HashMap::new(), false, &mut s).unwrap();
1365 
        assert_eq!((files.len(), s.sent), (2, 2));
1366 
        let known: HashMap<String, File> = files.iter().map(|f| (f.path.clone(), f.clone())).collect();
1367 
        let mut s = Sent::default();
1368 
        let again = copy_dir(&dest, src.path(), &known, false, &mut s).unwrap();
1369 
        assert_eq!((again, s.sent), (files.clone(), 0), "nothing new travels");
1370 
        // The first pass takes only the files that never change.
1371 
        let mut s = Sent::default();
1372 
        let first = copy_dir(&dest, src.path(), &HashMap::new(), true, &mut s).unwrap();
1373 
        assert_eq!(first.iter().map(|f| f.path.as_str()).collect::<Vec<_>>(), vec![".oxy/objects/loose/aa"]);
1374 
    }
1375
1376 
    #[test]
1377 
    fn run_names_sort_by_time() {
1378 
        assert_eq!(run_name(0), "19700101T000000Z");
1379 
        assert!(run_name(1_791_900_000_000) < run_name(1_791_903_600_000));
1380 
    }
1381 
}