OXYGENHUB
# 0cd118748131 oxygenhub / crates / hub / src / backups.rs
75.2 KiB · Rust · last changed Backups: send the length with every upload, an empty file's too 22 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
    /// An S3 bucket's region (migration 0063).
42
    #[sqlx(default)]
43
    pub region: String,
44
}
45
46
impl Setting {
47
    pub fn off(owner: &str, kind: &str) -> Setting {
48
        Setting {
49
            owner: owner.into(),
50
            kind: kind.into(),
51
            every: "off".into(),
52
            destination: "dir".into(),
53
            target: String::new(),
54
            prefix: String::new(),
55
            provider: String::new(),
56
            keep_days: 30,
57
            updated_ms: 0,
58
            region: String::new(),
59
        }
60
    }
61
62
    /// Seconds between runs; none when off.
63
    pub fn period_s(&self) -> Option<i64> {
64
        match self.every.as_str() {
65
            "hourly" => Some(3600),
66
            "daily" => Some(86_400),
67
            _ => None,
68
        }
69
    }
70
}
71
72
/// One run, as the settings page and the API show it.
73
#[derive(Clone, Debug, sqlx::FromRow, Serialize)]
74
pub struct Run {
75
    pub id: i64,
76
    pub owner: String,
77
    pub started_ms: i64,
78
    pub finished_ms: Option<i64>,
79
    pub status: String,
80
    pub at: String,
81
    pub repos: i64,
82
    pub files: i64,
83
    pub sent: i64,
84
    pub bytes: i64,
85
    pub detail: String,
86
    pub trigger: String,
87
}
88
89
pub async fn setting(pool: &Pool, owner: &str) -> sqlx::Result<Option<Setting>> {
90
    sqlx::query_as("SELECT * FROM backup_settings WHERE owner = ?").bind(owner).fetch_optional(pool).await
91
}
92
93
pub async fn save(pool: &Pool, s: &Setting) -> sqlx::Result<()> {
94
    sqlx::query(
95
        "INSERT INTO backup_settings (owner, kind, every, destination, target, prefix, provider, keep_days, updated_ms, region) \
96
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(owner) DO UPDATE SET kind = excluded.kind, every = excluded.every, \
97
         destination = excluded.destination, target = excluded.target, prefix = excluded.prefix, provider = excluded.provider, \
98
         keep_days = excluded.keep_days, updated_ms = excluded.updated_ms, region = excluded.region",
99
    )
100
    .bind(&s.owner)
101
    .bind(&s.kind)
102
    .bind(&s.every)
103
    .bind(&s.destination)
104
    .bind(&s.target)
105
    .bind(&s.prefix)
106
    .bind(&s.provider)
107
    .bind(s.keep_days)
108
    .bind(db::now_ms())
109
    .bind(&s.region)
110
    .execute(pool)
111
    .await?;
112
    Ok(())
113
}
114
115
pub async fn runs(pool: &Pool, owner: &str, limit: i64) -> sqlx::Result<Vec<Run>> {
116
    sqlx::query_as("SELECT * FROM backup_runs WHERE owner = ? ORDER BY id DESC LIMIT ?").bind(owner).bind(limit).fetch_all(pool).await
117
}
118
119
/// What a repository may choose about its backup (§5): with its owner's,
120
/// left out, or at least hourly or daily whatever its owner's schedule.
121
pub const REPO_CHOICES: &[(&str, &str)] =
122
    &[("", "With its owner's backup"), ("hourly", "At least every hour"), ("daily", "At least every day"), ("off", "Left out")];
123
124
/// Whether `row` is in its owner's run: never when it is left out, always
125
/// when it has its own schedule, else as its owner's schedule says (a run
126
/// asked for by hand takes every repository not left out).
127
pub fn included(s: &Setting, row: &db::repos::RepoRow, by_hand: bool) -> bool {
128
    match row.backup.as_str() {
129
        "off" => false,
130
        "hourly" | "daily" => true,
131
        _ => by_hand || s.every != "off",
132
    }
133
}
134
135
/// Seconds between an owner's runs: the shortest of its own schedule and
136
/// its repositories' own; none when nothing is scheduled.
137
pub fn period_with(s: &Setting, repos: &[db::repos::RepoRow]) -> Option<i64> {
138
    let own = repos.iter().filter_map(|r| match r.backup.as_str() {
139
        "hourly" => Some(3600),
140
        "daily" => Some(86_400),
141
        _ => None,
142
    });
143
    s.period_s().into_iter().chain(own).min()
144
}
145
146
/// One owner's backup as Admin › Backups shows it (§5): its setting, its
147
/// newest run, and how many runs failed in the last day. An administrator
148
/// sees that it ran, never what it wrote.
149
#[derive(Clone, Debug, Serialize)]
150
pub struct Overview {
151
    pub setting: Setting,
152
    pub last: Option<Run>,
153
    pub failed_day: i64,
154
    /// The repositories it covers now.
155
    pub repos: i64,
156
}
157
158
impl Overview {
159
    /// Where its runs go, as a person reads it.
160
    pub fn place(&self) -> String {
161
        let s = &self.setting;
162
        match s.destination.as_str() {
163
            "gcs" if s.prefix.is_empty() => format!("gs://{}", s.target),
164
            "gcs" => format!("gs://{}/{}", s.target, s.prefix),
165
            "s3" if s.prefix.is_empty() => format!("s3://{} ({})", s.target, s.region),
166
            "s3" => format!("s3://{}/{} ({})", s.target, s.prefix, s.region),
167
            _ => format!("{}/{}", s.target, s.owner),
168
        }
169
    }
170
}
171
172
/// Every owner's backup (Admin › Backups), and the owners of repositories
173
/// that none covers, with how many each has.
174
pub async fn overview(pool: &Pool) -> sqlx::Result<(Vec<Overview>, Vec<(String, i64)>)> {
175
    let settings: Vec<Setting> = sqlx::query_as("SELECT * FROM backup_settings ORDER BY owner").fetch_all(pool).await?;
176
    let counts: Vec<(String, i64)> =
177
        sqlx::query_as("SELECT owner, COUNT(*) FROM repositories GROUP BY owner ORDER BY owner").fetch_all(pool).await?;
178
    let day_ago = db::now_ms() - 86_400_000;
179
    let mut out = Vec::new();
180
    for s in settings {
181
        let last = runs(pool, &s.owner, 1).await?.into_iter().next();
182
        let failed_day: i64 =
183
            sqlx::query_scalar("SELECT COUNT(*) FROM backup_runs WHERE owner = ? AND status = 'failed' AND started_ms >= ?")
184
                .bind(&s.owner)
185
                .bind(day_ago)
186
                .fetch_one(pool)
187
                .await?;
188
        let repos = db::repos::list(pool).await?.iter().filter(|r| r.owner == s.owner && included(&s, r, false)).count() as i64;
189
        out.push(Overview { setting: s, last, failed_day, repos });
190
    }
191
    let uncovered = counts.into_iter().filter(|(owner, _)| !out.iter().any(|o| o.setting.owner == *owner && o.repos > 0)).collect();
192
    Ok((out, uncovered))
193
}
194
195
// ---------------------------------------------------------------- where
196
197
/// Where backups are written: keys are `/`-separated paths.
198
pub trait Destination: Send + Sync {
199
    fn exists(&self, key: &str) -> Result<bool>;
200
    /// Write `bytes` at `key`, whole or not at all.
201
    fn put(&self, key: &str, bytes: &[u8]) -> Result<()>;
202
    fn get(&self, key: &str) -> Result<Vec<u8>>;
203
    /// The names directly under `prefix` (files and "directories").
204
    fn list(&self, prefix: &str) -> Result<Vec<String>>;
205
    /// Remove `key` and everything under it.
206
    fn remove(&self, key: &str) -> Result<()>;
207
}
208
209
/// A directory on the hub's host.
210
pub struct Directory {
211
    pub root: PathBuf,
212
}
213
214
impl Directory {
215
    fn path(&self, key: &str) -> Result<PathBuf> {
216
        if key.split('/').any(|p| p == ".." || p == ".") || key.starts_with('/') {
217
            bail!("not a clean key: {key}");
218
        }
219
        Ok(self.root.join(key))
220
    }
221
}
222
223
impl Destination for Directory {
224
    fn exists(&self, key: &str) -> Result<bool> {
225
        Ok(self.path(key)?.exists())
226
    }
227
    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {
228
        let p = self.path(key)?;
229
        std::fs::create_dir_all(p.parent().context("no parent")?)?;
230
        let tmp = p.with_extension(format!("part-{}", std::process::id()));
231
        std::fs::write(&tmp, bytes)?;
232
        std::fs::rename(&tmp, &p)?;
233
        Ok(())
234
    }
235
    fn get(&self, key: &str) -> Result<Vec<u8>> {
236
        let p = self.path(key)?;
237
        std::fs::read(&p).with_context(|| format!("read {}", p.display()))
238
    }
239
    fn list(&self, prefix: &str) -> Result<Vec<String>> {
240
        let p = self.path(prefix)?;
241
        let mut out: Vec<String> = match std::fs::read_dir(&p) {
242
            Ok(d) => d.flatten().map(|e| e.file_name().to_string_lossy().into_owned()).collect(),
243
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
244
            Err(e) => return Err(e.into()),
245
        };
246
        out.sort();
247
        Ok(out)
248
    }
249
    fn remove(&self, key: &str) -> Result<()> {
250
        let p = self.path(key)?;
251
        if p.is_dir() {
252
            std::fs::remove_dir_all(p)?;
253
        } else if p.exists() {
254
            std::fs::remove_file(p)?;
255
        }
256
        Ok(())
257
    }
258
}
259
260
/// A Cloud Storage bucket, reached keylessly (docs/backups.md §3, §6):
261
/// the hub mints a short token as `backup:OWNER`, Google's STS exchanges
262
/// it for an access token through the owner's workload identity
263
/// provider, and that writes the bucket. No key exists anywhere; the
264
/// access token lasts an hour and is fetched again before it runs out.
265
pub struct Gcs {
266
    bucket: String,
267
    /// Keys go under this, when set.
268
    prefix: String,
269
    storage: String,
270
    sts: String,
271
    /// The provider's resource name, `//iam.googleapis.com/projects/…`.
272
    provider: String,
273
    subject: String,
274
    app: Shared,
275
    client: reqwest::Client,
276
    rt: tokio::runtime::Handle,
277
    token: std::sync::Mutex<Option<(String, i64)>>,
278
}
279
280
impl Gcs {
281
    pub fn new(app: &Shared, s: &Setting) -> Gcs {
282
        Gcs {
283
            bucket: s.target.clone(),
284
            prefix: s.prefix.trim_matches('/').to_string(),
285
            storage: app.config.backups.storage_url.trim_end_matches('/').to_string(),
286
            sts: app.config.backups.sts_url.trim_end_matches('/').to_string(),
287
            provider: s.provider.clone(),
288
            subject: format!("backup:{}", s.owner),
289
            app: app.clone(),
290
            client: reqwest::Client::builder().timeout(std::time::Duration::from_secs(300)).build().unwrap_or_default(),
291
            rt: tokio::runtime::Handle::current(),
292
            token: std::sync::Mutex::new(None),
293
        }
294
    }
295
296
    fn name(&self, key: &str) -> String {
297
        if self.prefix.is_empty() {
298
            key.to_string()
299
        } else {
300
            format!("{}/{key}", self.prefix)
301
        }
302
    }
303
304
    /// An access token, exchanged for a fresh one of the hub's when the
305
    /// last is within five minutes of running out.
306
    fn access(&self) -> Result<String> {
307
        let now = db::now_ms();
308
        if let Some((t, until)) = self.token.lock().unwrap().as_ref() {
309
            if *until - now > 300_000 {
310
                return Ok(t.clone());
311
            }
312
        }
313
        let (token, expires_in) = self.rt.block_on(async {
314
            let issuer = self.app.oidc().await?;
315
            // Google's default audience for a provider: its name over https.
316
            let audience = format!("https:{}", self.provider);
317
            let jwt = issuer.mint_for(self.app.config.base_url(), &self.subject, &audience, db::now_ms(), 600);
318
            let res = self
319
                .client
320
                .post(format!("{}/v1/token", self.sts))
321
                .json(&serde_json::json!({
322
                    "grantType": "urn:ietf:params:oauth:grant-type:token-exchange",
323
                    "audience": self.provider,
324
                    "scope": "https://www.googleapis.com/auth/devstorage.read_write",
325
                    "requestedTokenType": "urn:ietf:params:oauth:token-type:access_token",
326
                    "subjectToken": jwt,
327
                    "subjectTokenType": "urn:ietf:params:oauth:token-type:jwt",
328
                }))
329
                .send()
330
                .await?;
331
            let status = res.status();
332
            let body: serde_json::Value = res.json().await.unwrap_or_default();
333
            if !status.is_success() {
334
                bail!(
335
                    "Google's token exchange refused {}: {} (check the provider's issuer is this hub's address and its attribute condition admits backup:)",
336
                    self.subject,
337
                    body["error_description"].as_str().or(body["error"].as_str()).unwrap_or("no reason given")
338
                );
339
            }
340
            let token = body["access_token"].as_str().context("the token exchange gave no access token")?.to_string();
341
            anyhow::Ok((token, body["expires_in"].as_i64().unwrap_or(3600)))
342
        })?;
343
        *self.token.lock().unwrap() = Some((token.clone(), now + expires_in * 1000));
344
        Ok(token)
345
    }
346
347
    fn object_url(&self, key: &str) -> Result<reqwest::Url> {
348
        let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
349
        u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&self.name(key));
350
        Ok(u)
351
    }
