fix: files refactoring

docker-compose.yml

@@ -24,7 +24,7 @@ services:
       LOG: debug
       TZ: "Europe/Paris"
       #      TMPDIR: /scratch
-      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiODBkZTY1NDktNzZjYS00NGZhLWIyNTItMTY0MWI2YTQzZjJkIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
+      EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNGUyMzgxYzgtNzI4Mi00MzIxLTk1ZjQtYTU3NTRiZTc2YmFlIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ=="
       #CHUNK_SIZE_MB: "1"
       #POOLING: 1
       #RETRY_ATTEMPTS: 3

justfile

@@ -79,4 +79,8 @@ test:
     docker compose -f docker-compose.test.yml down --volumes
 
 seed-files:
-    for d in archive snapshot sync; do mkdir -p test-files/$d && echo one > test-files/$d/a.txt && echo two > test-files/$d/b.txt; done
+    for d in archive snapshot sync; do mkdir -p test-files/$d && echo one > test-files/$d/a.txt && echo two > test-files/$d/b.txt; done
+
+seed-files-1gb:
+    echo "Seeding files sources (1GB each)..."
+    bash ./scripts/files/seed-1gb.sh

scripts/README.md

@@ -11,6 +11,7 @@ make seed-postgres-1gb
 make seed-all   
 make seed-sqlite   
 make seed-sqlite SEED=big   
