feat: mirror the diary to Proton Drive as a device

The data directory is the whole diary, and until now keeping a copy of it
was someone's manual job. The server can now register itself in Proton
Drive as a device — its own sync root, next to the desktop clients — and
keep that folder equal to DIARY_DATA_DIR, encrypted client-side by the
same Rust SDK the Linux client uses.

One interactive `narl-diary proton-login`, then nothing prompts again.
The stored session carries the tokens, the mailbox password (the key
chain is rebuilt from it on every start) and the account key salts, which
have to be captured at login: `core/v4/keys/salts` needs the `locked`
scope, and only a password-login token has it. Refresh tokens are
single-use, so every rotation is written back before it can be needed.
It lives as a 0600 file next to the database, on the same volume as the
data it protects; `--features keyring` puts it in the OS keyring instead,
for hosts that have a Secret Service.

The mirror is one-way and change-driven: writes mark the diary dirty and
it is mirrored once the writing stops, with an interval as the backstop —
an hour of writing is one backup, not sixty. `diary.db` goes up as a
VACUUM INTO snapshot, so it is a real point in time rather than a file
copied out from under a writer, and as a new revision each time, so
Proton keeps the older ones. Uploads are immutable, so they are written
once and skipped by size afterwards. Deletions do not propagate unless
asked: a backup that forgets on command is one accident away from being
no backup at all.

Identification follows Proton's third-party rules — an honest
`external-drive-narl_diary@<version>-alpha`, the disclosure printed
before any credential is asked for, and remembered folder uids instead of
walking the tree every run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01D5Gk8FPkRXu1e6Nj2BCRE1

src/backup/sync.rs