352
353
    fn refused(&self, what: &str, status: reqwest::StatusCode, body: &str) -> anyhow::Error {
354
        let hint = if status == reqwest::StatusCode::FORBIDDEN || status == reqwest::StatusCode::UNAUTHORIZED {
355
            format!(" (grant roles/storage.objectAdmin on gs://{} to the principal …/subject/{})", self.bucket, self.subject)
356
        } else {
357
            String::new()
358
        };
359
        anyhow::anyhow!("GCS {what}: {status}{hint}: {}", body.chars().take(200).collect::<String>())
360
    }
361
362
    /// Every object name under `full` (no delimiter), for removal.
363
    fn all_under(&self, full: &str) -> Result<Vec<String>> {
364
        let mut out = Vec::new();
365
        let mut page: Option<String> = None;
366
        loop {
367
            let token = self.access()?;
368
            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
369
            u.query_pairs_mut().append_pair("prefix", full).append_pair("fields", "items(name),nextPageToken");
370
            if let Some(p) = &page {
371
                u.query_pairs_mut().append_pair("pageToken", p);
372
            }
373
            let body: serde_json::Value = self.rt.block_on(async {
374
                let res = self.client.get(u).bearer_auth(&token).send().await?;
375
                let status = res.status();
376
                let text = res.text().await?;
377
                if !status.is_success() {
378
                    return Err(self.refused("list", status, &text));
379
                }
380
                anyhow::Ok(serde_json::from_str(&text)?)
381
            })?;
382
            out.extend(body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str().map(String::from)));
383
            match body["nextPageToken"].as_str() {
384
                Some(p) => page = Some(p.to_string()),
385
                None => return Ok(out),
386
            }
387
        }
388
    }
389
}
390
391
/// An S3 bucket, written keylessly (§3): the hub's own short-lived token
392
/// (`sub = backup:OWNER`, audience `sts.amazonaws.com`) is exchanged at AWS
393
/// STS (AssumeRoleWithWebIdentity) for an hour's credentials of the role
394
/// the owner names, and they sign each request (Signature Version 4).
395
pub struct S3 {
396
    bucket: String,
397
    prefix: String,
398
    region: String,
399
    /// The bucket's endpoint: scheme, host and any path before keys.
400
    base: reqwest::Url,
401
    sts: String,
402
    /// The IAM role's ARN.
403
    role: String,
404
    subject: String,
405
    app: Shared,
406
    client: reqwest::Client,
407
    rt: tokio::runtime::Handle,
408
    creds: std::sync::Mutex<Option<Creds>>,
409
}
410
411
#[derive(Clone)]
412
struct Creds {
413
    id: String,
414
    secret: String,
415
    token: String,
416
    until: i64,
417
}
418
419
/// The audience AWS expects of a web identity token.
420
pub const AWS_AUDIENCE: &str = "sts.amazonaws.com";
421
422
/// Percent-encoding as AWS signs it: everything but unreserved characters,
423
/// and `/` too unless `keep_slash`.
424
pub fn aws_encode(s: &str, keep_slash: bool) -> String {
425
    let mut out = String::new();
426
    for b in s.bytes() {
427
        match b {
428
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => out.push(b as char),
429
            b'/' if keep_slash => out.push('/'),
430
            _ => out.push_str(&format!("%{b:02X}")),
431
        }
432
    }
433
    out
434
}
435
436
fn hmac(key: &[u8], data: &str) -> Vec<u8> {
437
    use hmac::Mac;
438
    let mut m = hmac::Hmac::<sha2::Sha256>::new_from_slice(key).expect("any key length");
439
    m.update(data.as_bytes());
440
    m.finalize().into_bytes().to_vec()
441
}
442
443
fn sha256_hex(data: &[u8]) -> String {
444
    use sha2::Digest;
445
    hex::encode(sha2::Sha256::digest(data))
446
}
447
448
/// AWS Signature Version 4: the `Authorization` header for one request.
449
/// `uri` is the path as encoded on the wire, `query` the decoded pairs,
450
/// `headers` the signed ones (any case), `payload` the body's SHA-256 in
451
/// hex, `amz_date` as `20130524T000000Z`.
452
#[allow(clippy::too_many_arguments)]
453
pub fn sigv4(
454
    method: &str,
455
    uri: &str,
456
    query: &[(String, String)],
457
    headers: &[(String, String)],
458
    payload: &str,
459
    (id, secret): (&str, &str),
460
    region: &str,
461
    service: &str,
462
    amz_date: &str,
463
) -> String {
464
    let mut q: Vec<(String, String)> = query.iter().map(|(k, v)| (aws_encode(k, false), aws_encode(v, false))).collect();
465
    q.sort();
466
    let query = q.iter().map(|(k, v)| format!("{k}={v}")).collect::<Vec<_>>().join("&");
467
    let mut h: Vec<(String, String)> = headers.iter().map(|(k, v)| (k.to_ascii_lowercase(), v.trim().to_string())).collect();
468
    h.sort();
469
    let canonical_headers: String = h.iter().map(|(k, v)| format!("{k}:{v}\n")).collect();
470
    let signed = h.iter().map(|(k, _)| k.as_str()).collect::<Vec<_>>().join(";");
471
    let request = format!("{method}\n{uri}\n{query}\n{canonical_headers}\n{signed}\n{payload}");
472
    let date = &amz_date[..8];
473
    let scope = format!("{date}/{region}/{service}/aws4_request");
474
    let to_sign = format!("AWS4-HMAC-SHA256\n{amz_date}\n{scope}\n{}", sha256_hex(request.as_bytes()));
475
    let key = hmac(&hmac(&hmac(&hmac(format!("AWS4{secret}").as_bytes(), date), region), service), "aws4_request");
476
    let signature = hex::encode(hmac(&key, &to_sign));
477
    format!("AWS4-HMAC-SHA256 Credential={id}/{scope}, SignedHeaders={signed}, Signature={signature}")
478
}
479
480
/// The text of the first `<tag>` in `xml`, unescaped.
481
fn xml_text(xml: &str, tag: &str) -> Option<String> {
482
    xml_blocks(xml, tag).into_iter().next().map(xml_unescape)
483
}
484
485
/// The inside of every `<tag>…</tag>` in `xml`.
486
fn xml_blocks<'a>(xml: &'a str, tag: &str) -> Vec<&'a str> {
487
    let (open, close) = (format!("<{tag}>"), format!("</{tag}>"));
488
    let mut out = Vec::new();
489
    let mut rest = xml;
490
    while let Some(i) = rest.find(&open) {
491
        let after = &rest[i + open.len()..];
492
        let Some(j) = after.find(&close) else { break };
493
        out.push(&after[..j]);
494
        rest = &after[j + close.len()..];
495
    }
496
    out