+just seed-files-1gb
 ```
 
 ## Verify commands 

src/domain/files/backup.rs

@@ -1,4 +1,4 @@
-use super::matcher::ExcludeMatcher;
+use super::matcher::{ExcludeMatcher, one_file_system, source_root};
 use super::platform::dev;
 use crate::services::backup::logger::JobLogger;
 use crate::services::config::DatabaseConfig;
@@ -20,20 +20,11 @@ pub async fn run(
 }
 
 fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Result<PathBuf> {
-    // Canonical paths so the output location can be recognised while walking.
-    let root = std::fs::canonicalize(&cfg.path)
-        .with_context(|| format!("cannot read source directory {}", cfg.path))?;
-    if !root.is_dir() {
-        bail!("source path {} is not a directory", root.display());
-    }
+    let root = source_root(cfg)?;
     let backup_real = std::fs::canonicalize(backup_dir)
         .with_context(|| format!("cannot read backup directory {}", backup_dir.display()))?;
     let matcher = ExcludeMatcher::from_config(cfg)?;
-    let one_file_system = cfg
-        .options
-        .get("one_file_system")
-        .and_then(|v| v.as_bool())
-        .unwrap_or(false);
+    let one_file_system = one_file_system(cfg);
     let root_dev = dev(&std::fs::metadata(&root)
         .with_context(|| format!("cannot read source directory {}", root.display()))?);
 
@@ -46,7 +37,7 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul
 
     logger.log("info", format!("Archiving {}", root.display()));
     let mut files = 0u64;
-    // Entries that could not be read: a mirror restore of this archive would delete them.
+
     let mut unreadable = 0usize;
     let mut walker = WalkDir::new(&root)
         .follow_links(false)
@@ -65,7 +56,6 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul
         let rel = entry.path().strip_prefix(&root)?.to_path_buf();
         let ft = entry.file_type();
 
-        // Never archive our own output (backup dir under the source, or the archive itself).
         let mut skip =
             matcher.is_excluded(&rel) || entry.path() == backup_real || entry.path() == out_real;
         if !skip && one_file_system && ft.is_dir() {
@@ -86,8 +76,6 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul
         }
 
         if ft.is_file() {
-            // Only a failure to open is skippable: it happens before anything is
-            // written. Every later error leaves the tar stream misaligned and must abort.
             let mut f = match File::open(entry.path()) {
                 Ok(f) => f,
                 Err(e) => {
@@ -100,20 +88,18 @@ fn archive(cfg: &DatabaseConfig, backup_dir: &Path, logger: &JobLogger) -> Resul
             let size = meta.len();
             let mut header = tar::Header::new_gnu();
             header.set_metadata(&meta);
-            // The header size is fixed up front: cap the copy if the file grew and
-            // zero-pad if it shrank, so the entry always matches its header.
+
             let reader = (&mut f).take(size).chain(std::io::repeat(0)).take(size);
             tar.append_data(&mut header, &rel, reader)
                 .with_context(|| format!("failed to archive {}", rel.display()))?;
             files += 1;
         } else if ft.is_dir() || ft.is_symlink() {
-            // These fail on lstat/readlink before the header is written.
+
             if let Err(e) = tar.append_path_with_name(entry.path(), &rel) {
                 logger.log("warn", format!("Skipping {}: {e}", rel.display()));
                 unreadable += 1;
             }
         } else {
-            // Sockets, fifos and devices hold no data to restore; restore keeps them too.
             logger.log("warn", format!("Skipping special file {}", rel.display()));
         }
     }

src/domain/files/matcher.rs

@@ -1,10 +1,8 @@
 use crate::services::config::DatabaseConfig;
-use anyhow::{Context, Result};
+use anyhow::{Context, Result, bail};
 use globset::{GlobBuilder, GlobSet, GlobSetBuilder};
 use std::path::{Path, PathBuf};
 
-/// Exclude patterns, matched against paths relative to the source root.
-/// No leading `/` matches at any depth; a leading `/` anchors at the root.
 pub struct ExcludeMatcher {
     set: GlobSet,
     literal: Vec<PathBuf>,
@@ -39,8 +37,6 @@ impl ExcludeMatcher {
         Self::new(&exclude_patterns(cfg))
     }
 
-    /// Also exclude exactly this root-relative path (and its subtree), compared
-    /// literally so glob characters in the name mean nothing.
     pub fn exclude_path(&mut self, relative: PathBuf) {
         self.literal.push(relative);
     }
@@ -50,7 +46,6 @@ impl ExcludeMatcher {
     }
 }
 
-/// The `options.exclude` patterns of a files source, as configured.
 pub fn exclude_patterns(cfg: &DatabaseConfig) -> Vec<String> {
     cfg.options
         .get("exclude")
@@ -64,10 +59,40 @@ pub fn exclude_patterns(cfg: &DatabaseConfig) -> Vec<String> {
         .unwrap_or_default()
 }
 
-/// `options.one_file_system` of a files source.
 pub fn one_file_system(cfg: &DatabaseConfig) -> bool {
     cfg.options
         .get("one_file_system")
         .and_then(|v| v.as_bool())
         .unwrap_or(false)
 }
+
+pub fn source_root(cfg: &DatabaseConfig) -> Result<PathBuf> {
+    let root = std::fs::canonicalize(&cfg.path)
+        .with_context(|| format!("cannot read source directory {}", cfg.path))?;
+    if !root.is_dir() {
+        bail!("source path {} is not a directory", root.display());
+    }
+    Ok(root)
+}
+
+pub(crate) fn bracket_negation(pattern: &str) -> String {
+    let mut out = String::with_capacity(pattern.len());
+    let mut chars = pattern.chars();
+    let mut in_class = false;
+    while let Some(c) = chars.next() {
+        out.push(c);
+        match c {
+            '\\' => out.extend(chars.next()),
+            '[' if !in_class => {
+                in_class = true;
+                if chars.as_str().starts_with('!') {
+                    chars.next();
+                    out.push('^');
+                }
+            }
+            ']' => in_class = false,
+            _ => {}
+        }
+    }
+    out
+}

src/domain/files/mod.rs

@@ -3,4 +3,6 @@ pub mod database;
 pub mod matcher;
 pub mod ping;
 pub mod platform;
+pub mod restic;
 pub mod restore;
+pub mod sync;

src/domain/files/ping.rs

@@ -4,7 +4,6 @@ use std::path::Path;
 use tokio::time::{Duration, timeout};
 
 pub async fn run(cfg: DatabaseConfig) -> Result<bool> {
-    // A hung mount (NFS) blocks is_dir/read_dir: keep it off the runtime and bounded.
     let check = tokio::task::spawn_blocking(move || {
         let root = Path::new(&cfg.path);
         root.is_dir() && std::fs::read_dir(root).is_ok()

src/domain/files/platform.rs

@@ -14,7 +14,6 @@ pub fn dev(_meta: &std::fs::Metadata) -> u64 {
     0
 }
 
-/// Whether the agent can restore file ownership.
 #[cfg(unix)]
 pub fn is_root() -> bool {
     unsafe { libc::geteuid() == 0 }

src/domain/files/restic/backup.rs

@@ -0,0 +1,106 @@
+use super::excludes::backup_excludes;
+use super::json::{BackupSummary, Snapshot, files_removed};
+use super::repo::{ResticRepo, short};
+use crate::domain::files::matcher::{exclude_patterns, one_file_system, source_root};
+use crate::services::backup::logger::JobLogger;
+use crate::services::config::DatabaseConfig;
+use anyhow::{Context, Result, bail};
+
+pub async fn snapshot(
+    repo: &ResticRepo,
+    cfg: &DatabaseConfig,
+    backup_storage_id: &str,
+    logger: &JobLogger,
+) -> Result<BackupSummary> {
+    let root = source_root(cfg)?;
+    repo.ensure_initialized(logger).await?;
+
+    let host = format!("portabase-{}", cfg.generated_id);
+    let mut args: Vec<String> = vec![
+        "backup".into(),
+        root.to_string_lossy().into_owned(),
+        "--json".into(),
+        "--host".into(),
+        host.clone(),
+        "--tag".into(),
+        "portabase".into(),
+        "--tag".into(),
+        format!("bs:{backup_storage_id}"),
+    ];
+    for pattern in backup_excludes(&root, &exclude_patterns(cfg))? {
+        args.push("--exclude".into());
+        args.push(pattern);
+    }
+    if one_file_system(cfg) {
+        args.push("--one-file-system".into());
+    }
+
+    logger.log("info", format!("Snapshotting {} into {}", root.display(), repo.repository()));
+    let run = repo.run(&args, logger).await?;
+    for error in &run.errors {
+        logger.log("warn", format!("Unreadable: {error}"));
+    }
+    let summary: Option<BackupSummary> = run
+        .summary
+        .clone()
+        .map(serde_json::from_value)
+        .transpose()
+        .context("unexpected restic backup summary")?;
+
+    match (run.code, summary) {
+        (0, Some(mut s)) => {
+            s.files_removed = match removed(repo, &host, &s, logger).await {
+                Ok(removed) => removed,
+                Err(e) => {
+                    logger.log("warn", format!("Could not count removed files: {e:#}"));
+                    None
+                }
+            };
+            let removed = s.files_removed.map(|n| format!(", {n} removed")).unwrap_or_default();
+            logger.log(
+                "info",
+                format!(
+                    "Snapshot {}: {} new, {} changed, {} unmodified{removed} file(s); {} byte(s) scanned, {} byte(s) of new data",
+                    short(&s.snapshot_id), s.files_new, s.files_changed, s.files_unmodified, s.total_bytes_processed, s.data_added_packed
+                ),
+            );
+            Ok(s)
+        }
+        (0, None) => bail!("restic backup returned no summary"),
+        (3, summary) => {
+            let mut fate = "snapshot discarded";
+            if let Some(s) = summary {
+                if repo.is_append_only() {
+                    logger.log(
+                        "warn",
+                        format!("Incomplete snapshot {} left for the dashboard to forget", short(&s.snapshot_id)),
+                    );
+                    fate = "snapshot not restorable (the dashboard forgets it)";
+                } else {
+                    let forget = repo.run(["forget", s.snapshot_id.as_str()], logger).await?;
+                    if forget.code != 0 {
+                        logger.log(
+                            "error",
+                            format!("Could not forget incomplete snapshot {}: {}", s.snapshot_id, forget.error("forget")),
+                        );
+                    }
+                }
+            }
+            bail!(
+                "{} file(s) could not be read; {fate} so a mirror restore cannot delete them (fix permissions or add an exclude)",
+                run.errors.len()
+            )
+        }
+        _ => Err(run.error("backup")),
+    }
+}
+
+async fn removed(repo: &ResticRepo, host: &str, s: &BackupSummary, logger: &JobLogger) -> Result<Option<u64>> {
+    let listed = repo.run(["snapshots", "--json", "--host", host], logger).await?;
+    if listed.code != 0 {
+        return Err(listed.error("snapshots"));
+    }
+    let snapshots: Vec<Snapshot> =
+        serde_json::from_str(listed.stdout.trim()).context("unexpected `restic snapshots` output")?;
+    Ok(files_removed(&snapshots, &s.snapshot_id, s.files_changed, s.files_unmodified))
+}

src/domain/files/restic/excludes.rs

@@ -1,7 +1,7 @@
+use crate::domain::files::matcher::bracket_negation;
 use anyhow::{Result, bail};
 use std::path::Path;
 
-/// Escapes restic/filepath.Match metacharacters so a literal path matches only itself.
 pub fn glob_escape(literal: &str) -> String {
     let mut out = String::with_capacity(literal.len());
     for c in literal.chars() {
@@ -13,31 +13,6 @@ pub fn glob_escape(literal: &str) -> String {
     out
 }
 
-/// globset (the P0 matcher) reads `[!x]` as bracket negation; restic (Go
-/// `filepath.Match`) and rclone only know `[^x]` and take `!` literally.
-pub(crate) fn bracket_negation(pattern: &str) -> String {
-    let mut out = String::with_capacity(pattern.len());
-    let mut chars = pattern.chars();
-    let mut in_class = false;
-    while let Some(c) = chars.next() {
-        out.push(c);
-        match c {
-            '\\' => out.extend(chars.next()),
-            '[' if !in_class => {
-                in_class = true;
-                if chars.as_str().starts_with('!') {
-                    chars.next();
-                    out.push('^');
-                }
-            }
-            ']' => in_class = false,
-            _ => {}
-        }
-    }
-    out
-}
-
-/// P0 patterns, trimmed like the P0 matcher, as `(anchored, pattern)`; a leading `/` marks anchored ones.
 fn normalized(patterns: &[String]) -> Result<Vec<(bool, String)>> {
     let mut out = Vec::new();
     for raw in patterns {
@@ -57,9 +32,6 @@ fn normalized(patterns: &[String]) -> Result<Vec<(bool, String)>> {
     Ok(out)
 }
 
-/// `restic backup` matches patterns against absolute paths, so every pattern is
-/// rooted at the source; unanchored ones would otherwise also match directories
-/// above it and exclude the whole source (spike S5).
 pub fn backup_excludes(root: &Path, patterns: &[String]) -> Result<Vec<String>> {
     let root = glob_escape(root.to_string_lossy().trim_end_matches('/'));
     Ok(normalized(patterns)?
@@ -70,7 +42,6 @@ pub fn backup_excludes(root: &Path, patterns: &[String]) -> Result<Vec<String>>
         .collect())
 }
 
-/// `restic restore <id>:<path>` matches patterns relative to that subfolder (spike S1).
 pub fn restore_excludes(patterns: &[String]) -> Result<Vec<String>> {
     Ok(normalized(patterns)?
         .into_iter()

src/domain/files/restic/json.rs

@@ -1,8 +1,6 @@
 use serde::Deserialize;
 use serde_json::{Value, json};
 
-/// One line of `restic --json` output. stdout carries `status` / `summary`,
-/// stderr carries `error` / `exit_error` and plain-text fatal messages.
 #[derive(Debug, PartialEq)]
 pub enum Line {
     Status { percent_done: f64 },
@@ -41,14 +39,11 @@ pub struct BackupSummary {
     pub files_new: u64,
     pub files_changed: u64,
     pub files_unmodified: u64,
-    /// Files of the parent snapshot this one no longer has: restic does not count them
-    /// (see `files_removed`). None when unknown.
     #[serde(skip)]
     pub files_removed: Option<u64>,
 }
 
 impl BackupSummary {
-    /// Counters the dashboard shows for this snapshot (`size` carries `data_added_packed`).
     pub fn report(&self) -> Value {
         json!({
             "filesNew": self.files_new,
@@ -59,14 +54,12 @@ impl BackupSummary {
     }
 }
 
-/// One entry of `restic snapshots --json`.
 #[derive(Debug, Deserialize)]
 pub struct Snapshot {
     pub id: String,
     pub paths: Vec<String>,
     #[serde(default)]
     pub parent: Option<String>,
-    /// Written by restic >= 0.17.
     #[serde(default)]
     pub summary: Option<SnapshotSummary>,
 }
@@ -76,10 +69,6 @@ pub struct SnapshotSummary {
     pub total_files_processed: u64,
 }
 
-/// Files of the parent missing from `snapshot_id`, from a `restic snapshots --json` listing:
-/// restic counts a file as changed or unmodified only when the parent has it, so the parent's
-/// file count minus those is what was removed. Some(0) without a parent; None when the parent
-/// is not listed or has no summary.
 pub fn files_removed(snapshots: &[Snapshot], snapshot_id: &str, changed: u64, unmodified: u64) -> Option<u64> {
     let find = |id: &str| snapshots.iter().find(|s| s.id == id);
     let Some(parent) = find(snapshot_id)?.parent.as_deref() else {

src/domain/files/restic/mod.rs

@@ -0,0 +1,6 @@
+pub mod backup;
+pub mod excludes;
+pub mod json;
+pub mod password;
+pub mod repo;
+pub mod restore;

src/domain/files/restic/password.rs

@@ -0,0 +1,7 @@
+use crate::utils::edge_key::derive_key;
+use anyhow::Result;
+
+pub fn derive_password(master_key_b64: &str, generated_id: &str) -> Result<String> {
+    let info = format!("portabase/restic/v1/{generated_id}");
+    Ok(hex::encode(derive_key(master_key_b64, info.as_bytes())?))
+}

src/domain/files/restic/repo.rs

@@ -1,11 +1,13 @@
 use super::json::{self, Line};
-use super::password::{derive_password, local_storage_password, LOCAL_STORAGE_USER};
+use super::password::derive_password;
 use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::backup::logger::JobLogger;
+use crate::services::storage::providers::local::{LOCAL_STORAGE_USER, local_storage_password};
 use crate::services::storage::providers::rclone::helpers::{remote_target, write_config};
 use crate::services::storage::providers::rclone::target::{RcloneTarget, rclone_target};
 use crate::settings::CONFIG;
 use crate::utils::edge_key::EdgeKey;
+use crate::utils::file::folder_or_default;
 use anyhow::{Context, Result, anyhow};
 use serde_json::Value;
 use std::ffi::{OsStr, OsString};
@@ -15,32 +17,20 @@ use tempfile::NamedTempFile;
 use tokio::io::{AsyncBufReadExt, BufReader};
 use tokio::process::Command;
 
-/// One source's repository on one storage channel:
-/// `rclone:<remote>:<base>/<folder>/restic/<generated_id>`, or
-/// `rest:<server_url>/storage/restic/<generated_id>/` on the dashboard's local storage.
 pub struct ResticRepo {
     repository: String,
     password: String,
-    /// `None` when the cache directory cannot be created: restic then runs `--no-cache`.
     cache_dir: Option<PathBuf>,
-    /// rclone config of `rclone:` repositories; `None` on the dashboard's REST server.
     config: Option<NamedTempFile>,
-    /// HTTP basic auth of `rest:` repositories (the dashboard's local storage).
     rest_auth: Option<(String, String)>,
-    /// Keeps temp files referenced by the config (sftp key) alive.
     _target: Option<RcloneTarget>,
 }
 
-/// What a finished restic command left behind.
 pub struct ResticRun {
     pub code: i32,
-    /// stdout lines that are neither `status` nor `summary` (e.g. `snapshots --json`).
     pub stdout: String,
-    /// The last `summary` line, if any.
     pub summary: Option<Value>,
-    /// `error` lines as "item: message" (unreadable files and the like).
     pub errors: Vec<String>,
-    /// Last plain-text / `exit_error` stderr lines, for failure messages.
     pub stderr_tail: String,
 }
 
@@ -55,8 +45,6 @@ impl ResticRun {
     }
 }
 
-/// A full snapshot id. Ids from the dashboard reach restic as positional arguments:
-/// anything else could be parsed as a flag (e.g. `--password-command=<shell>`).
 pub fn is_snapshot_id(id: &str) -> bool {
     id.len() == 64 && id.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'))
 }
@@ -65,7 +53,6 @@ pub fn short(id: &str) -> &str {
     id.get(..8).unwrap_or(id)
 }
 
-/// restic fails hard (not warns) on an unusable cache dir, so probe it up front.
 fn usable_cache_dir(cache_dir: PathBuf) -> Option<PathBuf> {
     match std::fs::create_dir_all(&cache_dir) {
         Ok(()) => Some(cache_dir),
@@ -90,17 +77,11 @@ impl ResticRepo {
             ));
         }
         let target = rclone_target(storage)?;
-        let folder = storage
-            .folder_name
-            .as_deref()
-            .map(|f| f.trim().trim_matches('/'))
-            .filter(|f| !f.is_empty())
-            .unwrap_or("backups");
         let mut repo = Self::new(
             &target.config_text,
             &target.remote_name,
             &target.base_path,
-            folder,
+            folder_or_default(storage.folder_name.as_deref()),
             generated_id,
             &password,
             cache_dir,
@@ -129,8 +110,6 @@ impl ResticRepo {
         })
     }
 
-    /// The dashboard's local storage (`rclone serve restic --append-only`):
-    /// `rest:<base_url>/<generated_id>/`, basic auth only in the child env.
     pub fn rest(base_url: &str, generated_id: &str, password: &str, server_password: &str, cache_dir: PathBuf) -> Self {
         Self {
             repository: format!("rest:{}/{generated_id}/", base_url.trim_end_matches('/')),
@@ -142,7 +121,6 @@ impl ResticRepo {
         }
     }
 
-    /// The dashboard's server refuses deletions: the agent cannot forget there.
     pub fn is_append_only(&self) -> bool {
         self.rest_auth.is_some()
     }
@@ -151,8 +129,6 @@ impl ResticRepo {
         &self.repository
     }
 
-    /// `restic <args> --retry-lock 5m` (+ `--no-cache` without a cache dir); secrets only in the child env. `status`
-    /// lines are logged every 10 %.
     pub async fn run<I, S>(&self, args: I, logger: &JobLogger) -> Result<ResticRun>
     where
         I: IntoIterator<Item = S>,
@@ -177,7 +153,6 @@ impl ResticRepo {
         let mut child = cmd
             .env("RESTIC_REPOSITORY", &self.repository)
             .env("RESTIC_PASSWORD", &self.password)
-            // No TTY: without this restic prints no status lines at all.
             .env("RESTIC_PROGRESS_FPS", "0.2")
             .stdin(Stdio::null())
             .stdout(Stdio::piped())
@@ -237,8 +212,6 @@ impl ResticRepo {
         Ok(run)
     }
 
-    /// `restic cat config`; exit 10 (no repository) → `restic init`. Callers hold
-    /// the source's FileLock, so two inits never race.
     pub async fn ensure_initialized(&self, logger: &JobLogger) -> Result<()> {
         let cat = self.run(["cat", "config"], logger).await?;
         match cat.code {

src/domain/files/restic/restore.rs

@@ -1,20 +1,15 @@
-use super::command::{ResticRepo, is_snapshot_id, short};
 use super::excludes::{glob_escape, restore_excludes};
 use super::json::Snapshot;
+use super::repo::{ResticRepo, is_snapshot_id, short};
 use crate::domain::files::matcher::{exclude_patterns, one_file_system};
 use crate::domain::files::platform::dev;
 use crate::domain::files::restore::ensure_restorable;
-use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::backup::logger::JobLogger;
-use crate::services::config::{DatabaseConfig, DbType};
-use crate::utils::edge_key::EdgeKey;
-use crate::utils::locks::{DbOpLock, FileLock};
+use crate::services::config::DatabaseConfig;
 use anyhow::{Context, Result, bail};
 use std::path::Path;
 use walkdir::WalkDir;
 
-/// Mirror restore of `snapshot_id` into `cfg.path`: files created since are
-/// deleted, paths matching the current exclude patterns are kept (spike S1).
 pub async fn restore(
     repo: &ResticRepo,
     cfg: &DatabaseConfig,
@@ -32,7 +27,6 @@ pub async fn restore(
     }
     let snapshots: Vec<Snapshot> = serde_json::from_str(listed.stdout.trim())
         .context("unexpected `restic snapshots` output")?;
-    // The original path inside the snapshot; the mount point may have moved since.
     let Some(original) = snapshots.first().and_then(|s| s.paths.first()) else {
         bail!("snapshot {} not found in {}", short(snapshot_id), repo.repository());
     };
@@ -78,8 +72,6 @@ pub async fn restore(
     Ok(())
 }
 
-/// Directories under `root` on another device. With `one_file_system` backup
-/// never entered them, so the mirror restore must not delete their content.
 fn mount_points(root: &Path) -> Vec<String> {
     let Ok(meta) = std::fs::metadata(root) else { return Vec::new() };
     let root_dev = dev(&meta);
@@ -99,26 +91,3 @@ fn mount_points(root: &Path) -> Vec<String> {
     }
     out
 }
-
-/// Snapshot restore job: the source's FileLock is held for the whole restore.
-pub async fn run(
-    edge_key: &EdgeKey,
-    cfg: &DatabaseConfig,
-    snapshot_id: &str,
-    storage: &DatabaseStorage,
-    logger: &JobLogger,
-) -> Result<()> {
-    if !matches!(cfg.db_type, DbType::Files) {
-        bail!("snapshot restore needs a files source, got {}", cfg.db_type.as_str());
-    }
-    FileLock::acquire(&cfg.generated_id, DbOpLock::Restore.as_str())
-        .await
-        .context("another backup or restore of this source is running")?;
-    let result = async {
-        let repo = ResticRepo::open(storage, &cfg.generated_id, edge_key)?;
-        restore(&repo, cfg, snapshot_id, logger).await
-    }
-    .await;
-    FileLock::release(&cfg.generated_id).await?;
-    result
-}

src/domain/files/restore.rs

@@ -1,4 +1,4 @@
-use super::matcher::ExcludeMatcher;
+use super::matcher::{ExcludeMatcher, one_file_system};
 use super::platform::{dev, is_root};
 use crate::services::backup::logger::JobLogger;
 use crate::services::config::DatabaseConfig;
@@ -14,25 +14,16 @@ pub async fn run(cfg: DatabaseConfig, archive: PathBuf, logger: Arc<JobLogger>)
 }
 
 fn restore(cfg: &DatabaseConfig, archive: &Path, logger: &JobLogger) -> Result<()> {
-    // Everything below works on the canonical path, like backup.rs.
     let root = ensure_restorable(Path::new(&cfg.path))?;
     let mut matcher = ExcludeMatcher::from_config(cfg)?;
-    let one_file_system = cfg
-        .options
-        .get("one_file_system")
-        .and_then(|v| v.as_bool())
-        .unwrap_or(false);
+    let one_file_system = one_file_system(cfg);
 
-    // Opened once: the wipe may delete the archive's path, never an open handle.
     let file = File::open(archive)
         .with_context(|| format!("cannot open archive {}", archive.display()))?;
 
-    // The wipe is irreversible: prove the whole archive reads back before touching data.
     logger.log("info", format!("Validating archive {}", archive.display()));
     validate_archive(&file).context("archive is corrupt or truncated, nothing was changed")?;
 
-    // Keep the archive's own directory (or the archive itself when it sits directly
-    // in the target), like backup.rs prunes its output.
     let archive_real = fs::canonicalize(archive)?;
     let keep = match archive_real.parent() {
         Some(dir) if dir != root => dir,
@@ -46,7 +37,7 @@ fn restore(cfg: &DatabaseConfig, archive: &Path, logger: &JobLogger) -> Result<(
         "info",
         format!("Removing non-excluded content of {}", root.display()),
     );
-    // Best effort: whatever could not be removed is kept, and extraction always runs.
+
     let failures = wipe_except_excluded(&root, &matcher, one_file_system, logger)?;
 
     logger.log(
@@ -65,9 +56,6 @@ fn restore(cfg: &DatabaseConfig, archive: &Path, logger: &JobLogger) -> Result<(
     Ok(())
 }
 
-/// Extract the whole archive into `root`. Filesystems that refuse chown/chmod/utime
-/// (NFS root_squash, CIFS, NFSv4 idmap mismatch) fail the first pass, after the wipe,
-/// so retry once without metadata: the content matters more than owner and mode.
 pub fn unpack(
     file: &File,
     root: &Path,
@@ -79,9 +67,6 @@ pub fn unpack(
             "warn",
             format!("extraction with ownership/permissions failed: {e:#}; retrying without metadata preservation"),
         );
-        // ponytail: tar-rs still chmods to `mode & 0o777` with preservation off, so a
-        // filesystem that rejects chmod on files the agent just created fails this pass
-        // too; per-entry extraction with deferred directories if that shows up.
         unpack_once(file, root, false, false)?;
     }
     Ok(())
@@ -103,7 +88,6 @@ fn unpack_once(
     Ok(())
 }
 
-/// Read every entry to the end, then drain the gzip stream so its CRC is checked.
 fn validate_archive(file: &File) -> Result<()> {
     let mut gz = GzDecoder::new(file);
     let mut ar = tar::Archive::new(&mut gz);
@@ -111,14 +95,10 @@ fn validate_archive(file: &File) -> Result<()> {
         io::copy(&mut entry?, &mut io::sink())?;
     }
     drop(ar);
-    // tar stops at its end marker, before the gzip trailer.
     gz.read_to_end(&mut Vec::new())?;
     Ok(())
 }
 
-/// Refuse targets a mirror restore must never touch, and return the canonical path to
-/// restore into. The canonical path is checked too, because `/data/..` style escapes
-/// resolve to `/`.
 pub fn ensure_restorable(root: &Path) -> Result<PathBuf> {
     if root.as_os_str().is_empty() || !root.is_absolute() || root == Path::new("/") {
         bail!(
@@ -140,18 +120,6 @@ pub fn ensure_restorable(root: &Path) -> Result<PathBuf> {
     Ok(canonical)
 }
 
-/// Delete everything under `root` that does not match an exclude pattern.
-/// Excluded entries (and their subtrees) are kept; `root` itself is kept.
-///
-/// Mount points (a directory on a different device than its parent): with
-/// `one_file_system` they are kept untouched like an excluded path, matching what
-/// backup never archived. Without it their contents are wiped but the mount point
-/// itself is never removed (`remove_dir` on it fails with EBUSY) and counts as kept.
-/// Special files (sockets, fifos, devices) are kept: backup never archives them.
-///
-/// Best effort: an entry that cannot be read or removed is logged, kept (so its
-/// parents are kept too) and counted; the returned count is how many failed.
-/// Only an unreadable `root` is an error, and then nothing was deleted.
 pub fn wipe_except_excluded(
     root: &Path,
     matcher: &ExcludeMatcher,
@@ -183,8 +151,6 @@ impl Wipe<'_> {
         self.failures += 1;
     }
 
-    /// Returns true when something was kept inside `dir`. `rel` is `dir` relative to
-    /// the root and `dir_dev` its device.
     fn dir(&mut self, dir: &Path, rel: &Path, dir_dev: u64) -> bool {
         let entries = match fs::read_dir(dir) {
             Ok(entries) => entries,
@@ -209,7 +175,7 @@ impl Wipe<'_> {
                 kept = true;
                 continue;
             }
-            // symlink_metadata: a symlinked directory is removed as a link, never followed.
+
             let meta = match fs::symlink_metadata(&path) {
                 Ok(meta) => meta,
                 Err(e) => {
@@ -233,7 +199,6 @@ impl Wipe<'_> {
             } else if ft.is_file() || ft.is_symlink() {
                 fs::remove_file(&path)
             } else {
-                // Socket, fifo or device: backup never archives them, so keep them.
                 kept = true;
                 continue;
             };

src/domain/files/sync/backup.rs

@@ -0,0 +1,45 @@
+use super::rclone::{replica_target, sync_dir};
+use super::stats::SyncStats;
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::backup::logger::JobLogger;
+use crate::services::config::DatabaseConfig;
+use crate::services::storage::providers::local::{LOCAL_STORAGE_USER, local_storage_password};
+use crate::services::storage::providers::rclone::helpers::{build_rclone_config, obscure_password};
+use crate::services::storage::providers::rclone::target::rclone_target;
+use crate::utils::edge_key::EdgeKey;
+use crate::utils::file::folder_or_default;
+use anyhow::Result;
+
+pub fn replica_path(storage: &DatabaseStorage, generated_id: &str) -> String {
+    format!("{}/sync/{generated_id}/current", folder_or_default(storage.folder_name.as_deref()))
+}
+
+const LOCAL_REMOTE: &str = "pblocal";
+
+pub fn local_sync_config(edge_key: &EdgeKey) -> Result<String> {
+    build_rclone_config(
+        LOCAL_REMOTE,
+        &[
+            ("type", "webdav".into()),
+            ("url", format!("{}/storage/sync", edge_key.server_url.trim_end_matches('/'))),
+            ("vendor", "rclone".into()),
+            ("user", LOCAL_STORAGE_USER.into()),
+            ("pass", obscure_password(&local_storage_password(&edge_key.master_key_b64)?)?),
+        ],
+    )
+}
+
+pub async fn replicate(edge_key: &EdgeKey, cfg: &DatabaseConfig, storage: &DatabaseStorage, logger: &JobLogger) -> Result<SyncStats> {
+    if storage.provider == "local" {
+        let dest = format!("{LOCAL_REMOTE}:{}/current", cfg.generated_id);
+        return sync_dir(&local_sync_config(edge_key)?, &dest, cfg, logger).await;
+    }
+    let target = rclone_target(storage)?;
+    let dest = replica_target(
+        &target.remote_name,
+        &target.base_path,
+        folder_or_default(storage.folder_name.as_deref()),
+        &cfg.generated_id,
+    );
+    sync_dir(&target.config_text, &dest, cfg, logger).await
+}

src/domain/files/sync/excludes.rs

@@ -1,7 +1,5 @@
-use crate::services::restic::excludes::bracket_negation;
+use crate::domain::files::matcher::bracket_negation;
 
-/// P0 exclude patterns → `rclone sync --exclude` values. rclone filters are
-/// relative to the sync root (spike 2026-10-03: 19/19 patterns give the P0 file set).
 pub fn sync_excludes(patterns: &[String]) -> Vec<String> {
     let mut out = Vec::new();
     for raw in patterns {
@@ -12,8 +10,6 @@ pub fn sync_excludes(patterns: &[String]) -> Vec<String> {
         };
         let mut rest = rest.trim_end_matches('/');
         if !anchored {
-            // Unanchored rclone patterns already match at any depth; a leading
-            // `**/` would stop them from matching at the root.
             while let Some(stripped) = rest.strip_prefix("**/") {
                 rest = stripped;
             }
@@ -22,7 +18,6 @@ pub fn sync_excludes(patterns: &[String]) -> Vec<String> {
             continue;
         }
         let rest = bracket_negation(rest);
-        // Collapse repeated /**/ into a single /**/
         let rest = collapse_double_stars(&rest);
         let full = if anchored { format!("/{rest}") } else { rest };
         for variant in zero_dir_variants(&full) {
@@ -33,8 +28,6 @@ pub fn sync_excludes(patterns: &[String]) -> Vec<String> {
     out
 }
 
-/// Collapse repeated `/**/` sequences into a single `/**/`.
-/// E.g., `a/**/**/b` becomes `a/**/b`.
 fn collapse_double_stars(pattern: &str) -> String {
     let mut result = pattern.to_string();
     while result.contains("/**/**/") {
@@ -43,12 +36,9 @@ fn collapse_double_stars(pattern: &str) -> String {
     result
 }
 
-/// rclone's `**` between slashes needs at least one directory; globset's also
-/// matches none. Emit every combination of each `/**/` kept or collapsed to `/`.
 fn zero_dir_variants(pattern: &str) -> Vec<String> {
     let parts: Vec<&str> = pattern.split("/**/").collect();
     let gaps = parts.len() - 1;
-    // ponytail: 2^gaps variants; past 4 gaps only "all kept" and "all collapsed".
     if gaps > 4 {
         return vec![pattern.to_string(), pattern.replace("/**/", "/")];
     }

src/domain/files/sync/mod.rs

@@ -0,0 +1,4 @@
+pub mod backup;
+pub mod excludes;
+pub mod rclone;
+pub mod stats;

src/domain/files/sync/rclone.rs

@@ -1,6 +1,6 @@
 use super::excludes::sync_excludes;
 use super::stats::{SyncStats, parse_log};
-use crate::domain::files::matcher::{exclude_patterns, one_file_system};
+use crate::domain::files::matcher::{exclude_patterns, one_file_system, source_root};
 use crate::services::backup::logger::JobLogger;
 use crate::services::config::DatabaseConfig;
 use crate::services::storage::providers::rclone::helpers::{remote_target, write_config};
@@ -9,22 +9,12 @@ use std::path::Path;
 use std::process::Stdio;
 use tokio::process::Command;
 
-/// `<remote>:<base>/<folder>/sync/<generated_id>/current`
 pub fn replica_target(remote_name: &str, base_path: &str, folder: &str, generated_id: &str) -> String {
     remote_target(remote_name, base_path, &format!("{folder}/sync/{generated_id}/current"))
 }
 
-/// Mirrors `cfg.path` onto `dest` (`<remote>:<path>`); files matching an exclude are removed
-/// from the replica too (`--delete-excluded`). rclone skips its deletions when an error
-/// happens before the delete phase ("not deleting files as there were IO errors");
-/// an error during the delete phase can leave the replica partially updated.
-/// An empty source is refused so an unmounted volume cannot wipe the replica.
 pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logger: &JobLogger) -> Result<SyncStats> {
-    let root = std::fs::canonicalize(&cfg.path)
-        .with_context(|| format!("cannot read source directory {}", cfg.path))?;
-    if !root.is_dir() {
-        bail!("source path {} is not a directory", root.display());
-    }
+    let root = source_root(cfg)?;
     if std::fs::read_dir(&root)
         .with_context(|| format!("cannot read source directory {}", root.display()))?
         .next()
@@ -45,7 +35,6 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge
         .arg(dest)
         .args([
             "--links", "--delete-excluded", "--use-json-log", "--stats", "1h", "--stats-log-level", "NOTICE", "--retries", "1",
-            // An unreachable endpoint fails in ~1–2 min instead of blocking the next runs.
             "--contimeout", "30s", "--low-level-retries", "3",
         ]);
     if one_file_system(cfg) {
@@ -80,7 +69,6 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge
         bail!("rclone sync failed ({}): {reason}; the replica may be partially updated", out.status);
     }
     let mut stats = log.stats.unwrap_or_default();
-    // Measured on the replica itself, so it is what is stored, not inferred from the source.
     stats.replica_bytes = match remote_size(config.path(), dest).await {
         Ok(bytes) => Some(bytes),
         Err(e) => {
@@ -96,7 +84,6 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge
     Ok(stats)
 }
 
-/// Bytes stored under `target` (`<remote>:<path>`), per `rclone size`.
 async fn remote_size(config: &Path, target: &str) -> Result<u64> {
     let out = Command::new("rclone")
         .args(["--contimeout", "30s", "--low-level-retries", "3", "--config"])

src/domain/files/sync/stats.rs

@@ -1,7 +1,6 @@
 use serde::Deserialize;
 use serde_json::{Value, json};
 
-/// Final `stats` object of `rclone --use-json-log --stats 1h --stats-log-level NOTICE`.
 #[derive(Debug, Default, Deserialize, PartialEq)]
 pub struct SyncStats {
     #[serde(default)]
@@ -14,13 +13,11 @@ pub struct SyncStats {
     pub errors: u64,
     #[serde(default, rename = "lastError")]
     pub last_error: Option<String>,
-    /// Replica size after a successful run, from `rclone size` on the destination (not an rclone stat).
     #[serde(skip)]
     pub replica_bytes: Option<u64>,
 }
 
 impl SyncStats {
-    /// Counters the dashboard shows for this run (`size` carries `bytes`).
     pub fn report(&self) -> Value {
         json!({
             "filesTransferred": self.transfers,
@@ -30,8 +27,6 @@ impl SyncStats {
     }
 }
 
-/// What `rclone sync` printed on stderr: its last stats object and its error
-/// messages (JSON `level: error|critical|alert|emergency` lines and any plain-text line).
 #[derive(Debug, Default)]
 pub struct SyncLog {
     pub stats: Option<SyncStats>,
@@ -61,7 +56,6 @@ pub fn parse_log(stderr: &str) -> SyncLog {
             }
         }
     }
-    // Keep only the last 20 messages
     if log.errors.len() > 20 {
         log.errors = log.errors.into_iter().rev().take(20).collect::<Vec<_>>();
         log.errors.reverse();

src/services/backup/executor.rs

@@ -1,10 +1,9 @@
+use super::files_engine::{self, FilesEngine};
 use super::logger::JobLogger;
 use super::models::BackupResult;
 use super::service::BackupService;
 use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::config::{DatabaseConfig, DbType};
-use crate::services::restic;
-use crate::services::sync;
 use crate::utils::common::BackupMethod;
 use crate::utils::locks::FileLock;
 
@@ -34,13 +33,9 @@ impl BackupService {
         let backup = self.create_backup_record(&generated_id, &method).await?;
         let backup_id = backup.backup.id;
 
-        if engine == "restic" || engine == "sync" {
+        if let Some(files_engine) = FilesEngine::parse(&engine) {
             if matches!(db_cfg.db_type, DbType::Files) {
-                let uploads = if engine == "restic" {
-                    restic::backup::run(&self.ctx, &db_cfg, &storages, &backup_id, &logger).await
-                } else {
-                    sync::backup::run(&self.ctx, &db_cfg, &storages, &backup_id, &logger).await
-                };
+                let uploads = files_engine::run(&self.ctx, &db_cfg, files_engine, &storages, &backup_id, &logger).await;
                 logger.log("info", "Database backup job finished".to_string());
                 let result = BackupResult {
                     generated_id: generated_id.clone(),
@@ -51,8 +46,6 @@ impl BackupService {
                 };
                 let duration_ms = start.elapsed().as_millis() as f64;
                 let logs = Arc::try_unwrap(logger).unwrap_or_else(|_| JobLogger::new()).into_entries();
-                // fileSize = average logical size across storages (send_result): restic reports
-                // the size each snapshot processed; sync reports none, so it stays null.
                 self.send_result(result, uploads, &backup_id, logs, duration_ms).await?;
                 return Ok(());
             }

src/services/backup/files_engine.rs

@@ -0,0 +1,166 @@
+use super::logger::JobLogger;
+use super::models::UploadResult;
+use crate::core::context::Context as CoreContext;
+use crate::domain::files::restic::backup::snapshot;
+use crate::domain::files::restic::repo::ResticRepo;
+use crate::domain::files::sync::backup::{replica_path, replicate};
+use crate::services::api::models::agent::status::DatabaseStorage;
+use crate::services::config::DatabaseConfig;
+use crate::utils::locks::{DbOpLock, FileLock};
+use anyhow::Result;
+use serde_json::Value;
+
+#[derive(Clone, Copy)]
+pub enum FilesEngine {
+    Restic,
+    Sync,
+}
+
+impl FilesEngine {
+    pub fn parse(engine: &str) -> Option<Self> {
+        match engine {
+            "restic" => Some(Self::Restic),
+            "sync" => Some(Self::Sync),
+            _ => None,
+        }
+    }
+
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Restic => "restic",
+            Self::Sync => "sync",
+        }
+    }
+
+    fn label(self) -> &'static str {
+        match self {
+            Self::Restic => "Snapshot",
+            Self::Sync => "Sync",
+        }
+    }
+}
+
+struct Transfer {
+    remote_path: String,
+    size: u64,
+    report: Value,
+    total_size: Option<u64>,
+}
+
+pub async fn run(
+    ctx: &CoreContext,
+    cfg: &DatabaseConfig,
+    engine: FilesEngine,
+    storages: &[DatabaseStorage],
+    backup_id: &str,
+    logger: &JobLogger,
+) -> Vec<UploadResult> {
+    if let Err(e) = FileLock::acquire(&cfg.generated_id, DbOpLock::Backup.as_str()).await {
+        logger.log("error", format!("{} aborted: {e}", engine.label()));
+        return Vec::new();
+    }
+    let mut results = Vec::with_capacity(storages.len());
+    for storage in storages {
+        results.push(one_storage(ctx, cfg, engine, storage, backup_id, logger).await);
+    }
+    if let Err(e) = FileLock::release(&cfg.generated_id).await {
+        logger.log("warn", format!("Failed to release the backup lock: {e}"));
+    }
+    results
+}
+
+async fn transfer(
+    ctx: &CoreContext,
+    cfg: &DatabaseConfig,
+    engine: FilesEngine,
+    storage: &DatabaseStorage,
+    backup_storage_id: &str,
+    logger: &JobLogger,
+) -> Result<Transfer> {
+    match engine {
+        FilesEngine::Restic => {
+            let repo = ResticRepo::open(storage, &cfg.generated_id, &ctx.edge_key)?;
+            let summary = snapshot(&repo, cfg, backup_storage_id, logger).await?;
+            Ok(Transfer {
+                size: summary.data_added_packed,
+                report: summary.report(),
+                total_size: Some(summary.total_bytes_processed),
+                remote_path: summary.snapshot_id,
+            })
+        }
+        FilesEngine::Sync => {
+            let stats = replicate(&ctx.edge_key, cfg, storage, logger).await?;
+            Ok(Transfer {
+                remote_path: replica_path(storage, &cfg.generated_id),
+                size: stats.bytes,
+                report: stats.report(),
+                total_size: stats.replica_bytes,
+            })
+        }
+    }
+}
+
+pub(crate) async fn one_storage(
+    ctx: &CoreContext,
+    cfg: &DatabaseConfig,
+    engine: FilesEngine,
+    storage: &DatabaseStorage,
+    backup_id: &str,
+    logger: &JobLogger,
+) -> UploadResult {
+    let agent_id = ctx.edge_key.agent_id.clone();
+    let failed = |error: String| UploadResult {
+        storage_id: storage.id.clone(),
+        success: false,
+        error: Some(error),
+        remote_file_path: None,
+        total_size: None,
+    };
+
+    let backup_storage_id = match ctx
+        .api
+        .backup_upload_init(agent_id.clone(), cfg.generated_id.clone(), storage.id.clone(), backup_id, Some(engine.as_str()))
+        .await
+    {
+        Ok(Some(response)) => response.backup_storage.id,
+        Ok(None) => {
+            logger.log("error", "Upload init returned empty response");
+            return failed("backup_upload_init returned empty response".into());
+        }
+        Err(e) => {
+            logger.log("error", format!("Upload init failed: {e}"));
+            return failed("backup_upload_init failed".into());
+        }
+    };
+
+    match transfer(ctx, cfg, engine, storage, &backup_storage_id, logger).await {
+        Ok(t) => match ctx
+            .api
+            .backup_upload_success(agent_id, cfg.generated_id.clone(), backup_storage_id, t.remote_path.clone(), t.size, t.report, backup_id)
+            .await
+        {
+            Ok(_) => UploadResult {
+                storage_id: storage.id.clone(),
+                success: true,
+                error: None,
+                remote_file_path: Some(t.remote_path),
+                total_size: t.total_size,
+            },
+            Err(e) => {
+                logger.log("error", format!("Upload status update failed for {}: {e}", storage.id));
+                failed(e.to_string())
+            }
+        },
+        Err(e) => {
+            logger.log("error", format!("{} to storage {} failed: {e:#}", engine.label(), storage.id));
+            if let Err(err) = ctx
+                .api
+                .backup_upload_status(agent_id, cfg.generated_id.clone(), backup_storage_id, "failed", String::new(), 0u64, backup_id)
+                .await
+            {
+                logger.log("error", format!("Failed-status update failed for {}: {err}", storage.id));
+            }
+            failed(format!("{e:#}"))
+        }
+    }
+}

src/services/backup/mod.rs

@@ -2,6 +2,7 @@ pub mod compressor;
 pub mod logger;
 pub mod dispatcher;
 pub mod executor;
+pub mod files_engine;
 pub mod helpers;
 pub mod models;
 pub mod result;

src/services/dashboard_config.rs

@@ -20,7 +20,6 @@ pub fn merge(local: &[DatabaseConfig], dashboard: &[DatabaseConfig]) -> Database
     DatabasesConfig { databases }
 }
 
-/// `generated_id`s of local (databases.json) sources that no dashboard config replaces.
 pub fn local_only_ids(local: &[DatabaseConfig], dashboard: &[DatabaseConfig]) -> HashSet<String> {
     local
         .iter()

src/services/mod.rs

@@ -3,8 +3,6 @@ pub mod backup;
 pub mod config;
 pub mod cron;
 pub mod dashboard_config;
-pub mod restic;
 pub mod restore;
 pub mod status;
 pub mod storage;
-pub mod sync;

src/services/restic/backup.rs

@@ -1,224 +0,0 @@
-use super::command::{ResticRepo, short};
-use super::excludes::backup_excludes;
-use super::json::{BackupSummary, Snapshot, files_removed};
-use crate::core::context::Context as CoreContext;
-use crate::domain::files::matcher::{exclude_patterns, one_file_system};
-use crate::services::api::models::agent::status::DatabaseStorage;
-use crate::services::backup::logger::JobLogger;
-use crate::services::backup::models::UploadResult;
-use crate::services::config::DatabaseConfig;
-use crate::utils::locks::{DbOpLock, FileLock};
-use anyhow::{Context, Result, bail};
-
-/// One snapshot of `cfg.path`, tagged with its `backup_storage` id.
-/// Exit 3 (unreadable files) fails the run and forgets the snapshot: a mirror
-/// restore of it would delete the files it is missing (P0 rule R21). On an
-/// append-only repository the agent cannot forget it, so the dashboard does.
-pub async fn snapshot(
-    repo: &ResticRepo,
-    cfg: &DatabaseConfig,
-    backup_storage_id: &str,
-    logger: &JobLogger,
-) -> Result<BackupSummary> {
-    let root = std::fs::canonicalize(&cfg.path)
-        .with_context(|| format!("cannot read source directory {}", cfg.path))?;
-    if !root.is_dir() {
-        bail!("source path {} is not a directory", root.display());
-    }
-    repo.ensure_initialized(logger).await?;
-
-    // Stable host: container hostnames change on recreate and would break
-    // parent-snapshot detection (full rescan every time).
-    let host = format!("portabase-{}", cfg.generated_id);
-    let mut args: Vec<String> = vec![
-        "backup".into(),
-        root.to_string_lossy().into_owned(),
-        "--json".into(),
-        "--host".into(),
-        host.clone(),
-        "--tag".into(),
-        "portabase".into(),
-        "--tag".into(),
-        format!("bs:{backup_storage_id}"),
-    ];
-    for pattern in backup_excludes(&root, &exclude_patterns(cfg))? {
-        args.push("--exclude".into());
-        args.push(pattern);
-    }
-    if one_file_system(cfg) {
-        args.push("--one-file-system".into());
-    }
-
-    logger.log("info", format!("Snapshotting {} into {}", root.display(), repo.repository()));
-    let run = repo.run(&args, logger).await?;
-    for error in &run.errors {
-        logger.log("warn", format!("Unreadable: {error}"));
-    }
-    let summary: Option<BackupSummary> = run
-        .summary
-        .clone()
-        .map(serde_json::from_value)
-        .transpose()
-        .context("unexpected restic backup summary")?;
-
-    match (run.code, summary) {
-        (0, Some(mut s)) => {
-            s.files_removed = match removed(repo, &host, &s, logger).await {
-                Ok(removed) => removed,
-                Err(e) => {
-                    logger.log("warn", format!("Could not count removed files: {e:#}"));
-                    None
-                }
-            };
-            let removed = s.files_removed.map(|n| format!(", {n} removed")).unwrap_or_default();
-            logger.log(
-                "info",
-                // data_added_packed: new file and directory blobs, compressed and encrypted. The
-                // repository grows a little more (pack headers, index and snapshot files).
-                format!(
-                    "Snapshot {}: {} new, {} changed, {} unmodified{removed} file(s); {} byte(s) scanned, {} byte(s) of new data",
-                    short(&s.snapshot_id), s.files_new, s.files_changed, s.files_unmodified, s.total_bytes_processed, s.data_added_packed
-                ),
-            );
-            Ok(s)
-        }
-        (0, None) => bail!("restic backup returned no summary"),
-        (3, summary) => {
-            let mut fate = "snapshot discarded";
-            if let Some(s) = summary {
-                if repo.is_append_only() {
-                    logger.log(
-                        "warn",
-                        format!("Incomplete snapshot {} left for the dashboard to forget", short(&s.snapshot_id)),
-                    );
-                    fate = "snapshot not restorable (the dashboard forgets it)";
-                } else {
-                    let forget = repo.run(["forget", s.snapshot_id.as_str()], logger).await?;
-                    if forget.code != 0 {
-                        logger.log(
-                            "error",
-                            format!("Could not forget incomplete snapshot {}: {}", s.snapshot_id, forget.error("forget")),
-                        );
-                    }
-                }
-            }
-            bail!(
-                "{} file(s) could not be read; {fate} so a mirror restore cannot delete them (fix permissions or add an exclude)",
-                run.errors.len()
-            )
-        }
-        _ => Err(run.error("backup")),
-    }
-}
-
-/// Removed files of snapshot `s`, against the parent restic actually used (one listing of
-/// the source's snapshots, no index load).
-async fn removed(repo: &ResticRepo, host: &str, s: &BackupSummary, logger: &JobLogger) -> Result<Option<u64>> {
-    let listed = repo.run(["snapshots", "--json", "--host", host], logger).await?;
-    if listed.code != 0 {
-        return Err(listed.error("snapshots"));
-    }
-    let snapshots: Vec<Snapshot> =
-        serde_json::from_str(listed.stdout.trim()).context("unexpected `restic snapshots` output")?;
-    Ok(files_removed(&snapshots, &s.snapshot_id, s.files_changed, s.files_unmodified))
-}
-
-/// Snapshots every storage in turn (one scan each, v1). Never errors: each
-/// storage reports its own `backup_storage` row, like the archive uploader.
-pub async fn run(
-    ctx: &CoreContext,
-    cfg: &DatabaseConfig,
-    storages: &[DatabaseStorage],
-    backup_id: &str,
-    logger: &JobLogger,
-) -> Vec<UploadResult> {
-    if let Err(e) = FileLock::acquire(&cfg.generated_id, DbOpLock::Backup.as_str()).await {
-        logger.log("error", format!("Snapshot aborted: {e}"));
-        return Vec::new();
-    }
-    let mut results = Vec::with_capacity(storages.len());
-    for storage in storages {
-        results.push(one_storage(ctx, cfg, storage, backup_id, logger).await);
-    }
-    if let Err(e) = FileLock::release(&cfg.generated_id).await {
-        logger.log("warn", format!("Failed to release the backup lock: {e}"));
-    }
-    results
-}
-
-pub(crate) async fn one_storage(
-    ctx: &CoreContext,
-    cfg: &DatabaseConfig,
-    storage: &DatabaseStorage,
-    backup_id: &str,
-    logger: &JobLogger,
-) -> UploadResult {
-    let agent_id = ctx.edge_key.agent_id.clone();
-    let failed = |error: String| UploadResult {
-        storage_id: storage.id.clone(),
-        success: false,
-        error: Some(error),
-        remote_file_path: None,
-        total_size: None,
-    };
-
-    let backup_storage_id = match ctx
-        .api
-        .backup_upload_init(agent_id.clone(), cfg.generated_id.clone(), storage.id.clone(), backup_id, Some("restic"))
-        .await
-    {
-        Ok(Some(response)) => response.backup_storage.id,
-        Ok(None) => {
-            logger.log("error", "Upload init returned empty response");
-            return failed("backup_upload_init returned empty response".into());
-        }
-        Err(e) => {
-            logger.log("error", format!("Upload init failed: {e}"));
-            return failed("backup_upload_init failed".into());
-        }
-    };
-
-    let outcome = match ResticRepo::open(storage, &cfg.generated_id, &ctx.edge_key) {
-        Ok(repo) => snapshot(&repo, cfg, &backup_storage_id, logger).await,
-        Err(e) => Err(e),
-    };
-
-    match outcome {
-        Ok(summary) => match ctx
-            .api
-            .backup_upload_success(
-                agent_id,
-                cfg.generated_id.clone(),
-                backup_storage_id,
-                summary.snapshot_id.clone(),
-                summary.data_added_packed,
-                summary.report(),
-                backup_id,
-            )
-            .await
-        {
-            Ok(_) => UploadResult {
-                storage_id: storage.id.clone(),
-                success: true,
-                error: None,
-                remote_file_path: Some(summary.snapshot_id),
-                total_size: Some(summary.total_bytes_processed),
-            },
-            Err(e) => {
-                logger.log("error", format!("Upload status update failed for {}: {e}", storage.id));
-                failed(e.to_string())
-            }
-        },
-        Err(e) => {
-            logger.log("error", format!("Snapshot to storage {} failed: {e:#}", storage.id));
-            if let Err(err) = ctx
-                .api
-                .backup_upload_status(agent_id, cfg.generated_id.clone(), backup_storage_id, "failed", String::new(), 0u64, backup_id)
-                .await
-            {
-                logger.log("error", format!("Failed-status update failed for {}: {err}", storage.id));
-            }
-            failed(format!("{e:#}"))
-        }
-    }
-}

src/services/restic/mod.rs

@@ -1,7 +0,0 @@
-//! Snapshots mode: restic repositories reached through the P1 rclone mapper.
-pub mod backup;
-pub mod command;
-pub mod excludes;
-pub mod json;
-pub mod password;
-pub mod restore;

src/services/restic/password.rs

@@ -1,42 +0,0 @@
-use anyhow::{Context, Result};
-use base64::{Engine as _, engine::general_purpose};
-use hmac::{Hmac, Mac};
-use sha2::Sha256;
-
-type HmacSha256 = Hmac<Sha256>;
-
-/// Repository password: hex(HKDF-SHA256(master key, salt = "",
-/// info = "portabase/restic/v1/<generated_id>", 32 bytes)). The dashboard derives
-/// the same value (`src/lib/restic/repo.ts`), so it is never stored nor sent.
-pub fn derive_password(master_key_b64: &str, generated_id: &str) -> Result<String> {
-    let master_key = general_purpose::STANDARD
-        .decode(master_key_b64.trim())
-        .context("master key is not valid base64")?;
-    let info = format!("portabase/restic/v1/{generated_id}");
-    Ok(hex::encode(hkdf_sha256_32(&master_key, info.as_bytes())))
-}
-
-/// User of the dashboard's local-storage servers (`/storage/restic`, `/storage/sync`).
-pub const LOCAL_STORAGE_USER: &str = "portabase";
-
-/// Their password: hex(HKDF-SHA256(master key, salt = "", info = "portabase/local-storage/v1",
-/// 32 bytes)). The dashboard derives the same value (`src/lib/local-storage/credential.ts`).
-pub fn local_storage_password(master_key_b64: &str) -> Result<String> {
-    let master_key = general_purpose::STANDARD
-        .decode(master_key_b64.trim())
-        .context("master key is not valid base64")?;
-    Ok(hex::encode(hkdf_sha256_32(&master_key, b"portabase/local-storage/v1")))
-}
-
-/// RFC 5869 with an empty salt and a single expand block (L = HashLen = 32).
-pub fn hkdf_sha256_32(ikm: &[u8], info: &[u8]) -> [u8; 32] {
-    // An empty HMAC key is zero-padded: the RFC's default salt of HashLen zeros.
-    let mut extract = HmacSha256::new_from_slice(&[]).expect("HMAC takes any key length");
-    extract.update(ikm);
-    let prk = extract.finalize().into_bytes();
-
-    let mut expand = HmacSha256::new_from_slice(&prk).expect("HMAC takes any key length");
-    expand.update(info);
-    expand.update(&[1]);
-    expand.finalize().into_bytes().into()
-}

src/services/restore/executor.rs

@@ -2,9 +2,11 @@ use super::models::RestoreResult;
 use super::service::RestoreService;
 use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::backup::logger::JobLogger;
-use crate::services::config::DatabaseConfig;
-use crate::services::restic;
-use anyhow::Result;
+use crate::domain::files::restic::repo::ResticRepo;
+use crate::domain::files::restic::restore::restore;
+use crate::services::config::{DatabaseConfig, DbType};
+use crate::utils::locks::{DbOpLock, FileLock};
+use anyhow::{Context, Result, bail};
 use std::sync::Arc;
 use std::time::Instant;
 use tempfile::TempDir;
@@ -58,9 +60,7 @@ impl RestoreService {
         logger.log("info", "Snapshot restore job started".to_string());
 
         let outcome = match (snapshot_id, storage) {
-            (Some(id), Some(storage)) => {
-                restic::restore::run(&self.ctx.edge_key, &cfg, &id, &storage, &logger).await
-            }
+            (Some(id), Some(storage)) => self.restic_restore(&cfg, &id, &storage, &logger).await,
             _ => Err(anyhow::anyhow!(
                 "incomplete snapshot restore payload (snapshotId or storage missing)"
             )),
@@ -82,4 +82,26 @@ impl RestoreService {
         self.send_result(result, logs, duration_ms).await?;
         Ok(())
     }
+
+    async fn restic_restore(
+        &self,
+        cfg: &DatabaseConfig,
+        snapshot_id: &str,
+        storage: &DatabaseStorage,
+        logger: &JobLogger,
+    ) -> Result<()> {
+        if !matches!(cfg.db_type, DbType::Files) {
+            bail!("snapshot restore needs a files source, got {}", cfg.db_type.as_str());
+        }
+        FileLock::acquire(&cfg.generated_id, DbOpLock::Restore.as_str())
+            .await
+            .context("another backup or restore of this source is running")?;
+        let result = async {
+            let repo = ResticRepo::open(storage, &cfg.generated_id, &self.ctx.edge_key)?;
+            restore(&repo, cfg, snapshot_id, logger).await
+        }
+        .await;
+        FileLock::release(&cfg.generated_id).await?;
+        result
+    }
 }

src/services/status.rs

@@ -36,7 +36,6 @@ pub fn resolve_dashboard_config(
     Ok(())
 }
 
-/// A storage channel sent encrypted as a one-element JSON array.
 fn decrypt_storage(ciphertext: &str, master_key_b64: &str, what: &str) -> Result<Option<DatabaseStorage>, String> {
     let plaintext = decrypt_json_gcm(ciphertext, master_key_b64)
         .map_err(|e| format!("Failed to decrypt {what} storage: {e}"))?;
@@ -45,7 +44,6 @@ fn decrypt_storage(ciphertext: &str, master_key_b64: &str, what: &str) -> Result
     Ok(storages.into_iter().next())
 }
 
-/// Decrypts the storage channel of a snapshot restore (`restore.storageCiphertext`).
 pub fn resolve_restore_storage(status: &mut DatabaseStatus, master_key_b64: &str) -> Result<(), String> {
     if let Some(ciphertext) = status.data.restore.storage_ciphertext.clone() {
         status.data.restore.storage = decrypt_storage(&ciphertext, master_key_b64, "restore")?;
@@ -53,8 +51,6 @@ pub fn resolve_restore_storage(status: &mut DatabaseStatus, master_key_b64: &str
     Ok(())
 }
 
-/// `method` reported in the ping for a files source declared in databases.json;
-/// `None` for dashboard-managed sources (the dashboard owns their method) and other dbms.
 pub fn payload_method(db: &DatabaseConfig, local_ids: &HashSet<String>) -> Option<&'static str> {
     (matches!(db.db_type, DbType::Files) && local_ids.contains(&db.generated_id)).then(|| files_method(db))
 }

src/services/storage/providers/local.rs

@@ -3,15 +3,23 @@ use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::backup::models::{BackupResult, UploadResult};
 use crate::services::storage::StorageProvider;
 use crate::utils::common::BackupMethod;
+use crate::utils::edge_key::derive_key;
 use crate::utils::file::{full_file_name, full_file_path};
 use crate::utils::stream::build_stream;
 use crate::utils::tus::upload_to_tus_stream_with_headers;
+use anyhow::Result;
 use async_trait::async_trait;
 use reqwest::header::{HeaderMap, HeaderValue};
 use std::sync::Arc;
 use tokio::fs;
 use tracing::error;
 
+pub const LOCAL_STORAGE_USER: &str = "portabase";
+
+pub fn local_storage_password(master_key_b64: &str) -> Result<String> {
+    Ok(hex::encode(derive_key(master_key_b64, b"portabase/local-storage/v1")?))
+}
+
 pub struct LocalProvider;
 
 #[async_trait]

src/services/sync/backup.rs

@@ -1,147 +0,0 @@
-//! Sync as a backup method: one `rclone sync` per storage channel, each reported
-//! as its own `backup_storage` row (`engine = sync`), like restic snapshots.
-use super::rclone::{replica_target, sync_dir};
-use super::stats::SyncStats;
-use crate::core::context::Context as CoreContext;
-use crate::services::api::models::agent::status::DatabaseStorage;
-use crate::services::backup::logger::JobLogger;
-use crate::services::backup::models::UploadResult;
-use crate::services::config::DatabaseConfig;
-use crate::services::restic::password::{LOCAL_STORAGE_USER, local_storage_password};
-use crate::services::storage::providers::rclone::helpers::{build_rclone_config, obscure_password};
-use crate::services::storage::providers::rclone::target::rclone_target;
-use crate::utils::edge_key::EdgeKey;
-use crate::utils::locks::{DbOpLock, FileLock};
-use anyhow::Result;
-
-/// Syncs every storage in turn, under the source's backup lock. Never errors:
-/// each storage reports its own `backup_storage` row.
-pub async fn run(
-    ctx: &CoreContext,
-    cfg: &DatabaseConfig,
-    storages: &[DatabaseStorage],
-    backup_id: &str,
-    logger: &JobLogger,
-) -> Vec<UploadResult> {
-    if let Err(e) = FileLock::acquire(&cfg.generated_id, DbOpLock::Backup.as_str()).await {
-        logger.log("error", format!("Sync aborted: {e}"));
-        return Vec::new();
-    }
-    let mut results = Vec::with_capacity(storages.len());
-    for storage in storages {
-        results.push(one_storage(ctx, cfg, storage, backup_id, logger).await);
-    }
-    if let Err(e) = FileLock::release(&cfg.generated_id).await {
-        logger.log("warn", format!("Failed to release the backup lock: {e}"));
-    }
-    results
-}
-
-fn folder(storage: &DatabaseStorage) -> &str {
-    storage
-        .folder_name
-        .as_deref()
-        .map(|f| f.trim().trim_matches('/'))
-        .filter(|f| !f.is_empty())
-        .unwrap_or("backups")
-}
-
-/// `<folder>/sync/<generated_id>/current`, relative to the channel's base path.
-pub fn replica_path(storage: &DatabaseStorage, generated_id: &str) -> String {
-    format!("{}/sync/{generated_id}/current", folder(storage))
-}
-
-const LOCAL_REMOTE: &str = "pblocal";
-
-/// rclone config of the dashboard's local storage (`rclone serve webdav` at `/storage/sync`).
-pub fn local_sync_config(edge_key: &EdgeKey) -> Result<String> {
-    build_rclone_config(
-        LOCAL_REMOTE,
-        &[
-            ("type", "webdav".into()),
-            ("url", format!("{}/storage/sync", edge_key.server_url.trim_end_matches('/'))),
-            // rclone's own WebDAV extensions keep modification times: unchanged files are skipped.
-            ("vendor", "rclone".into()),
-            ("user", LOCAL_STORAGE_USER.into()),
-            ("pass", obscure_password(&local_storage_password(&edge_key.master_key_b64)?)?),
-        ],
-    )
-}
-
-async fn replicate(ctx: &CoreContext, cfg: &DatabaseConfig, storage: &DatabaseStorage, logger: &JobLogger) -> Result<SyncStats> {
-    if storage.provider == "local" {
-        let dest = format!("{LOCAL_REMOTE}:{}/current", cfg.generated_id);
-        return sync_dir(&local_sync_config(&ctx.edge_key)?, &dest, cfg, logger).await;
-    }
-    let target = rclone_target(storage)?;
-    let dest = replica_target(&target.remote_name, &target.base_path, folder(storage), &cfg.generated_id);
-    sync_dir(&target.config_text, &dest, cfg, logger).await
-}
-
-pub(crate) async fn one_storage(
-    ctx: &CoreContext,
-    cfg: &DatabaseConfig,
-    storage: &DatabaseStorage,
-    backup_id: &str,
-    logger: &JobLogger,
-) -> UploadResult {
-    let agent_id = ctx.edge_key.agent_id.clone();
-    let failed = |error: String| UploadResult {
-        storage_id: storage.id.clone(),
-        success: false,
-        error: Some(error),
-        remote_file_path: None,
-        total_size: None,
-    };
-
-    let backup_storage_id = match ctx
-        .api
-        .backup_upload_init(agent_id.clone(), cfg.generated_id.clone(), storage.id.clone(), backup_id, Some("sync"))
-        .await
-    {
-        Ok(Some(response)) => response.backup_storage.id,
-        Ok(None) => {
-            logger.log("error", "Upload init returned empty response");
-            return failed("backup_upload_init returned empty response".into());
-        }
-        Err(e) => {
-            logger.log("error", format!("Upload init failed: {e}"));
-            return failed("backup_upload_init failed".into());
-        }
-    };
-
-    match replicate(ctx, cfg, storage, logger).await {
-        Ok(stats) => {
-            let path = replica_path(storage, &cfg.generated_id);
-            match ctx
-                .api
-                .backup_upload_success(agent_id, cfg.generated_id.clone(), backup_storage_id, path.clone(), stats.bytes, stats.report(), backup_id)
-                .await
-            {
-                Ok(_) => UploadResult {
-                    storage_id: storage.id.clone(),
-                    success: true,
-                    error: None,
-                    remote_file_path: Some(path),
-                    // backups.file_size: the replica's size, like an archive's file size.
-                    total_size: stats.replica_bytes,
-                },
-                Err(e) => {
-                    logger.log("error", format!("Upload status update failed for {}: {e}", storage.id));
-                    failed(e.to_string())
-                }
-            }
-        }
-        Err(e) => {
-            logger.log("error", format!("Sync to storage {} failed: {e:#}", storage.id));
-            if let Err(err) = ctx
-                .api
-                .backup_upload_status(agent_id, cfg.generated_id.clone(), backup_storage_id, "failed", String::new(), 0u64, backup_id)
-                .await
-            {
-                logger.log("error", format!("Failed-status update failed for {}: {err}", storage.id));
-            }
-            failed(format!("{e:#}"))
-        }
-    }
-}

src/services/sync/mod.rs

@@ -1,6 +0,0 @@
-//! Sync mode: a destination kept identical to a files source (`rclone sync`).
-//! Replication, not backup: deletions propagate.
-pub mod backup;
-pub mod excludes;
-pub mod stats;
-pub mod rclone;

src/tests/domain/files/mod.rs

@@ -1,3 +1,6 @@
+mod restic;
+mod sync;
+
 use crate::domain::files::matcher::ExcludeMatcher;
 use std::path::Path;
 

src/tests/domain/files/restic.rs

@@ -1,13 +1,11 @@
-//! Snapshots mode (restic). Pure helpers first; the integration tests further
-//! down need `restic` and `rclone` on PATH.
-
 use crate::services::api::models::agent::status::DatabaseStorage;
 use crate::services::backup::logger::JobLogger;
-use crate::services::restic::command::is_snapshot_id;
-use crate::services::restic::excludes::{backup_excludes, glob_escape, restore_excludes};
-use crate::services::restic::json::{self, BackupSummary, Line};
-use crate::services::restic::password::{derive_password, hkdf_sha256_32};
+use crate::domain::files::restic::repo::is_snapshot_id;
+use crate::domain::files::restic::excludes::{backup_excludes, glob_escape, restore_excludes};
+use crate::domain::files::restic::json::{self, BackupSummary, Line};
+use crate::domain::files::restic::password::derive_password;
 use crate::tests::domain::files::files_config;
+use crate::utils::edge_key::hkdf_sha256_32;
 use serde_json::json;
 use std::path::Path;
 
@@ -34,7 +32,6 @@ fn hkdf_matches_rfc5869_case_3_first_block() {
 
 #[test]
 fn password_matches_the_dashboard_vector() {
-    // Same vector as src/lib/restic/repo.ts in the dashboard (spike S4).
     let password = derive_password(
         "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=",
         "test-generated-id",
@@ -83,7 +80,6 @@ fn restore_excludes_are_relative_to_the_snapshot_subfolder() {
 
 #[test]
 fn bracket_negation_is_rewritten_for_restic() {
-    // globset reads `[!a]`; restic (Go filepath.Match) only knows `[^a]`.
     assert_eq!(
         backup_excludes(Path::new("/data"), &strings(&["[!a]*.log", "/x/[!0-9]", r"\[!a]", "[a[!]"])).unwrap(),
         strings(&["/data/**/[^a]*.log", "/data/x/[^0-9]", r"/data/**/\[!a]", "/data/**/[a[!]"])
@@ -106,7 +102,6 @@ fn glob_escape_escapes_every_metacharacter() {
 
 #[test]
 fn json_lines_are_classified() {
-    // Recorded from restic 0.19.1 (paths shortened).
     let status = r#"{"message_type":"status","percent_done":0.9999997777778271,"total_files":4,"files_done":2,"total_bytes":9000002,"bytes_done":9000000,"error_count":1}"#;
     let error = r#"{"message_type":"error","error":{"message":"open /src/u: permission denied"},"during":"archival","item":"/src/u"}"#;
     let exit = r#"{"message_type":"exit_error","code":3,"message":"Warning: at least one source file could not be read"}"#;
@@ -137,7 +132,7 @@ fn backup_summary_deserializes_from_a_recorded_line() {
 
 #[test]
 fn removed_files_come_from_the_parent_summary() {
-    use crate::services::restic::json::{Snapshot, files_removed};
+    use crate::domain::files::restic::json::{Snapshot, files_removed};
     let listing: Vec<Snapshot> = serde_json::from_value(serde_json::json!([
         {"id": "p", "paths": ["/src"], "summary": {"total_files_processed": 3, "files_new": 3}},
         {"id": "c", "paths": ["/src"], "parent": "p"},
@@ -168,27 +163,13 @@ fn a_local_channel_opens_the_dashboard_rest_repository() {
     assert!(repo.is_append_only());
 }
 
-#[test]
-fn local_storage_password_matches_the_dashboard_vector() {
-    use crate::services::restic::password::local_storage_password;
-    // Same vector as src/lib/local-storage/credential.ts in the dashboard.
-    assert_eq!(
-        local_storage_password("AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=").unwrap(),
-        "8abfb788701b8c502658154746af6af768ea08f579381ad73b0f1e7dd43caea2"
-    );
-}
-
-// ---------- integration: restic + rclone on PATH ----------
-
 use crate::domain::files::platform::is_root;
-use crate::services::restic::backup::snapshot;
-use crate::services::restic::command::ResticRepo;
-use crate::services::restic::restore::restore;
+use crate::domain::files::restic::backup::snapshot;
+use crate::domain::files::restic::repo::ResticRepo;
+use crate::domain::files::restic::restore::restore;
 use std::fs;
 use tempfile::TempDir;
 
-/// A repository on an rclone `alias` remote pointing at a temp dir (test-only:
-/// production channels always go through `rclone_target`).
 fn local_repo(dir: &Path, generated_id: &str) -> ResticRepo {
     fs::create_dir_all(dir).unwrap();
     let config = format!("[pb]\ntype = alias\nremote = {}\n", dir.display());
@@ -204,7 +185,6 @@ fn tree(root: &Path) {
     fs::write(root.join("app.log"), "log").unwrap();
 }
 
-/// `relative/path=content` for every regular file, sorted.
 fn files(root: &Path) -> Vec<String> {
     let mut out: Vec<String> = walkdir::WalkDir::new(root)
         .min_depth(1)
@@ -242,8 +222,6 @@ async fn first_snapshot_initializes_the_repository_and_the_second_reuses_its_par
     assert_eq!(second.files_removed, Some(0));
 
     let tagged = repo.run(["snapshots", "--json", "--tag", "bs:bs-2"], &logger).await.unwrap();
-    // The tag selects only the second snapshot, and its `parent` is the first
-    // (a plain `contains(first)` check would match that `parent` field).
     let listed: serde_json::Value = serde_json::from_str(tagged.stdout.trim()).unwrap();
     let listed = listed.as_array().unwrap();
     assert_eq!(listed.len(), 1, "{}", tagged.stdout);
@@ -291,7 +269,7 @@ async fn restore_is_a_mirror_that_keeps_excluded_paths() {
 #[tokio::test]
 async fn unreadable_file_fails_the_snapshot_and_forgets_it() {
     if is_root() {
-        return; // root reads a 000 file anyway
+        return;
     }
     use std::os::unix::fs::PermissionsExt;
     let base = TempDir::new().unwrap();
@@ -313,7 +291,6 @@ async fn unreadable_file_fails_the_snapshot_and_forgets_it() {
 
 #[tokio::test]
 async fn unanchored_exclude_never_matches_directories_above_the_source() {
-    // Regression for spike S5: the source lives under a directory named like the pattern.
     let base = TempDir::new().unwrap();
     let src = base.path().join("tmp/src");
     fs::create_dir_all(src.join("tmp")).unwrap();
@@ -367,7 +344,7 @@ async fn an_unusable_cache_dir_runs_restic_without_cache() {
     let cfg = files_config(src.path(), &[]);
     let dir = base.path().join("repo");
     fs::create_dir_all(&dir).unwrap();
-    let blocker = tempfile::NamedTempFile::new().unwrap(); // a regular file: nothing can be created below it
+    let blocker = tempfile::NamedTempFile::new().unwrap();
     let config = format!("[pb]\ntype = alias\nremote = {}\n", dir.display());
     let repo = ResticRepo::new(&config, "pb", "", "backups", &cfg.generated_id, "test-password", blocker.path().join("cache")).unwrap();
     snapshot(&repo, &cfg, "bs-1", &JobLogger::new()).await.unwrap();

src/tests/domain/files/sync.rs

@@ -1,7 +1,5 @@
-//! Sync mode (rclone replica). Pure helpers first; integration tests need `rclone` on PATH.
-
-use crate::services::sync::excludes::sync_excludes;
-use crate::services::sync::stats::{SyncStats, parse_log};
+use crate::domain::files::sync::excludes::sync_excludes;
+use crate::domain::files::sync::stats::{SyncStats, parse_log};
 
 fn strings(items: &[&str]) -> Vec<String> {
     items.iter().map(|s| s.to_string()).collect()
@@ -109,11 +107,9 @@ fn sync_excludes_collapses_repeated_double_stars() {
     );
 }
 
-// ---------- integration: rclone on PATH, `alias` remote under a temp dir ----------
-
 use crate::domain::files::platform::is_root;
 use crate::services::backup::logger::JobLogger;
-use crate::services::sync::rclone::{replica_target, sync_dir};
+use crate::domain::files::sync::rclone::{replica_target, sync_dir};
 use crate::tests::domain::files::files_config;
 use std::fs;
 use std::path::{Path, PathBuf};
@@ -128,7 +124,6 @@ fn replica(store: &Path, generated_id: &str) -> PathBuf {
     store.join("backups/sync").join(generated_id).join("current")
 }
 
-/// `relative/path=content` for every regular file, sorted.
 fn files(root: &Path) -> Vec<String> {
     let mut out: Vec<String> = walkdir::WalkDir::new(root)
         .min_depth(1)
@@ -202,7 +197,7 @@ async fn a_later_exclude_removes_files_from_the_replica() {
 #[tokio::test]
 async fn failed_sync_keeps_files_the_failed_run_would_delete() {
     if is_root() {
-        return; // root reads a 000 file anyway
+        return;
     }
     use std::os::unix::fs::PermissionsExt;
     let store = TempDir::new().unwrap();
@@ -214,7 +209,6 @@ async fn failed_sync_keeps_files_the_failed_run_would_delete() {
 
     fs::remove_file(src.path().join("docs/b.txt")).unwrap();
     let locked = src.path().join("docs/a.txt");
-    // rclone only opens a file whose size/mtime changed, so the unreadable file must differ from the replica's copy.
     fs::write(&locked, "changed and now unreadable").unwrap();
     fs::set_permissions(&locked, fs::Permissions::from_mode(0o000)).unwrap();
     let err = sync_dir(&config, &replica_target("pb", "", "backups", &cfg.generated_id), &cfg, &JobLogger::new()).await.unwrap_err();
@@ -234,10 +228,8 @@ async fn empty_source_is_refused_and_the_replica_is_untouched() {
     assert!(!store.path().join("backups").exists());
 }
 
-// ---------- sync as a backup method ----------
-
 use crate::services::api::models::agent::status::DatabaseStorage;
-use crate::services::sync::backup::replica_path;
+use crate::domain::files::sync::backup::replica_path;
 use serde_json::json;
 
 #[test]
@@ -252,7 +244,7 @@ fn replica_path_uses_the_channel_folder() {
 
 #[test]
 fn local_sync_config_points_at_the_dashboard_webdav_server() {
-    use crate::services::sync::backup::local_sync_config;
+    use crate::domain::files::sync::backup::local_sync_config;
     use crate::utils::edge_key::EdgeKey;
     let edge_key = EdgeKey {
         server_url: "http://dash:8887/".into(),

src/tests/services/api_models_tests.rs

@@ -329,7 +329,7 @@ fn ping_payload_sends_method_only_when_set() {
 #[test]
 fn upload_status_sends_stats_only_when_set() {
     use crate::services::api::endpoints::agent::backup::upload::status::StatusUploadRequest;
-    use crate::services::sync::stats::SyncStats;
+    use crate::domain::files::sync::stats::SyncStats;
     let request = |stats: Option<serde_json::Value>| StatusUploadRequest {
         generated_id: "g".into(),
         backup_storage_id: "bs".into(),

src/tests/services/local_storage_tests.rs

@@ -4,9 +4,10 @@
 
 use crate::domain::files::platform::is_root;
 use crate::services::backup::logger::JobLogger;
-use crate::services::restic::backup::snapshot;
-use crate::services::restic::command::ResticRepo;
-use crate::services::restic::restore::restore;
+use crate::domain::files::restic::backup::snapshot;
+use crate::domain::files::restic::repo::ResticRepo;
+use crate::domain::files::restic::restore::restore;
+use crate::services::storage::providers::local::local_storage_password;
 use crate::tests::domain::files::files_config;
 use std::fs;
 use std::path::Path;
@@ -134,8 +135,7 @@ async fn local_sync_mirrors_through_the_webdav_server() {
     use crate::core::context::Context;
     use crate::services::api::ApiClient;
     use crate::services::api::models::agent::status::DatabaseStorage;
-    use crate::services::restic::password::local_storage_password;
-    use crate::services::sync::backup::one_storage;
+    use crate::services::backup::files_engine::{FilesEngine, one_storage};
     use crate::utils::edge_key::EdgeKey;
     use serde_json::json;
     use wiremock::matchers::{body_partial_json, method, path};
@@ -177,14 +177,22 @@ async fn local_sync_mirrors_through_the_webdav_server() {
     let cfg = files_config(&src, &[]);
     let replica = store.path().join(&cfg.generated_id).join("current");
 
-    let first = one_storage(&ctx, &cfg, &storage, "backup-1", &JobLogger::new()).await;
+    let first = one_storage(&ctx, &cfg, FilesEngine::Sync, &storage, "backup-1", &JobLogger::new()).await;
     assert!(first.success, "{:?}", first.error);
     assert_eq!(first.remote_file_path.as_deref(), Some(format!("backups/sync/{}/current", cfg.generated_id).as_str()));
     assert_eq!(fs::read_to_string(replica.join("docs/b.txt")).unwrap(), "b");
 
     fs::remove_file(src.join("docs/b.txt")).unwrap();
-    let second = one_storage(&ctx, &cfg, &storage, "backup-2", &JobLogger::new()).await;
+    let second = one_storage(&ctx, &cfg, FilesEngine::Sync, &storage, "backup-2", &JobLogger::new()).await;
     assert!(second.success, "{:?}", second.error);
     assert!(!replica.join("docs/b.txt").exists(), "deletion propagated");
     assert!(replica.join("a.txt").exists());
 }
+
+#[test]
+fn local_storage_password_matches_the_dashboard_vector() {
+    assert_eq!(
+        local_storage_password(MASTER_KEY_B64).unwrap(),
+        "8abfb788701b8c502658154746af6af768ea08f579381ad73b0f1e7dd43caea2"
+    );
+}

src/tests/services/mod.rs

@@ -5,6 +5,4 @@ mod config_tests;
 mod cron_tests;
 mod dashboard_config_tests;
 pub(crate) mod local_storage_tests;
-mod restic_tests;
 mod restore_downloader_tests;
-mod sync_tests;

src/tests/storage/rclone_target.rs

@@ -194,8 +194,8 @@ async fn gcs_emulator_target_reaches_fake_gcs_through_rclone() {
 #[tokio::test]
 async fn s3_target_hosts_a_restic_repository() {
     use crate::services::backup::logger::JobLogger;
-    use crate::services::restic::backup::snapshot;
-    use crate::services::restic::command::ResticRepo;
+    use crate::domain::files::restic::backup::snapshot;
+    use crate::domain::files::restic::repo::ResticRepo;
     use base64::{Engine as _, engine::general_purpose};
 
     let (_container, endpoint) = start_minio().await; // "http://host:port"
@@ -236,7 +236,7 @@ async fn s3_target_hosts_a_restic_repository() {
 #[tokio::test]
 async fn s3_targets_host_one_sync_replica_per_channel() {
     use crate::services::backup::logger::JobLogger;
-    use crate::services::sync::backup::one_storage;
+    use crate::services::backup::files_engine::{FilesEngine, one_storage};
     use crate::tests::services::backup_uploader_tests::ctx_pointing_at;
     use wiremock::matchers::{body_partial_json, method, path};
     use wiremock::{Mock, MockServer, ResponseTemplate};
@@ -284,7 +284,7 @@ async fn s3_targets_host_one_sync_replica_per_channel() {
     let ctx = ctx_pointing_at(server.uri());
 
     for c in &channels {
-        let result = one_storage(&ctx, &cfg, c, "backup-1", &JobLogger::new()).await;
+        let result = one_storage(&ctx, &cfg, FilesEngine::Sync, c, "backup-1", &JobLogger::new()).await;
         assert!(result.success, "{:?}", result.error);
         assert_eq!(result.remote_file_path.as_deref(), Some(format!("backups/sync/{}/current", cfg.generated_id).as_str()));
         assert_eq!(result.total_size, Some(2), "backups.file_size is the replica size");

src/utils/edge_key.rs

@@ -1,6 +1,9 @@
+use anyhow::Context;
 use base64::{Engine as _, engine::general_purpose};
+use hmac::{Hmac, Mac};
 use serde::Deserialize;
 use serde_json::Value;
+use sha2::Sha256;
 use thiserror::Error;
 
 #[derive(Debug, Deserialize)]
@@ -45,3 +48,21 @@ pub fn decode_edge_key(edge_key: &str) -> Result<EdgeKey, EdgeKeyError> {
         Err(EdgeKeyError::InvalidKey)
     }
 }
+
+pub fn derive_key(master_key_b64: &str, info: &[u8]) -> anyhow::Result<[u8; 32]> {
+    let master_key = general_purpose::STANDARD
+        .decode(master_key_b64.trim())
+        .context("master key is not valid base64")?;
+    Ok(hkdf_sha256_32(&master_key, info))
+}
+
+pub fn hkdf_sha256_32(ikm: &[u8], info: &[u8]) -> [u8; 32] {
+    let mut extract = Hmac::<Sha256>::new_from_slice(&[]).expect("HMAC takes any key length");
+    extract.update(ikm);
+    let prk = extract.finalize().into_bytes();
+
+    let mut expand = Hmac::<Sha256>::new_from_slice(&prk).expect("HMAC takes any key length");
+    expand.update(info);
+    expand.update(&[1]);
+    expand.finalize().into_bytes().into()
+}

src/utils/file.rs

@@ -46,14 +46,15 @@ pub fn full_file_name(encrypt: bool) -> String {
     }
 }
 
-pub fn full_file_path(file_name: &String, folder_name: Option<&str>) -> String {
-    let folder_name = folder_name
-        .map(str::trim)
-        .map(|folder_name| folder_name.trim_matches(char::from(47)))
+pub fn folder_or_default(folder_name: Option<&str>) -> &str {
+    folder_name
+        .map(|folder_name| folder_name.trim().trim_matches('/'))
         .filter(|folder_name| !folder_name.is_empty())
-        .unwrap_or("backups");
+        .unwrap_or("backups")
+}
 
-    format!("{}/{}/{}", folder_name, chrono::Local::now().format("%Y-%m-%d"), file_name)
+pub fn full_file_path(file_name: &String, folder_name: Option<&str>) -> String {
+    format!("{}/{}/{}", folder_or_default(folder_name), chrono::Local::now().format("%Y-%m-%d"), file_name)
 }
 
 const CHUNK_SIZE: usize = 16 * 1024 * 1024;