@@ -0,0 +1,388 @@
+//! The mirror itself: make the device folder look like `DIARY_DATA_DIR`.
+//!
+//! The remote layout is deliberately the local one, not a rendering of it:
+//!
+//! ```text
+//! narl-diary/            the device's sync root
+//!   RESTORE.txt          what this is and how to put it back
+//!   diary.db             a consistent snapshot, one revision per change
+//!   uploads/<uuid>       every uploaded file, under the name the database knows
+//! ```
+//!
+//! so restoring is copying two things into an empty data directory, with no
+//! tool in between. The readable-anywhere view of the diary is what `:export!`
+//! is for; this is the copy that can be *restored*.
+
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context, Result};
+use proton_drive_rs::ProtonDriveClient;
+use proton_sdk::ids::{LinkId, NodeUid, VolumeId};
+use sha2::{Digest, Sha256};
+use sqlx::{Row, SqlitePool};
+
+use crate::config::Config;
+
+/// What one run did, for the log line and the status endpoint.
+#[derive(Debug, Default, Clone, Copy, serde::Serialize)]
+pub struct Summary {
+    pub uploaded: usize,
+    pub skipped: usize,
+    pub pruned: usize,
+    pub bytes: u64,
+}
+
+const RESTORE: &str = "\
+This is a Proton Drive mirror of a ~/diary server.
+
+  diary.db          the whole diary: entries, media metadata, share tokens
+  uploads/<uuid>    every uploaded file; diary.db gives each one its real name
+
+To restore, stop the diary, empty its data directory, and copy both back:
+
+  DIARY_DATA_DIR/diary.db
+  DIARY_DATA_DIR/uploads/
+
+The database is a snapshot taken with `VACUUM INTO`, so it is internally
+consistent and carries no write-ahead log: it can be opened directly.
+
+Each file above keeps its history — Proton Drive holds the previous revisions
+of diary.db, so an accident that was mirrored can still be undone.
+";
+
+/// Mirror everything that has changed. Returns what it did.
+pub async fn run(
+    db: &SqlitePool,
+    config: &Config,
+    client: &ProtonDriveClient,
+    root: &NodeUid,
+    uploads_folder: &NodeUid,
+) -> Result<Summary> {
+    let mut mirror = Mirror {
+        db,
+        client,
+        summary: Summary::default(),
+    };
+
+    // Written once, and only ever re-read by a human who has lost everything
+    // else — so it goes up before the data it explains.
+    mirror
+        .bytes(root, "RESTORE.txt", "RESTORE.txt", "text/plain", RESTORE.as_bytes())
+        .await?;
+
+    mirror.database(config, root).await?;
+    mirror.media(config, uploads_folder).await?;
+
+    if config.backup.prune {
+        mirror.prune().await?;
+    }
+
+    Ok(mirror.summary)
+}
+
+struct Mirror<'a> {
+    db: &'a SqlitePool,
+    client: &'a ProtonDriveClient,
+    summary: Summary,
+}
+
+impl Mirror<'_> {
+    /// Snapshot the database and mirror it.
+    ///
+    /// `VACUUM INTO` is the only honest way to copy a live SQLite database: it
+    /// takes a read transaction, so the copy is a single point in time with the
+    /// write-ahead log already folded in. Copying the file by hand while the
+    /// diary is being written to would produce something that may or may not
+    /// open.
+    async fn database(&mut self, config: &Config, root: &NodeUid) -> Result<()> {
+        let snapshot = config.data_dir.join(".diary-backup.db");
+        // A snapshot left behind by a killed run would fail the next one:
+        // VACUUM INTO refuses to write over an existing file.
+        let _ = tokio::fs::remove_file(&snapshot).await;
+
+        sqlx::query("VACUUM INTO ?")
+            .bind(snapshot.to_string_lossy().as_ref())
+            .execute(self.db)
+            .await
+            .context("could not snapshot the database")?;
+
+        let result = self
+            .file(root, "diary.db", "diary.db", "application/vnd.sqlite3", &snapshot)
+            .await;
+
+        let _ = tokio::fs::remove_file(&snapshot).await;
+        result
+    }
+
+    /// Mirror every uploaded file that is not up there yet.
+    ///
+    /// Uploads are immutable — a uuid names one set of bytes forever — so a
+    /// file already recorded at the same size is not read, let alone hashed. A
+    /// diary with a decade of photographs in it must not re-read the lot every
+    /// hour.
+    async fn media(&mut self, config: &Config, uploads_folder: &NodeUid) -> Result<()> {
+        let media: Vec<(String, i64)> =
+            sqlx::query("SELECT id, size FROM media ORDER BY created_at")
+                .fetch_all(self.db)
+                .await?
+                .into_iter()
+                .map(|row| (row.get("id"), row.get("size")))
+                .collect();
+
+        for (id, size) in media {
+            let key = format!("uploads/{id}");
+            if let Some(tracked) = self.tracked(&key).await? {
+                if tracked.size == size {
+                    self.summary.skipped += 1;
+                    continue;
+                }
+            }
+
+            let path = config.uploads_dir().join(&id);
+            if !path.exists() {
+                // A row whose file has gone missing is a local inconsistency,
+                // not a reason to abandon the rest of the backup.
+                tracing::warn!(media = %id, "media row has no file on disk; skipping");
+                continue;
+            }
+
+            self.file(uploads_folder, &key, &id, "application/octet-stream", &path)
+                .await?;
+        }
+
+        Ok(())
+    }
+
+    /// Trash remote files whose local rows are gone. Off unless asked for.
+    async fn prune(&mut self) -> Result<()> {
+        let orphans: Vec<(String, String)> = sqlx::query(
+            "SELECT path, node_uid FROM backup_files
+              WHERE path LIKE 'uploads/%'
+                AND substr(path, 9) NOT IN (SELECT id FROM media)",
+        )
+        .fetch_all(self.db)
+        .await?
+        .into_iter()
+        .map(|row| (row.get("path"), row.get("node_uid")))
+        .collect();
+
+        if orphans.is_empty() {
+            return Ok(());
+        }
+
+        let uids: Vec<NodeUid> = orphans.iter().filter_map(|(_, uid)| parse_uid(uid)).collect();
+        for (uid, result) in self.client.trash_nodes(&uids).await? {
+            if let Err(e) = result {
+                tracing::warn!(node = %uid, error = %e, "could not trash a pruned file");
+            }
+        }
+
+        for (path, _) in &orphans {
+            sqlx::query("DELETE FROM backup_files WHERE path = ?")
+                .bind(path)
+                .execute(self.db)
+                .await?;
+            self.summary.pruned += 1;
+        }
+
+        Ok(())
+    }
+
+    /// Mirror bytes held in memory, through a temporary file so the upload path
+    /// is the same one everything else takes.
+    async fn bytes(
+        &mut self,
+        parent: &NodeUid,
+        key: &str,
+        name: &str,
+        mime: &str,
+        bytes: &[u8],
+    ) -> Result<()> {
+        let hash = hex::encode(Sha256::digest(bytes));
+        if self.is_current(key, &hash).await? {
+            return Ok(());
+        }
+
+        let temporary = tempfile::NamedTempFile::new()?;
+        tokio::fs::write(temporary.path(), bytes).await?;
+        self.upload(parent, key, name, mime, temporary.path(), &hash)
+            .await
+    }
+
+    /// Hash a local file and mirror it if the remote copy is not already it.
+    async fn file(
+        &mut self,
+        parent: &NodeUid,
+        key: &str,
+        name: &str,
+        mime: &str,
+        path: &Path,
+    ) -> Result<()> {
+        let hash = hash_file(path.to_path_buf()).await?;
+        if self.is_current(key, &hash).await? {
+            return Ok(());
+        }
+
+        self.upload(parent, key, name, mime, path, &hash).await
+    }
+
+    /// Upload `path`, as a new revision of the node already tracked for `key`
+    /// when there is one, and as a new file otherwise.
+    ///
+    /// A tracked node that has been deleted in the Drive UI is not an error:
+    /// the mirror notices the node is gone and re-creates the file.
+    async fn upload(
+        &mut self,
+        parent: &NodeUid,
+        key: &str,
+        name: &str,
+        mime: &str,
+        path: &Path,
+        hash: &str,
+    ) -> Result<()> {
+        let size = tokio::fs::metadata(path).await?.len();
+        let modified = crate::now();
+
+        let existing = match self.tracked(key).await? {
+            Some(tracked) => self
+                .client
+                .get_node(&tracked.uid)
+                .await
+                .ok()
+                .flatten()
+                .map(|_| tracked.uid),
+            None => None,
+        };
+
+        let uid = match existing {
+            Some(uid) => {
+                let file = std::fs::File::open(path)?;
+                self.client
+                    .upload_new_revision_from(&uid, file, size as i64, Vec::new(), Some(modified))
+                    .await
+                    .with_context(|| format!("could not upload a new revision of {name}"))?;
+                uid
+            }
+            None => {
+                let file = std::fs::File::open(path)?;
+                self.client
+                    .upload_file_from(
+                        parent,
+                        name,
+                        mime,
+                        file,
+                        size as i64,
+                        Vec::new(),
+                        Some(modified),
+                        false,
+                    )
+                    .await
+                    .with_context(|| format!("could not upload {name}"))?
+            }
+        };
+
+        self.record(key, &uid, hash, size as i64).await?;
+        self.summary.uploaded += 1;
+        self.summary.bytes += size;
+        tracing::info!(file = key, bytes = size, "mirrored to Proton Drive");
+        Ok(())
+    }
+
+    /// Whether the remote copy of `key` is already these bytes.
+    async fn is_current(&mut self, key: &str, hash: &str) -> Result<bool> {
+        let current = self
+            .tracked(key)
+            .await?
+            .is_some_and(|tracked| tracked.hash == hash);
+        if current {
+            self.summary.skipped += 1;
+        }
+        Ok(current)
+    }
+
+    async fn tracked(&self, path: &str) -> Result<Option<Tracked>> {
+        let row = sqlx::query("SELECT node_uid, hash, size FROM backup_files WHERE path = ?")
+            .bind(path)
+            .fetch_optional(self.db)
+            .await?;
+
+        Ok(row.and_then(|row| {
+            let uid: String = row.get("node_uid");
+            parse_uid(&uid).map(|uid| Tracked {
+                uid,
+                hash: row.get("hash"),
+                size: row.get("size"),
+            })
+        }))
+    }
+
+    async fn record(&self, path: &str, uid: &NodeUid, hash: &str, size: i64) -> Result<()> {
+        sqlx::query(
+            "INSERT INTO backup_files (path, node_uid, hash, size, uploaded_at)
+             VALUES (?, ?, ?, ?, ?)
+             ON CONFLICT (path) DO UPDATE SET
+                 node_uid = excluded.node_uid,
+                 hash = excluded.hash,
+                 size = excluded.size,
+                 uploaded_at = excluded.uploaded_at",
+        )
+        .bind(path)
+        .bind(uid.to_string())
+        .bind(hash)
+        .bind(size)
+        .bind(crate::now())
+        .execute(self.db)
+        .await?;
+        Ok(())
+    }
+}
+
+struct Tracked {
+    uid: NodeUid,
+    hash: String,
+    size: i64,
+}
+
+/// `<volume>~<link>`, the form [`NodeUid`] prints. Anything else is a row this
+/// build did not write and is treated as untracked rather than trusted.
+pub(crate) fn parse_uid(text: &str) -> Option<NodeUid> {
+    let (volume, link) = text.split_once('~')?;
+    if volume.is_empty() || link.is_empty() {
+        return None;
+    }
+    Some(NodeUid::new(VolumeId::new(volume), LinkId::new(link)))
+}
+
+/// Streaming sha256 of a file, on the blocking pool: the database snapshot is
+/// as large as the diary, and photographs are larger.
+async fn hash_file(path: PathBuf) -> Result<String> {
+    tokio::task::spawn_blocking(move || -> Result<String> {
+        let mut file = std::fs::File::open(&path)
+            .with_context(|| format!("could not read {}", path.display()))?;
+        let mut hasher = Sha256::new();
+        std::io::copy(&mut file, &mut hasher)?;
+        Ok(hex::encode(hasher.finalize()))
+    })
+    .await?
+}
+
+#[cfg(test)]
+mod tests {
+    use super::parse_uid;
+
+    #[test]
+    fn node_uids_survive_a_round_trip() {
+        let uid = parse_uid("vol-1~link-2").expect("a well-formed uid parses");
+        assert_eq!(uid.volume_id.as_str(), "vol-1");
+        assert_eq!(uid.link_id.as_str(), "link-2");
+        assert_eq!(uid.to_string(), "vol-1~link-2");
+    }
+
+    #[test]
+    fn a_row_this_build_did_not_write_is_untracked() {
+        assert!(parse_uid("").is_none());
+        assert!(parse_uid("no-separator").is_none());
+        assert!(parse_uid("~link").is_none());
+        assert!(parse_uid("volume~").is_none());
+    }
+}