497
}
498
499
fn xml_unescape(s: &str) -> String {
500
    s.replace("&lt;", "<").replace("&gt;", ">").replace("&quot;", "\"").replace("&apos;", "'").replace("&amp;", "&")
501
}
502
503
fn amz_date(ms: i64) -> String {
504
    let t = time::OffsetDateTime::from_unix_timestamp_nanos(ms as i128 * 1_000_000).unwrap_or(time::OffsetDateTime::UNIX_EPOCH);
505
    format!("{:04}{:02}{:02}T{:02}{:02}{:02}Z", t.year(), u8::from(t.month()), t.day(), t.hour(), t.minute(), t.second())
506
}
507
508
impl S3 {
509
    pub fn new(app: &Shared, s: &Setting) -> S3 {
510
        let url = app.config.backups.s3_url.replace("{bucket}", &s.target).replace("{region}", &s.region);
511
        S3 {
512
            bucket: s.target.clone(),
513
            prefix: s.prefix.trim_matches('/').to_string(),
514
            region: s.region.clone(),
515
            base: reqwest::Url::parse(url.trim_end_matches('/')).unwrap_or_else(|_| reqwest::Url::parse("https://invalid.").unwrap()),
516
            sts: app.config.backups.aws_sts_url.trim_end_matches('/').to_string(),
517
            role: s.provider.clone(),
518
            subject: format!("backup:{}", s.owner),
519
            app: app.clone(),
520
            client: reqwest::Client::builder().timeout(std::time::Duration::from_secs(300)).build().unwrap_or_default(),
521
            rt: tokio::runtime::Handle::current(),
522
            creds: std::sync::Mutex::new(None),
523
        }
524
    }
525
526
    fn name(&self, key: &str) -> String {
527
        if self.prefix.is_empty() {
528
            key.to_string()
529
        } else {
530
            format!("{}/{key}", self.prefix)
531
        }
532
    }
533
534
    /// Credentials of the owner's role, exchanged again when the last are
535
    /// within five minutes of running out.
536
    fn creds(&self) -> Result<Creds> {
537
        let now = db::now_ms();
538
        if let Some(c) = self.creds.lock().unwrap().as_ref() {
539
            if c.until - now > 300_000 {
540
                return Ok(c.clone());
541
            }
542
        }
543
        let session: String = format!("oxygenhub-{}", self.subject.trim_start_matches("backup:"))
544
            .chars()
545
            .filter(|c| c.is_ascii_alphanumeric() || "=,.@-_".contains(*c))
546
            .take(64)
547
            .collect();
548
        let c = self.rt.block_on(async {
549
            let issuer = self.app.oidc().await?;
550
            let jwt = issuer.mint_for(self.app.config.base_url(), &self.subject, AWS_AUDIENCE, db::now_ms(), 600);
551
            let res = self
552
                .client
553
                .post(format!("{}/", self.sts))
554
                .form(&[
555
                    ("Action", "AssumeRoleWithWebIdentity"),
556
                    ("Version", "2011-06-15"),
557
                    ("RoleArn", self.role.as_str()),
558
                    ("RoleSessionName", session.as_str()),
559
                    ("WebIdentityToken", jwt.as_str()),
560
                    ("DurationSeconds", "3600"),
561
                ])
562
                .send()
563
                .await?;
564
            let status = res.status();
565
            let text = res.text().await.unwrap_or_default();
566
            if !status.is_success() {
567
                bail!(
568
                    "AWS STS refused {} for {}: {} (check the role's trust policy: this hub's address as its OIDC provider, audience {AWS_AUDIENCE}, and sub {})",
569
                    self.subject,
570
                    self.role,
571
                    xml_text(&text, "Message").unwrap_or_else(|| status.to_string()),
572
                    self.subject
573
                );
574
            }
575
            let get = |t: &str| xml_text(&text, t).with_context(|| format!("AWS STS gave no {t}"));
576
            anyhow::Ok(Creds { id: get("AccessKeyId")?, secret: get("SecretAccessKey")?, token: get("SessionToken")?, until: now + 3_600_000 })
577
        })?;
578
        *self.creds.lock().unwrap() = Some(c.clone());
579
        Ok(c)
580
    }
581
582
    /// One signed request: `key` is an object's (none for the bucket).
583
    fn send(
584
        &self,
585
        method: reqwest::Method,
586
        key: Option<&str>,
587
        query: &[(String, String)],
588
        body: Vec<u8>,
589
    ) -> Result<(reqwest::StatusCode, Vec<u8>)> {
590
        let c = self.creds()?;
591
        let base_path = self.base.path().trim_end_matches('/');
592
        let uri = match key {
593
            Some(k) => format!("{base_path}/{}", aws_encode(&self.name(k), true)),
594
            None => format!("{base_path}/"),
595
        };
596
        let host = match self.base.port() {
597
            Some(p) => format!("{}:{p}", self.base.host_str().unwrap_or_default()),
598
            None => self.base.host_str().unwrap_or_default().to_string(),
599
        };
600
        let payload = sha256_hex(&body);
601
        let date = amz_date(db::now_ms());
602
        let headers = vec![
603
            ("host".to_string(), host.clone()),
604
            ("x-amz-content-sha256".to_string(), payload.clone()),
605
            ("x-amz-date".to_string(), date.clone()),
606
            ("x-amz-security-token".to_string(), c.token.clone()),
607
        ];
608
        let auth = sigv4(method.as_str(), &uri, query, &headers, &payload, (&c.id, &c.secret), &self.region, "s3", &date);
609
        let mut q: Vec<(String, String)> = query.iter().map(|(k, v)| (aws_encode(k, false), aws_encode(v, false))).collect();
610
        q.sort();
611
        let qs = q.iter().map(|(k, v)| format!("{k}={v}")).collect::<Vec<_>>().join("&");
612
        let url = format!("{}://{host}{uri}{}{qs}", self.base.scheme(), if qs.is_empty() { "" } else { "?" });
613
        let method_has_body = method == reqwest::Method::PUT;
614
        self.rt.block_on(async {
615
            let mut req = self
616
                .client
617
                .request(method, &url)
618
                .header("authorization", auth)
619
                .header("x-amz-content-sha256", &payload)
620
                .header("x-amz-date", &date)
621
                .header("x-amz-security-token", &c.token);
622
            if !body.is_empty() {
623
                req = req.header("content-type", "application/octet-stream");
624
            }
625
            // Always the length, an empty body's too: S3 and GCS refuse
626
            // an upload without one (411 Length Required).
627
            if method_has_body {
628
                req = req.header("content-length", body.len().to_string()).body(body);
629
            }
630
            let res = req.send().await?;
631
            let status = res.status();
632
            anyhow::Ok((status, res.bytes().await?.to_vec()))
633
        })
634
    }
635
636
    fn refused(&self, what: &str, status: reqwest::StatusCode, body: &[u8]) -> anyhow::Error {
637
        let text = String::from_utf8_lossy(body);
638
        let hint = if status == reqwest::StatusCode::FORBIDDEN {
639
            format!(" (the role {} needs s3:GetObject, s3:PutObject, s3:DeleteObject and s3:ListBucket on {})", self.role, self.bucket)
640
        } else {
641
            String::new()
642
        };
643
        anyhow::anyhow!("S3 {what}: {status}{hint}: {}", xml_text(&text, "Message").unwrap_or_else(|| text.chars().take(200).collect()))
644
    }
645
646
    /// The keys under `full` (`delimited`: and the next level's prefixes).
647
    fn listing(&self, full: &str, delimited: bool) -> Result<(Vec<String>, Vec<String>)> {
648
        let (mut keys, mut prefixes) = (Vec::new(), Vec::new());
649
        let mut token: Option<String> = None;
650
        loop {
651
            let mut q = vec![("list-type".to_string(), "2".to_string()), ("prefix".to_string(), full.to_string())];
652
            if delimited {
653
                q.push(("delimiter".to_string(), "/".to_string()));
654
            }
655
            if let Some(t) = &token {
656
                q.push(("continuation-token".to_string(), t.clone()));
657
            }
658
            let (status, body) = self.send(reqwest::Method::GET, None, &q, Vec::new())?;
659
            if !status.is_success() {
660
                return Err(self.refused("list", status, &body));
661
            }
662
            let text = String::from_utf8_lossy(&body).into_owned();
663
            for c in xml_blocks(&text, "Contents") {
664
                keys.extend(xml_text(c, "Key"));
665
            }
666
            for c in xml_blocks(&text, "CommonPrefixes") {
667
                prefixes.extend(xml_text(c, "Prefix"));
668
            }
669
            match (xml_text(&text, "IsTruncated").as_deref(), xml_text(&text, "NextContinuationToken")) {
670
                (Some("true"), Some(t)) => token = Some(t),
671
                _ => return Ok((keys, prefixes)),
672
            }
673
        }
674
    }
675
}
676
677
impl Destination for S3 {
678
    fn exists(&self, key: &str) -> Result<bool> {
679
        let (status, body) = self.send(reqwest::Method::HEAD, Some(key), &[], Vec::new())?;
680
        match status {
681
            s if s.is_success() => Ok(true),
682
            reqwest::StatusCode::NOT_FOUND => Ok(false),
683
            s => Err(self.refused("read", s, &body)),
684
        }
685
    }
686
    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {
687
        let (status, body) = self.send(reqwest::Method::PUT, Some(key), &[], bytes.to_vec())?;
688
        if !status.is_success() {
689
            return Err(self.refused("write", status, &body));
690
        }
691
        Ok(())
692
    }
693
    fn get(&self, key: &str) -> Result<Vec<u8>> {
694
        let (status, body) = self.send(reqwest::Method::GET, Some(key), &[], Vec::new())?;
695
        if !status.is_success() {
696
            return Err(self.refused(&format!("read {key}"), status, &body));
697
        }
698
        Ok(body)
699
    }
700
    fn list(&self, prefix: &str) -> Result<Vec<String>> {
701
        let full = format!("{}/", self.name(prefix.trim_end_matches('/')));
702
        let (keys, prefixes) = self.listing(&full, true)?;
703
        let mut out: std::collections::BTreeSet<String> = keys.iter().map(|k| k.trim_start_matches(&full).to_string()).collect();
704
        out.extend(prefixes.iter().map(|p| p.trim_start_matches(&full).trim_end_matches('/').to_string()));
705
        Ok(out.into_iter().filter(|n| !n.is_empty()).collect())
706
    }
707
    fn remove(&self, key: &str) -> Result<()> {
708
        let full = self.name(key);
709
        let (mut keys, _) = self.listing(&format!("{full}/"), false)?;
710
        keys.push(full);
711
        for k in keys {
712
            let rel = if self.prefix.is_empty() { k.clone() } else { k.trim_start_matches(&format!("{}/", self.prefix)).to_string() };
713
            let (status, body) = self.send(reqwest::Method::DELETE, Some(&rel), &[], Vec::new())?;
714
            if !status.is_success() && status != reqwest::StatusCode::NOT_FOUND {
715
                return Err(self.refused("remove", status, &body));
716
            }
717
        }
718
        Ok(())
719
    }
720
}
721
722
impl Destination for Gcs {
723
    fn exists(&self, key: &str) -> Result<bool> {
724
        let token = self.access()?;
725
        let mut u = self.object_url(key)?;
726
        u.query_pairs_mut().append_pair("fields", "name");
727
        self.rt.block_on(async {
728
            let res = self.client.get(u).bearer_auth(&token).send().await?;
729
            match res.status() {
730
                s if s.is_success() => Ok(true),
731
                reqwest::StatusCode::NOT_FOUND => Ok(false),
732
                s => Err(self.refused("read", s, &res.text().await.unwrap_or_default())),
733
            }
734
        })
735
    }
736
    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {
737
        let token = self.access()?;
738
        let mut u = reqwest::Url::parse(&format!("{}/upload/storage/v1/b/{}/o", self.storage, self.bucket))?;
739
        u.query_pairs_mut().append_pair("uploadType", "media").append_pair("name", &self.name(key));
740
        let body = bytes.to_vec();
741
        self.rt.block_on(async {
742
            // The length always, an empty object's too: Google refuses an
743
            // upload without one (411 Length Required).
744
            let res = self
745
                .client
746
                .post(u)
747
                .bearer_auth(&token)
748
                .header("content-type", "application/octet-stream")
749
                .header("content-length", body.len().to_string())
750
                .body(body)
751
                .send()
752
                .await?;
753
            let status = res.status();
754
            if !status.is_success() {
755
                return Err(self.refused("write", status, &res.text().await.unwrap_or_default()));
756
            }
757
            Ok(())
758
        })
759
    }
760
    fn get(&self, key: &str) -> Result<Vec<u8>> {
761
        let token = self.access()?;
762
        let mut u = self.object_url(key)?;
763
        u.query_pairs_mut().append_pair("alt", "media");
764
        self.rt.block_on(async {
765
            let res = self.client.get(u).bearer_auth(&token).send().await?;
766
            let status = res.status();
767
            if !status.is_success() {
768
                return Err(self.refused(&format!("read {key}"), status, &res.text().await.unwrap_or_default()));
769
            }
770
            Ok(res.bytes().await?.to_vec())
771
        })
772
    }
773
    fn list(&self, prefix: &str) -> Result<Vec<String>> {
774
        let full = format!("{}/", self.name(prefix.trim_end_matches('/')));
775
        let mut out = std::collections::BTreeSet::new();
776
        let mut page: Option<String> = None;
777
        loop {
778
            let token = self.access()?;
779
            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
780
            u.query_pairs_mut()
781
                .append_pair("prefix", &full)
782
                .append_pair("delimiter", "/")
783
                .append_pair("fields", "items(name),prefixes,nextPageToken");
784
            if let Some(p) = &page {
785
                u.query_pairs_mut().append_pair("pageToken", p);
786
            }
787
            let body: serde_json::Value = self.rt.block_on(async {
788
                let res = self.client.get(u).bearer_auth(&token).send().await?;
789
                let status = res.status();
790
                let text = res.text().await?;
791
                if !status.is_success() {
792
                    return Err(self.refused("list", status, &text));
793
                }
794
                anyhow::Ok(serde_json::from_str(&text)?)
795
            })?;
796
            for n in body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str()) {
797
                out.insert(n.trim_start_matches(&full).to_string());
798
            }
799
            for p in body["prefixes"].as_array().into_iter().flatten().filter_map(|p| p.as_str()) {
800
                out.insert(p.trim_start_matches(&full).trim_end_matches('/').to_string());
801
            }
802
            match body["nextPageToken"].as_str() {
803
                Some(p) => page = Some(p.to_string()),
804
                None => return Ok(out.into_iter().filter(|n| !n.is_empty()).collect()),
805
            }
806
        }
807
    }
808
    fn remove(&self, key: &str) -> Result<()> {
809
        let mut names = self.all_under(&format!("{}/", self.name(key)))?;
810
        names.push(self.name(key));
811
        for n in names {
812
            let token = self.access()?;
813
            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;
814
            u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&n);
815
            self.rt.block_on(async {
816
                let res = self.client.delete(u).bearer_auth(&token).send().await?;
817
                match res.status() {
818
                    s if s.is_success() || s == reqwest::StatusCode::NOT_FOUND => Ok(()),
819
                    s => Err(self.refused("delete", s, &res.text().await.unwrap_or_default())),
820
                }
821
            })?;
822
        }
823
        Ok(())
824
    }
825
}
826
827
/// Check a destination end to end: write a probe, read it back, remove it.
828
pub fn probe(dest: &dyn Destination) -> Result<()> {
829
    let marker = format!("probe {}", db::now_ms());
830
    dest.put(".oxyhub-probe", marker.as_bytes())?;
831
    if dest.get(".oxyhub-probe")? != marker.as_bytes() {
832
        bail!("the probe read back differently");
833
    }
834
    dest.remove(".oxyhub-probe")
835
}
836
837
/// The owner's destination, from its setting and the site's rules.
838
pub fn destination(app: &Shared, s: &Setting) -> Result<Box<dyn Destination>> {
839
    match s.destination.as_str() {
840
        "dir" => {
841
            let allowed = &app.config.backups.directories;
842
            let target = PathBuf::from(&s.target);
843
            if !allowed.contains(&target) {
844
                bail!("{} is not a backup directory this hub allows (hub.toml [backups] directories)", s.target);
845
            }
846
            let mut root = target.join(&s.owner);
847
            if !s.prefix.is_empty() {
848
                root = root.join(s.prefix.trim_matches('/'));
849
            }
850
            Ok(Box::new(Directory { root }))
851
        }
852
        "gcs" => {
853
            if s.target.is_empty() || s.provider.is_empty() {
854
                bail!("a GCS destination needs its bucket and its workload identity provider");
855
            }
856
            Ok(Box::new(Gcs::new(app, s)))
857
        }
858
        "s3" => {
859
            if s.target.is_empty() || s.provider.is_empty() || s.region.is_empty() {
860
                bail!("an S3 destination needs its bucket, its region and the IAM role the hub assumes");
861
            }
862
            Ok(Box::new(S3::new(app, s)))
863
        }
864
        other => bail!("unknown destination {other:?}"),
865
    }
866
}
867
868
// ---------------------------------------------------------------- what
869
870
/// One file of a repository's store, in a manifest.
871
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
872
pub struct File {
873
    pub path: String,
874
    pub size: u64,
875
    /// Its modification time in nanoseconds: unchanged size and time mean
876
    /// unchanged content, so it is not hashed again.
877
    pub mtime: i128,
878
    pub hash: String,
879
}
880
881
/// A repository at one run.
882
#[derive(Clone, Debug, Serialize, Deserialize)]
883
pub struct Manifest {
884
    pub owner: String,
885
    pub name: String,
886
    /// `repos` or `wikis`.
887
    pub kind: String,
888
    pub files: Vec<File>,
889
}
890
891
fn blob_key(hash: &str) -> String {
892
    format!("blobs/{}/{hash}", &hash[..2])
893
}
894
895
/// The large files that never change once written come first.
896
fn immutable(path: &str) -> bool {
897
    path.starts_with(".oxy/objects/") && (path.contains("/pack-") || path.contains("/loose/"))
898
}
899
900
#[derive(Default)]
901
struct Sent {
902
    files: i64,
903
    sent: i64,
904
    bytes: i64,
905
}
906
907
/// Walk `dir`, sending each file the destination lacks; `known` says which
908
/// paths are unchanged since the last run (and their hashes).
909
fn copy_dir(dest: &dyn Destination, dir: &Path, known: &HashMap<String, File>, only_immutable: bool, sent: &mut Sent) -> Result<Vec<File>> {
910
    let mut out = Vec::new();
911
    let mut stack = vec![dir.to_path_buf()];
912
    while let Some(d) = stack.pop() {
913
        let entries = match std::fs::read_dir(&d) {
914
            Ok(e) => e,
915
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue,
916
            Err(e) => return Err(e.into()),
917
        };
918
        for e in entries.flatten() {
919
            let p = e.path();
920
            let ft = e.file_type()?;
921
            if ft.is_dir() {
922
                stack.push(p);
923
                continue;
924
            }
925
            if !ft.is_file() {
926
                continue;
927
            }
928
            let rel = p.strip_prefix(dir).unwrap_or(&p).to_string_lossy().replace('\\', "/");
929
            if only_immutable && !immutable(&rel) {
930
                continue;
931
            }
932
            // A file packed away while the walk ran is simply gone.
933
            let Ok(md) = std::fs::metadata(&p) else { continue };
934
            let mtime =
935
                md.modified().ok().and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()).map(|d| d.as_nanos() as i128).unwrap_or(0);
936
            let size = md.len();
937
            sent.files += 1;
938
            if let Some(k) = known.get(&rel).filter(|k| k.size == size && k.mtime == mtime) {
939
                out.push(k.clone());
940
                continue;
941
            }
942
            let Ok(bytes) = std::fs::read(&p) else { continue };
943
            let hash = blake3::hash(&bytes).to_hex().to_string();
944
            let key = blob_key(&hash);
945
            if !dest.exists(&key)? {
946
                dest.put(&key, &bytes)?;
947
                sent.sent += 1;
948
                sent.bytes += bytes.len() as i64;
949
            }
950
            out.push(File { path: rel, size, mtime, hash });
951
        }
952
    }
953
    out.sort_by(|a, b| a.path.cmp(&b.path));
954
    Ok(out)
955
}
956
957
// ---------------------------------------------------------------- records
958
959
/// Tables and columns never written: secrets, credentials, sessions.
960
fn private_table(t: &str) -> bool {
961
    ["secret", "token", "session", "password", "totp", "recovery", "sqlx"].iter().any(|w| t.contains(w))
962
}
963
964
fn private_column(c: &str) -> bool {
965
    ["secret", "token", "password", "salt", "totp", "sealed", "hash_key"].iter().any(|w| c.contains(w))
966
}
967
968
async fn tables(pool: &Pool) -> Result<Vec<String>> {
969
    let mut t: Vec<String> =
970
        sqlx::query_scalar("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'").fetch_all(pool).await?;
971
    t.retain(|t| !private_table(t));
972
    t.sort();
973
    Ok(t)
974
}
975
976
async fn columns(pool: &Pool, table: &str) -> Result<Vec<String>> {
977
    Ok(sqlx::query_scalar(&format!("SELECT name FROM pragma_table_info('{table}')")).fetch_all(pool).await?)
978
}
979
980
/// The rows of `table` matching `cond`, as a JSON array, private columns
981
/// left out and blobs as hex.
982
async fn rows(pool: &Pool, table: &str, cond: &str) -> Result<serde_json::Value> {
983
    let cols = columns(pool, table).await?;
984
    let fields: Vec<String> = cols
985
        .iter()
986
        .filter(|c| !private_column(c))
987
        .map(|c| format!("'{c}', CASE WHEN typeof(\"{c}\") = 'blob' THEN hex(\"{c}\") ELSE \"{c}\" END"))
988
        .collect();
989
    if fields.is_empty() {
990
        return Ok(serde_json::json!([]));
991
    }
992
    let sql = format!("SELECT json_group_array(json_object({})) FROM \"{table}\" WHERE {cond}", fields.join(", "));
993
    let text: String = sqlx::query_scalar(&sql).fetch_one(pool).await?;
994
    Ok(serde_json::from_str(&text)?)
995
}
996
997
/// Every row about something: the tables with column `key` equal to `id`,
998
/// then, three levels deep, the tables whose foreign keys point at rows
999
/// already taken (an issue's comments, a proposal's reviews and their
1000
/// threads).
1001
async fn records(pool: &Pool, key: &str, id: i64) -> Result<serde_json::Value> {
1002
    let all = tables(pool).await?;
1003
    let mut conds: Vec<(String, String)> = Vec::new();
1004
    for t in &all {
1005
        if columns(pool, t).await?.iter().any(|c| c == key) {
1006
            conds.push((t.clone(), format!("\"{key}\" = {id}")));
1007
        }
1008
    }
1009
    for _ in 0..3 {
1010
        let mut more = Vec::new();
1011
        for t in &all {
1012
            if conds.iter().any(|(c, _)| c == t) || more.iter().any(|(c, _): &(String, String)| c == t) {
1013
                continue;
1014
            }
1015
            let fks: Vec<(String, String, String)> =
1016
                sqlx::query_as(&format!("SELECT \"table\", \"from\", \"to\" FROM pragma_foreign_key_list('{t}')")).fetch_all(pool).await?;
1017
            if let Some((parent, from, to, pcond)) = fks
1018
                .iter()
1019
                .find_map(|(p, f, to)| conds.iter().find(|(c, _)| c == p).map(|(_, pc)| (p.clone(), f.clone(), to.clone(), pc.clone())))
1020
            {
1021
                more.push((t.clone(), format!("\"{from}\" IN (SELECT \"{to}\" FROM \"{parent}\" WHERE {pcond})")));
1022
            }
1023
        }
1024
        if more.is_empty() {
1025
            break;
1026
        }
1027
        conds.extend(more);
1028
    }
1029
    let mut out = serde_json::Map::new();
1030
    for (t, cond) in conds {
1031
        let r = rows(pool, &t, &cond).await?;
1032
        if r.as_array().is_some_and(|a| !a.is_empty()) {
1033
            out.insert(t, r);
1034
        }
1035
    }
1036
    Ok(serde_json::Value::Object(out))
1037
}
1038
1039
/// The key of a run's records naming the accounts, teams and organisations
1040
/// their rows point at, by id: `{"accounts": {"7": "andy"}, …}`.
1041
const NAMES: &str = "_names";
1042
1043
/// Tables a restore leaves out: the search index (rebuilt from what is
1044
/// restored), and what is passing or personal rather than the
1045
/// repository's: inboxes, webhook deliveries, runners' declines.
1046
const NOT_RESTORED: [&str; 5] = ["search", "notifications", "deliveries", "flow_declines", NAMES];
1047
1048
async fn fk_list(pool: &Pool, table: &str) -> Result<Vec<(String, String, String)>> {
1049
    Ok(sqlx::query_as(&format!("SELECT \"table\", \"from\", \"to\" FROM pragma_foreign_key_list('{table}')")).fetch_all(pool).await?)
1050
}
1051
1052
/// The names behind the account, team and organisation ids that `rec`'s
1053
/// rows point at, so a restore on another hub finds them again by name.
1054
async fn names_of(pool: &Pool, rec: &serde_json::Value) -> Result<serde_json::Value> {
1055
    let mut ids: HashMap<String, std::collections::BTreeSet<i64>> = HashMap::new();
1056
    for (table, rows) in rec.as_object().into_iter().flatten() {
1057
        let Some(rows) = rows.as_array() else { continue };
1058
        for (parent, from, _) in fk_list(pool, table).await? {
1059
            if ["accounts", "teams", "organisations"].contains(&parent.as_str()) {
1060
                ids.entry(parent).or_default().extend(rows.iter().filter_map(|r| r[&from].as_i64()));
1061
            }
1062
        }
1063
    }
1064
    let mut out = serde_json::Map::new();
1065
    for (kind, sql) in [
1066
        ("accounts", "SELECT id, username FROM accounts WHERE id = ?"),
1067
        ("organisations", "SELECT id, name FROM organisations WHERE id = ?"),
1068
        ("teams", "SELECT t.id, o.name || '/' || t.name FROM teams t JOIN organisations o ON o.id = t.org_id WHERE t.id = ?"),
1069
    ] {
1070
        let mut m = serde_json::Map::new();
1071
        for id in ids.get(kind).into_iter().flatten() {
1072
            if let Some((id, name)) = sqlx::query_as::<_, (i64, String)>(sql).bind(id).fetch_optional(pool).await? {
1073
                m.insert(id.to_string(), name.into());
1074
            }
1075
        }
1076
        out.insert(kind.into(), m.into());
1077
    }
1078
    Ok(out.into())
1079
}
1080
1081
/// The id `name` has on this hub as one of `_names`' kinds.
1082
async fn id_by_name(pool: &Pool, kind: &str, name: &str) -> Result<Option<i64>> {
1083
    Ok(match kind {
1084
        "accounts" => sqlx::query_scalar("SELECT id FROM accounts WHERE username = ?").bind(name).fetch_optional(pool).await?,
1085
        "organisations" => sqlx::query_scalar("SELECT id FROM organisations WHERE name = ?").bind(name).fetch_optional(pool).await?,
1086
        "teams" => match name.split_once('/') {
1087
            Some((org, team)) => {
1088
                sqlx::query_scalar("SELECT t.id FROM teams t JOIN organisations o ON o.id = t.org_id WHERE o.name = ? AND t.name = ?")
1089
                    .bind(org)
1090
                    .bind(team)
1091
                    .fetch_optional(pool)
1092
                    .await?
1093
            }
1094
            None => None,
1095
        },
1096
        _ => None,
1097
    })
1098
}
1099
1100
/// One column of a table on this hub.
1101
struct Col {
1102
    name: String,
1103
    ty: String,
1104
    notnull: bool,
1105
    default: Option<String>,
1106
    pk: i64,
1107
}
1108
1109
async fn table_cols(pool: &Pool, table: &str) -> Result<Vec<Col>> {
1110
    let rows: Vec<(String, String, i64, Option<String>, i64)> =
1111
        sqlx::query_as(&format!("SELECT name, type, \"notnull\", dflt_value, pk FROM pragma_table_info('{table}')"))
1112
            .fetch_all(pool)
1113
            .await?;
1114
    Ok(rows.into_iter().map(|(name, ty, notnull, default, pk)| Col { name, ty, notnull: notnull != 0, default, pk }).collect())
1115
}
1116
1117
/// Bind one value of a backed-up row: blobs were written as hex.
1118
fn bind_json<'q>(
1119
    q: sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>>,
1120
    v: serde_json::Value,
1121
    blob: bool,
1122
) -> sqlx::query::Query<'q, sqlx::Sqlite, sqlx::sqlite::SqliteArguments<'q>> {
1123
    match v {
1124
        serde_json::Value::Null => q.bind(None::<i64>),
1125
        serde_json::Value::Bool(b) => q.bind(b as i64),
1126
        serde_json::Value::Number(n) => match n.as_i64() {
1127
            Some(i) => q.bind(i),
1128
            None => q.bind(n.as_f64().unwrap_or(0.0)),
1129
        },
1130
        serde_json::Value::String(s) if blob => q.bind(hex::decode(&s).unwrap_or_else(|_| s.into_bytes())),
1131
        serde_json::Value::String(s) => q.bind(s),
1132
        other => q.bind(other.to_string()),
1133
    }
1134
}
1135
1136
/// Restore one repository's records from a run into this hub, the
1137
/// repository being `repo` here and `old` in the backup (docs/backups.md
1138
/// §7): its issues, proposals, labels, milestones, releases and their
1139
/// assets, checks, Flow runs and the rest, each table after the tables it
1140
/// points at, every id mapped to its new one and every account, team and
1141
/// organisation found again by name. A row pointing at an account this
1142
/// hub lacks keeps a blank reference where it may, and is left out where
1143
/// it may not. What needs re-entering (webhook secrets, never backed up)
1144
/// and what was left out come back as notes.
1145
async fn restore_records(
1146
    app: &Shared,
1147
    dest: &std::sync::Arc<dyn Destination>,
1148
    rec: &serde_json::Value,
1149
    old: i64,
1150
    repo: i64,
1151
) -> Result<Vec<String>> {
1152
    let pool = &app.db;
1153
    let obj = rec.as_object().context("the records are not an object")?;
1154
    let mut by_name: HashMap<(String, i64), Option<i64>> = HashMap::new();
1155
    for (kind, m) in obj.get(NAMES).and_then(|n| n.as_object()).into_iter().flatten() {
1156
        for (id, name) in m.as_object().into_iter().flatten() {
1157
            let (Ok(id), Some(name)) = (id.parse::<i64>(), name.as_str()) else { continue };
1158
            by_name.insert((kind.clone(), id), id_by_name(pool, kind, name).await?);
1159
        }
1160
    }
1161
    let here: std::collections::HashSet<String> = tables(pool).await?.into_iter().collect();
1162
    let mut notes = Vec::new();
1163
    let mut todo: Vec<String> = Vec::new();
1164
    for t in obj.keys().filter(|t| t.as_str() != "repositories" && !NOT_RESTORED.contains(&t.as_str())) {
1165
        if here.contains(t) {
1166
            todo.push(t.clone());
1167
        } else {
1168
            notes.push(format!("{t}: not a table on this hub, left out"));
1169
        }
1170
    }
1171
    let mut fks: HashMap<String, Vec<(String, String, String)>> = HashMap::new();
1172
    for t in &todo {
1173
        fks.insert(t.clone(), fk_list(pool, t).await?);
1174
    }
1175
    // Each table after the tables of the records it points at.
1176
    let mut order = Vec::new();
1177
    while !todo.is_empty() {
1178
        let mut ready: Vec<String> =
1179
            todo.iter().filter(|t| fks[*t].iter().all(|(p, _, _)| p == *t || !todo.contains(p))).cloned().collect();
1180
        if ready.is_empty() {
1181
            ready.push(todo[0].clone());
1182
        }
1183
        todo.retain(|t| !ready.contains(t));
1184
        order.extend(ready);
1185
    }
1186
    let mut maps: HashMap<String, HashMap<i64, i64>> = HashMap::new();
1187
    maps.insert("repositories".into(), [(old, repo)].into());
1188
    let mut left_out: std::collections::BTreeMap<String, usize> = Default::default();
1189
    let mut tx = pool.begin().await?;
1190
    for table in &order {
1191
        let cols = table_cols(pool, table).await?;
1192
        let keys: Vec<&str> = cols.iter().filter(|c| c.pk > 0).map(|c| c.name.as_str()).collect();
1193
        let by_id = keys == ["id"] && cols.iter().any(|c| c.name == "id" && c.ty.eq_ignore_ascii_case("INTEGER"));
1194
        let verb = if keys == ["repo_id"] {
1195
            "INSERT OR REPLACE"
1196
        } else if by_id {
1197
            "INSERT"
1198
        } else {
1199
            "INSERT OR IGNORE"
1200
        };
1201
        let mut rows: Vec<&serde_json::Map<String, serde_json::Value>> =
1202
            obj[table].as_array().into_iter().flatten().filter_map(|r| r.as_object()).collect();
1203
        rows.sort_by_key(|r| r.get("id").and_then(|v| v.as_i64()).unwrap_or(0));
1204
        'row: for row in rows {
1205
            let (mut names, mut values) = (Vec::new(), Vec::new());
1206
            for c in &cols {
1207
                if by_id && c.name == "id" {
1208
                    continue;
1209
                }
1210
                let mut v = row.get(&c.name).cloned().filter(|v| !v.is_null());
1211
                if let Some((parent, _, _)) = fks[table].iter().find(|(_, from, _)| *from == c.name) {
1212
                    if let Some(was) = v.as_ref().and_then(|v| v.as_i64()) {
1213
                        let now = match parent.as_str() {
1214
                            "accounts" | "teams" | "organisations" => by_name.get(&(parent.clone(), was)).copied().flatten(),
1215
                            p => maps.get(p).and_then(|m| m.get(&was)).copied(),
1216
                        };
1217
                        if now.is_none() && c.notnull {
1218
                            *left_out.entry(table.clone()).or_default() += 1;
1219
                            continue 'row;
1220
                        }
1221
                        v = now.map(serde_json::Value::from);
1222
                    }
1223
                }
1224
                let blob = c.ty.to_ascii_uppercase().contains("BLOB");
1225
                match v {
1226
                    Some(v) => {
1227
                        names.push(c.name.clone());
1228
                        values.push((v, blob));
1229
                    }
1230
                    // Never backed up (a secret): blank here, to enter again.
1231
                    None if c.notnull && c.default.is_none() => {
1232
                        names.push(c.name.clone());
1233
                        let blank = if c.ty.to_ascii_uppercase().contains("INT") { serde_json::json!(0) } else { serde_json::json!("") };
1234
                        values.push((blank, false));
1235
                    }
1236
                    None => {}
1237
                }
1238
            }
1239
            let sql = format!(
1240
                "{verb} INTO \"{table}\" ({}) VALUES ({})",
1241
                names.iter().map(|n| format!("\"{n}\"")).collect::<Vec<_>>().join(", "),
1242
                vec!["?"; names.len()].join(", ")
1243
            );
1244
            let mut q = sqlx::query(&sql);
1245
            for (v, blob) in values {
1246
                q = bind_json(q, v, blob);
1247
            }
1248
            let done = q.execute(&mut *tx).await.with_context(|| format!("restore a row of {table}"))?;
1249
            if let (true, Some(was)) = (by_id, row.get("id").and_then(|v| v.as_i64())) {
1250
                maps.entry(table.clone()).or_default().insert(was, done.last_insert_rowid());
1251
            }
1252
        }
1253
    }
1254
    tx.commit().await?;
1255
    for (table, n) in left_out {
1256
        notes.push(format!("{table}: {n} row(s) left out: they name accounts or rows this hub lacks"));
1257
    }
1258
    // Search finds what came back.
1259
    for (kind, table) in [("issue", "issues"), ("proposal", "proposals")] {
1260
        let found: Vec<(i64, i64, String, String)> =
1261
            sqlx::query_as(&format!("SELECT id, number, title, body FROM {table} WHERE repo_id = ?")).bind(repo).fetch_all(pool).await?;
1262
        for (id, number, title, body) in found {
1263
            db::notify::index(pool, kind, repo, id, number, &title, &body).await?;
1264
        }
1265
    }
1266
    // Webhook secrets were never written: they are entered again.
1267
    let hooks: Vec<String> =
1268
        sqlx::query_scalar("SELECT url FROM hooks WHERE repo_id = ? AND secret = ''").bind(repo).fetch_all(pool).await?;
1269
    for url in hooks {
1270
        notes.push(format!("webhook {url}: enter its secret again (Settings › Webhooks)"));
1271
    }
1272
    // Release assets, each checked against its name.
1273
    for hash in obj.get("release_assets").and_then(|a| a.as_array()).into_iter().flatten().filter_map(|a| a["hash"].as_str()) {
1274
        if hash.len() != 64 || !hash.bytes().all(|b| b.is_ascii_hexdigit()) {
1275
            continue;
1276
        }
1277
        let path = crate::web::releases::blob_path(app, hash);
1278
        if path.exists() {
1279
            continue;
1280
        }
1281
        let got = tokio::task::block_in_place(|| dest.get(&blob_key(hash)));
1282
        match got {
1283
            Ok(bytes) if blake3::hash(&bytes).to_hex().as_str() == hash => {
1284
                std::fs::create_dir_all(path.parent().context("no parent")?)?;
1285
                std::fs::write(&path, &bytes)?;
1286
            }
1287
            Ok(_) => notes.push(format!("release asset {}: the copy does not match its hash", &hash[..12])),
1288
            Err(_) => notes.push(format!("release asset {}: not in the backup", &hash[..12])),
1289
        }
1290
    }
1291
    Ok(notes)
1292
}
1293
1294
// ---------------------------------------------------------------- a run
1295
1296
fn run_name(ms: i64) -> String {
1297
    let t = time::OffsetDateTime::from_unix_timestamp_nanos(ms as i128 * 1_000_000).unwrap_or(time::OffsetDateTime::UNIX_EPOCH);
1298
    format!("{:04}{:02}{:02}T{:02}{:02}{:02}Z", t.year(), u8::from(t.month()), t.day(), t.hour(), t.minute(), t.second())
1299
}
1300
1301
/// The last whole run's manifests, by `kind/name`.
1302
fn last_manifests(dest: &dyn Destination) -> HashMap<String, HashMap<String, File>> {
1303
    let mut out = HashMap::new();
1304
    let Ok(at) = dest.get("LATEST").map(|b| String::from_utf8_lossy(&b).trim().to_string()) else { return out };
1305
    for name in dest.list(&format!("runs/{at}/repos")).unwrap_or_default() {
1306
        if let Ok(m) = dest.get(&format!("runs/{at}/repos/{name}")).and_then(|b| Ok(serde_json::from_slice::<Manifest>(&b)?)) {
1307
            out.insert(format!("{}/{}", m.kind, m.name), m.files.into_iter().map(|f| (f.path.clone(), f)).collect());
1308
        }
1309
    }
1310
    out
1311
}
1312
1313
/// Back up `owner` now, as its setting says. The run is recorded whatever
1314
/// happens; a failure notifies the owner once until a run succeeds.
1315
pub async fn run(app: &Shared, owner: &str, trigger: &str) -> Result<Run> {
1316
    let s = setting(&app.db, owner).await?.with_context(|| format!("{owner} has no backup set up"))?;
1317
    let started = db::now_ms();
1318
    let at = run_name(started);
1319
    let id: i64 =
1320
        sqlx::query_scalar("INSERT INTO backup_runs (owner, started_ms, status, at, trigger) VALUES (?, ?, 'running', ?, ?) RETURNING id")
1321
            .bind(owner)
1322
            .bind(started)
1323
            .bind(&at)
1324
            .bind(trigger)
1325
            .fetch_one(&app.db)
1326
            .await?;
1327
    let outcome = run_into(app, &s, &at, trigger == "now").await;
1328
    let (status, detail, sent) = match outcome {
1329
        Ok((sent, problems)) if problems.is_empty() => ("ok", String::new(), sent),
1330
        Ok((sent, problems)) => ("partial", problems.join("; "), sent),
1331
        Err(e) => ("failed", format!("{e:#}"), (0, Sent::default())),
1332
    };
1333
    sqlx::query("UPDATE backup_runs SET finished_ms = ?, status = ?, detail = ?, repos = ?, files = ?, sent = ?, bytes = ? WHERE id = ?")
1334
        .bind(db::now_ms())
1335
        .bind(status)
1336
        .bind(&detail)
1337
        .bind(sent.0)
1338
        .bind(sent.1.files)
1339
        .bind(sent.1.sent)
1340
        .bind(sent.1.bytes)
1341
        .bind(id)
1342
        .execute(&app.db)
1343
        .await?;
1344
    if status != "ok" {
1345
        notify_failure(app, &s, status, &detail).await;
1346
    }
1347
    let r: Run = sqlx::query_as("SELECT * FROM backup_runs WHERE id = ?").bind(id).fetch_one(&app.db).await?;
1348
    Ok(r)
1349
}
1350
1351
async fn run_into(app: &Shared, s: &Setting, at: &str, by_hand: bool) -> Result<((i64, Sent), Vec<String>)> {
1352
    let dest: std::sync::Arc<dyn Destination> = destination(app, s)?.into();
1353
    let known = {
1354
        let d = dest.clone();
1355
        tokio::task::spawn_blocking(move || last_manifests(d.as_ref())).await?
1356
    };
1357
    // Its repositories, but those left out (§5).
1358
    let mine: Vec<db::repos::RepoRow> =
1359
        db::repos::list(&app.db).await?.into_iter().filter(|r| r.owner == s.owner && included(s, r, by_hand)).collect();
1360
    let mut problems = Vec::new();
1361
    let mut sent = Sent::default();
1362
    let mut repos = 0i64;
1363
    for row in &mine {
1364
        let Some(name) = RepoName::new(&row.owner, &row.name) else { continue };
1365
        for (kind, store) in [("repos", &app.repos), ("wikis", &app.wikis)] {
1366
            if !store.exists(&name) {
1367
                continue;
1368
            }
1369
            let dir = store.dir(&name);
1370
            let k = known.get(&format!("{kind}/{}", row.name)).cloned().unwrap_or_default();
1371
            // The large, unchanging files without the lock, then the whole
1372
            // walk again under the read lock: pushes wait for that moment.
1373
            let (d1, dir1, k1) = (dest.clone(), dir.clone(), k.clone());
1374
            let first = tokio::task::spawn_blocking(move || {
1375
                let mut s = Sent::default();
1376
                copy_dir(d1.as_ref(), &dir1, &k1, true, &mut s).map(|f| (f, s))
1377
            })
1378
            .await?;
1379
            let mut k2 = k;
1380
            if let Ok((files, _)) = &first {
1381
                for f in files {
1382
                    k2.insert(f.path.clone(), f.clone());
1383
                }
1384
            }
1385
            let guard = store.read_guard(&name).await;
1386
            let (d2, dir2) = (dest.clone(), dir.clone());
1387
            let second = tokio::task::spawn_blocking(move || {
1388
                let _guard = guard;
1389
                let mut s = Sent::default();
1390
                copy_dir(d2.as_ref(), &dir2, &k2, false, &mut s).map(|f| (f, s))
1391
            })
1392
            .await?;
1393
            match second {
1394
                Ok((files, s2)) => {
1395
                    let s1 = first.map(|(_, s)| s).unwrap_or_default();
1396
                    sent.files += s2.files;
1397
                    sent.sent += s1.sent + s2.sent;
1398
                    sent.bytes += s1.bytes + s2.bytes;
1399
                    let m = Manifest { owner: row.owner.clone(), name: row.name.clone(), kind: kind.into(), files };
1400
                    let key = format!("runs/{at}/repos/{kind}-{}.json", row.name);
1401
                    tokio::task::block_in_place(|| dest.put(&key, &serde_json::to_vec(&m)?))?;
1402
                    if kind == "repos" {
1403
                        repos += 1;
1404
                    }
1405
                }
1406
                Err(e) => problems.push(format!("{}/{} ({kind}): {e:#}", row.owner, row.name)),
1407
            }
1408
        }
1409
        let rec = async {
1410
            let mut r = records(&app.db, "repo_id", row.id).await?;
1411
            // The repository's own row, which restore recreates first, and
1412
            // the names behind the accounts, teams and organisations its
1413
            // rows point at, which another hub finds them by.
1414
            let own = rows(&app.db, "repositories", &format!("id = {}", row.id)).await?;
1415
            let names = names_of(&app.db, &r).await?;
1416
            if let Some(m) = r.as_object_mut() {
1417
                m.insert("repositories".into(), own);
1418
                m.insert(NAMES.into(), names);
1419
            }
1420
            anyhow::Ok(r)
1421
        };
1422
        match rec.await {
1423
            Ok(r) => {
1424
                tokio::task::block_in_place(|| dest.put(&format!("runs/{at}/records/{}.json", row.name), &serde_json::to_vec(&r)?))?;
1425
                // Release assets live beside the stores, named by their
1426
                // BLAKE3: each sent once, as the stores' files are.
1427
                for hash in r["release_assets"].as_array().into_iter().flatten().filter_map(|a| a["hash"].as_str()) {
1428
                    if hash.len() != 64 || !hash.bytes().all(|b| b.is_ascii_hexdigit()) {
1429
                        continue;
1430
                    }
1431
                    let path = crate::web::releases::blob_path(app, hash);
1432
                    let key = blob_key(hash);
1433
                    let put = tokio::task::block_in_place(|| -> Result<Option<i64>> {
1434
                        if dest.exists(&key)? {
1435
                            return Ok(None);
1436
                        }
1437
                        let bytes = std::fs::read(&path).with_context(|| format!("read release asset {}", &hash[..12]))?;
1438
                        dest.put(&key, &bytes)?;
1439
                        Ok(Some(bytes.len() as i64))
1440
                    });
1441
                    match put {
1442
                        Ok(Some(n)) => {
1443
                            sent.files += 1;
1444
                            sent.sent += 1;
1445
                            sent.bytes += n;
1446
                        }
1447
                        Ok(None) => sent.files += 1,
1448
                        Err(e) => problems.push(format!("{}/{} release asset: {e:#}", row.owner, row.name)),
1449
                    }
1450
                }
1451
            }
1452
            Err(e) => problems.push(format!("{}/{} records: {e:#}", row.owner, row.name)),
1453
        }
1454
    }
1455
    // The owner itself.
1456
    let owner_rows = if s.kind == "org" {
1457
        match db::accounts::org_by_name(&app.db, &s.owner).await? {
1458
            Some(o) => records(&app.db, "org_id", o.id).await?,
1459
            None => serde_json::json!({}),
1460
        }
1461
    } else {
1462
        match db::accounts::account_by_name(&app.db, &s.owner).await? {
1463
            Some(a) => records(&app.db, "account_id", a.id).await?,
1464
            None => serde_json::json!({}),
1465
        }
1466
    };
1467
    // The blocking calls of a destination (GCS drives its client on this
1468
    // runtime) run where blocking is allowed.
1469
    tokio::task::block_in_place(|| -> Result<()> {
1470
        dest.put(&format!("runs/{at}/owner.json"), &serde_json::to_vec(&owner_rows)?)?;
1471
        dest.put(&format!("runs/{at}/DONE"), at.as_bytes())?;
1472
        dest.put("LATEST", at.as_bytes())?;
1473
        prune(dest.as_ref(), s.keep_days)
1474
    })?;
1475
    Ok(((repos, sent), problems))
1476
}
1477
1478
/// Drop whole runs older than `keep_days`, and the blobs no kept run
1479
/// names. The newest whole run is always kept.
1480
pub fn prune(dest: &dyn Destination, keep_days: i64) -> Result<()> {
1481
    let cutoff = run_name(db::now_ms() - keep_days.max(1) * 86_400_000);
1482
    let runs = dest.list("runs")?;
1483
    let whole: Vec<&String> = runs.iter().filter(|r| dest.exists(&format!("runs/{r}/DONE")).unwrap_or(false)).collect();
1484
    let newest = whole.last().map(|s| s.to_string());
1485
    let mut removed = false;
1486
    for r in &runs {
1487
        if r.as_str() < cutoff.as_str() && Some(r.to_string()) != newest {
1488
            dest.remove(&format!("runs/{r}"))?;
1489
            removed = true;
1490
        }
1491
    }
1492
    if !removed {
1493
        return Ok(());
1494
    }
1495
    let mut wanted = std::collections::HashSet::new();
1496
    for r in dest.list("runs")? {
1497
        for m in dest.list(&format!("runs/{r}/repos"))? {
1498
            if let Ok(m) = dest.get(&format!("runs/{r}/repos/{m}")).and_then(|b| Ok(serde_json::from_slice::<Manifest>(&b)?)) {
1499
                wanted.extend(m.files.into_iter().map(|f| f.hash));
1500
            }
1501
        }
1502
    }
1503
    for shard in dest.list("blobs")? {
1504
        for b in dest.list(&format!("blobs/{shard}"))? {
1505
            if !wanted.contains(&b) {
1506
                dest.remove(&format!("blobs/{shard}/{b}"))?;
1507
            }
1508
        }
1509
    }
1510
    Ok(())
1511
}
1512
1513
async fn notify_failure(app: &Shared, s: &Setting, status: &str, detail: &str) {
1514
    // Once per spell of failures: not when the run before failed too.
1515
    let before: Option<String> =
1516
        sqlx::query_scalar("SELECT status FROM backup_runs WHERE owner = ? AND finished_ms IS NOT NULL ORDER BY id DESC LIMIT 1 OFFSET 1")
1517
            .bind(&s.owner)
1518
            .fetch_optional(&app.db)
1519
            .await
1520
            .ok()
1521
            .flatten();
1522
    if before.as_deref().is_some_and(|b| b != "ok") {
1523
        return;
1524
    }
1525
    let (who, url): (Vec<i64>, String) = if s.kind == "org" {
1526
        match db::accounts::org_by_name(&app.db, &s.owner).await.ok().flatten() {
1527
            Some(o) => (
1528
                db::accounts::org_members(&app.db, o.id)
1529
                    .await
1530
                    .unwrap_or_default()
1531
                    .into_iter()
1532
                    .filter(|(_, r)| r == "owner")
1533
                    .map(|(a, _)| a.id)
1534
                    .collect(),
1535
                format!("/orgs/{}/settings#backups", s.owner),
1536
            ),
1537
            None => return,
1538
        }
1539
    } else {
1540
        match db::accounts::account_by_name(&app.db, &s.owner).await.ok().flatten() {
1541
            Some(a) => (vec![a.id], "/settings/backups".into()),
1542
            None => return,
1543
        }
1544
    };
1545
    let title = format!("Backup of {} {status}: {}", s.owner, detail.chars().take(160).collect::<String>());
1546
    for id in who {
1547
        let _ = db::notify::add(
1548
            &app.db,
1549
            db::notify::New {
1550
                account_id: id,
1551
                repo_id: None,
1552
                reason: db::notify::Reason::Participating,
1553
                kind: "backup",
1554
                title: &title,
1555
                url: &url,
1556
                actor: "",
1557
            },
1558
        )
1559
        .await;
1560
    }
1561
}
1562
1563
/// Run each owner's backup when it is due, one at a time. A run left
1564
/// "running" by a restart is marked failed first.
1565
pub async fn scheduler(app: Shared) {
1566
    let _ = sqlx::query(
1567
        "UPDATE backup_runs SET status = 'failed', detail = 'the hub restarted during the run', finished_ms = ? WHERE status = 'running'",
1568
    )
1569
    .bind(db::now_ms())
1570
    .execute(&app.db)
1571
    .await;
1572
    loop {
1573
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
1574
        for s in due(&app.db).await.unwrap_or_default() {
1575
            if let Err(e) = run(&app, &s.owner, "schedule").await {
1576
                tracing::warn!("backup of {}: {e:#}", s.owner);
1577
            }
1578
        }
1579
    }
1580
}
1581
1582
/// The owners whose next run is due now: by their own schedule, or by
1583
/// the shortest schedule one of their repositories chose (§5). A run takes
1584
/// every included repository and sends only what changed, so each run
1585
/// stays whole and a restore point has everything.
1586
pub async fn due(pool: &Pool) -> sqlx::Result<Vec<Setting>> {
1587
    let all: Vec<Setting> = sqlx::query_as("SELECT * FROM backup_settings WHERE updated_ms > 0").fetch_all(pool).await?;
1588
    let repos = db::repos::list(pool).await?;
1589
    let now = db::now_ms();
1590
    let mut out = Vec::new();
1591
    for s in all {
1592
        let theirs: Vec<db::repos::RepoRow> = repos.iter().filter(|r| r.owner == s.owner).cloned().collect();
1593
        let Some(period) = period_with(&s, &theirs) else { continue };
1594
        let last: Option<i64> =
1595
            sqlx::query_scalar("SELECT MAX(started_ms) FROM backup_runs WHERE owner = ?").bind(&s.owner).fetch_one(pool).await?;
1596
        if last.is_none_or(|l| now - l >= period * 1000) {
1597
            out.push(s);
1598
        }
1599
    }
1600
    Ok(out)
1601
}
1602
1603
/// Restore every repository of the whole run `at` (the latest when none)
1604
/// from `dest` into `repos_root/OWNER/NAME` and `wikis_root/…`, refusing to
1605
/// overwrite one that exists. Each file is checked against its hash.
1606
pub fn restore(dest: &dyn Destination, at: Option<&str>, repos_root: &Path, wikis_root: &Path) -> Result<Vec<String>> {
1607
    let at = match at {
1608
        Some(a) => a.to_string(),
1609
        None => String::from_utf8_lossy(&dest.get("LATEST").context("no whole run at this destination")?).trim().to_string(),
1610
    };
1611
    if !dest.exists(&format!("runs/{at}/DONE"))? {
1612
        bail!("run {at} is not whole (no DONE): pick another with --run");
1613
    }
1614
    let mut done = Vec::new();
1615
    for name in dest.list(&format!("runs/{at}/repos"))? {
1616
        let m: Manifest = serde_json::from_slice(&dest.get(&format!("runs/{at}/repos/{name}"))?)?;
1617
        let root = if m.kind == "wikis" { wikis_root } else { repos_root };
1618
        let dir = root.join(&m.owner).join(&m.name);
1619
        if dir.exists() {
1620
            bail!("{} exists: restore into an empty hub, or move it aside", dir.display());
1621
        }
1622
        for f in &m.files {
1623
            let bytes = dest.get(&blob_key(&f.hash))?;
1624
            if blake3::hash(&bytes).to_hex().as_str() != f.hash {
1625
                bail!("{}/{} {}: the copy does not match its hash", m.owner, m.name, f.path);
1626
            }
1627
            let p = dir.join(&f.path);
1628
            std::fs::create_dir_all(p.parent().context("no parent")?)?;
1629
            std::fs::write(&p, &bytes)?;
1630
        }
1631
        done.push(format!("{}/{} ({})", m.owner, m.name, m.kind));
1632
    }
1633
    Ok(done)
1634
}
1635
1636
/// Restore the whole run `at` (the latest when none) into this hub: each
1637
/// repository's and wiki's store byte for byte, then its row and settings,
1638
/// then its records (issues, proposals, releases and the rest), so it shows
1639
/// on the hub as it was (docs/backups.md §7). What was restored, then any
1640
/// notes (`note: …`) on what needs entering again or was left out.
1641
pub async fn restore_into_hub(app: &Shared, dest: std::sync::Arc<dyn Destination>, at: Option<String>) -> Result<Vec<String>> {
1642
    let (repos_root, wikis_root) = (app.repos.root().to_path_buf(), app.wikis.root().to_path_buf());
1643
    let (d, a) = (dest.clone(), at.clone());
1644
    let done = tokio::task::spawn_blocking(move || restore(d.as_ref(), a.as_deref(), &repos_root, &wikis_root)).await??;
1645
    let at = match at {
1646
        Some(a) => a,
1647
        None => String::from_utf8_lossy(&tokio::task::block_in_place(|| dest.get("LATEST"))?).trim().to_string(),
1648
    };
1649
    let mut notes = Vec::new();
1650
    for name in tokio::task::block_in_place(|| dest.list(&format!("runs/{at}/records")))? {
1651
        let rec: serde_json::Value =
1652
            serde_json::from_slice(&tokio::task::block_in_place(|| dest.get(&format!("runs/{at}/records/{name}")))?)?;
1653
        let Some(row) = rec["repositories"].as_array().and_then(|a| a.first()) else { continue };
1654
        let (owner, repo) = (row["owner"].as_str().unwrap_or_default(), row["name"].as_str().unwrap_or_default());
1655
        if owner.is_empty() || db::repos::get(&app.db, owner, repo).await?.is_some() {
1656
            continue;
1657
        }
1658
        let visibility = match row["visibility"].as_str() {
1659
            Some("private") => hub_api::Visibility::Private,
1660
            Some("internal") => hub_api::Visibility::Internal,
1661
            _ => hub_api::Visibility::Public,
1662
        };
1663
        let made = db::repos::create(&app.db, owner, repo, row["description"].as_str().unwrap_or_default(), visibility).await?;
1664
        // Its settings as they were: every column the row kept but its
1665
        // name, its id and a fork's parent (another hub's id).
1666
        let cols = table_cols(&app.db, "repositories").await?;
1667
        let set: Vec<(&Col, serde_json::Value)> = cols
1668
            .iter()
1669
            .filter(|c| !["id", "owner", "name", "forked_from"].contains(&c.name.as_str()))
1670
            .filter_map(|c| row.get(&c.name).filter(|v| !v.is_null()).map(|v| (c, v.clone())))
1671
            .collect();
1672
        if !set.is_empty() {
1673
            let sql = format!(
1674
                "UPDATE repositories SET {} WHERE id = ?",
1675
                set.iter().map(|(c, _)| format!("\"{}\" = ?", c.name)).collect::<Vec<_>>().join(", ")
1676
            );
1677
            let mut q = sqlx::query(&sql);
1678
            for (c, v) in set {
1679
                q = bind_json(q, v, c.ty.to_ascii_uppercase().contains("BLOB"));
1680
            }
1681
            q.bind(made.id).execute(&app.db).await?;
1682
        }
1683
        let old = row["id"].as_i64().unwrap_or(-1);
1684
        for n in restore_records(app, &dest, &rec, old, made.id).await.with_context(|| format!("restore {owner}/{repo}'s records"))? {
1685
            notes.push(format!("note: {owner}/{repo}: {n}"));
1686
        }
1687
    }
1688
    let mut done = done;
1689
    done.extend(notes);
1690
    Ok(done)
1691
}
1692
1693
#[cfg(test)]
1694
mod tests {
1695
    use super::*;
1696
1697
    #[test]
1698
    fn a_directory_destination_keeps_keys_inside_its_root() {
1699
        let d = tempfile::tempdir().unwrap();
1700
        let dest = Directory { root: d.path().to_path_buf() };
1701
        dest.put("runs/a/DONE", b"a").unwrap();
1702
        assert_eq!(dest.get("runs/a/DONE").unwrap(), b"a");
1703
        assert_eq!(dest.list("runs").unwrap(), vec!["a"]);
1704
        assert!(dest.put("../escape", b"x").is_err() && dest.put("/abs", b"x").is_err());
1705
        dest.remove("runs").unwrap();
1706
        assert!(dest.list("runs").unwrap().is_empty());
1707
    }
1708
1709
    #[test]
1710
    fn files_are_sent_once_and_unchanged_ones_not_read_again() {
1711
        let src = tempfile::tempdir().unwrap();
1712
        let out = tempfile::tempdir().unwrap();
1713
        let dest = Directory { root: out.path().to_path_buf() };
1714
        std::fs::create_dir_all(src.path().join(".oxy/objects/loose")).unwrap();
1715
        std::fs::write(src.path().join(".oxy/objects/loose/aa"), b"object").unwrap();
1716
        std::fs::write(src.path().join(".oxy/ops-head"), b"ref one").unwrap();
1717
        let mut s = Sent::default();
1718
        let files = copy_dir(&dest, src.path(), &HashMap::new(), false, &mut s).unwrap();
1719
        assert_eq!((files.len(), s.sent), (2, 2));
1720
        let known: HashMap<String, File> = files.iter().map(|f| (f.path.clone(), f.clone())).collect();
1721
        let mut s = Sent::default();
1722
        let again = copy_dir(&dest, src.path(), &known, false, &mut s).unwrap();
1723
        assert_eq!((again, s.sent), (files.clone(), 0), "nothing new travels");
1724
        // The first pass takes only the files that never change.
1725
        let mut s = Sent::default();
1726
        let first = copy_dir(&dest, src.path(), &HashMap::new(), true, &mut s).unwrap();
1727
        assert_eq!(first.iter().map(|f| f.path.as_str()).collect::<Vec<_>>(), vec![".oxy/objects/loose/aa"]);
1728
    }
1729
1730
    #[test]
1731
    fn requests_are_signed_as_aws_documents() {
1732
        // AWS's own example (S3 API reference, "Signature calculations":
1733
        // GET Object), with its published signature.
1734
        let empty = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
1735
        let headers = vec![
1736
            ("Host".to_string(), "examplebucket.s3.amazonaws.com".to_string()),
1737
            ("Range".to_string(), "bytes=0-9".to_string()),
1738
            ("x-amz-content-sha256".to_string(), empty.to_string()),
1739
            ("x-amz-date".to_string(), "20130524T000000Z".to_string()),
1740
        ];
1741
        let auth = sigv4(
1742
            "GET",
1743
            "/test.txt",
1744
            &[],
1745
            &headers,
1746
            empty,
1747
            ("AKIAIOSFODNN7EXAMPLE", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"),
1748
            "us-east-1",
1749
            "s3",
1750
            "20130524T000000Z",
1751
        );
1752
        assert_eq!(
1753
            auth,
1754
            "AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request, \
1755
             SignedHeaders=host;range;x-amz-content-sha256;x-amz-date, \
1756
             Signature=f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"
1757
        );
1758
        assert_eq!(aws_encode("runs/2026 x+y", true), "runs/2026%20x%2By");
1759
        assert_eq!(aws_encode("a/b", false), "a%2Fb");
1760
        assert_eq!(xml_text("<A><Key>runs/x&amp;y</Key></A>", "Key").as_deref(), Some("runs/x&y"));
1761
    }
1762
1763
    #[test]
1764
    fn run_names_sort_by_time() {
1765
        assert_eq!(run_name(0), "19700101T000000Z");
1766
        assert!(run_name(1_791_900_000_000) < run_name(1_791_903_600_000));
1767
    }
1768
}