aven_core/
data_safety.rs

1use anyhow::{Context, Result, bail};
2use std::fs;
3use std::io::Write as _;
4use std::path::{Path, PathBuf};
5
6use crate::db::{self, Database};
7
8mod archive;
9pub(crate) mod export_types;
10mod import;
11mod integrity;
12mod scan;
13pub(crate) mod tables;
14pub(crate) mod validation;
15
16pub use export_types::{
17    AvenExport, BlobInventoryExportRow, ChangeRow, ConflictRow, ExportTables, FieldVersionRow,
18    LabelRow, MetaRow, MetadataFieldIdAliasRow, MetadataFieldRow, NoteRow, ProjectIdAliasRow,
19    ProjectPathRow, ProjectRow, RecurrenceOccurrenceRow, RecurrencePauseIntervalRow,
20    RecurrenceSeriesLabelRow, RecurrenceSeriesMetadataRow, RecurrenceSeriesRow,
21    SharedHistoryProvenanceRow, TaskAttachmentRow, TaskDependencyRow, TaskEpicLinkRow,
22    TaskLabelRow, TaskMetadataRow, TaskRelatedLinkRow, TaskRow, WorkspaceRow,
23};
24use export_types::{EXPORT_FORMAT, EXPORT_VERSION};
25
26#[derive(Debug, Clone)]
27pub struct IntegrityReport {
28    pub quick_check_ok: bool,
29    pub quick_check_value: String,
30    pub checks: Vec<IntegrityCheck>,
31}
32
33#[derive(Debug, Clone)]
34pub struct IntegrityCheck {
35    pub label: &'static str,
36    pub ok: bool,
37    pub value: String,
38}
39
40pub(crate) fn ensure_integrity_ok(report: &IntegrityReport) -> Result<()> {
41    integrity::ensure_ok(report)
42}
43
44pub(crate) async fn integrity_report_in_transaction(
45    conn: &mut sqlx::SqliteConnection,
46) -> Result<IntegrityReport> {
47    integrity::database_report(conn).await
48}
49
50pub(crate) async fn scan_export_tables(conn: &mut sqlx::SqliteConnection) -> Result<ExportTables> {
51    Ok(ExportTables {
52        workspaces: scan::scan_workspaces(conn).await?,
53        projects: scan::scan_projects(conn).await?,
54        project_paths: scan::scan_project_paths(conn).await?,
55        project_id_aliases: scan::scan_project_id_aliases(conn).await?,
56        labels: scan::scan_labels(conn).await?,
57        metadata_fields: scan::scan_metadata_fields(conn).await?,
58        metadata_field_id_aliases: scan::scan_metadata_field_id_aliases(conn).await?,
59        tasks: scan::scan_tasks(conn).await?,
60        task_metadata: scan::scan_task_metadata(conn).await?,
61        task_labels: scan::scan_task_labels(conn).await?,
62        notes: scan::scan_notes(conn).await?,
63        task_dependencies: scan::scan_task_dependencies(conn).await?,
64        task_epic_links: scan::scan_task_epic_links(conn).await?,
65        task_related_links: scan::scan_task_related_links(conn).await?,
66        task_attachments: scan::scan_task_attachments(conn).await?,
67        blob_inventory: scan::scan_blob_inventory(conn).await?,
68        recurrence_series: scan::scan_recurrence_series(conn).await?,
69        recurrence_series_labels: scan::scan_recurrence_series_labels(conn).await?,
70        recurrence_series_metadata: scan::scan_recurrence_series_metadata(conn).await?,
71        recurrence_occurrences: scan::scan_recurrence_occurrences(conn).await?,
72        recurrence_pause_intervals: scan::scan_recurrence_pause_intervals(conn).await?,
73        changes: scan::scan_changes(conn).await?,
74        shared_history_provenance: scan::scan_shared_history_provenance(conn).await?,
75        field_versions: scan::scan_field_versions(conn).await?,
76        conflicts: scan::scan_conflicts(conn).await?,
77        meta: scan::scan_meta(conn).await?,
78    })
79}
80
81/// Writes a portable export as an owner-only file, atomically replacing an
82/// existing regular file while refusing any other existing file type.
83pub fn write_export_file(output: &Path, bytes: &[u8]) -> Result<()> {
84    if let Some(parent) = output
85        .parent()
86        .filter(|parent| !parent.as_os_str().is_empty())
87    {
88        crate::private_fs::create_dir_all(parent)
89            .with_context(|| format!("could not create {}", parent.display()))?;
90    }
91    match fs::symlink_metadata(output) {
92        Ok(metadata) if !metadata.file_type().is_file() => {
93            bail!("error export-output-not-regular path={}", output.display());
94        }
95        Ok(_) => {}
96        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
97        Err(error) => {
98            return Err(error).with_context(|| {
99                format!("could not inspect export destination {}", output.display())
100            });
101        }
102    }
103
104    let mut temp = crate::private_fs::sibling_tempfile(output)
105        .with_context(|| format!("could not create temporary file for {}", output.display()))?;
106    temp.write_all(bytes)
107        .with_context(|| format!("could not write temporary export file {}", output.display()))?;
108    temp.persist(output)
109        .map(|_| ())
110        .map_err(|error| error.error)
111        .with_context(|| format!("could not replace export file {}", output.display()))
112}
113
114impl Database {
115    pub async fn export_data(&self, exported_at: String) -> Result<AvenExport> {
116        let mut conn = self.acquire_writer().await?;
117        let mut tx = db::begin_immediate(&mut conn).await?;
118        let schema_version = db::current_schema_version(&mut tx).await?;
119        let mut tables = scan_export_tables(&mut tx).await?;
120        let bound: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM local_seed_source) OR EXISTS(SELECT 1 FROM local_peer_enrollment)")
121            .fetch_one(&mut *tx)
122            .await?;
123        if bound {
124            tables.meta.retain(|m| {
125                m.key != "sync_server_url"
126                    && m.key != "e2ee_association"
127                    && m.key != "e2ee_data_only"
128            });
129            tables.meta.push(MetaRow {
130                key: "e2ee_data_only".into(),
131                value: "1".into(),
132            });
133        }
134        let version = if !tables.shared_history_provenance.is_empty() {
135            EXPORT_VERSION
136        } else if tables.task_related_links.is_empty() {
137            2
138        } else {
139            export_types::RELATED_LINKS_EXPORT_VERSION
140        };
141        tx.commit().await?;
142        Ok(AvenExport {
143            format: EXPORT_FORMAT.to_string(),
144            version,
145            exported_at,
146            schema_version,
147            blobs_included: false,
148            tables,
149        })
150    }
151
152    pub async fn validate_import_data(&self, export: &AvenExport) -> Result<()> {
153        let mut conn = self.acquire_reader().await?;
154        validation::ensure_supported_export(&mut conn, export).await?;
155        validation::validate_export_snapshot(export)
156    }
157
158    pub async fn import_data(&self, export: &AvenExport) -> Result<IntegrityReport> {
159        let _installation = self.import_installation_guard()?;
160        let mut conn = self.acquire_writer().await?;
161        validation::ensure_supported_export(&mut conn, export).await?;
162        validation::validate_export_snapshot(export)?;
163        let target_client_id = db::get_meta(&mut conn, "client_id")
164            .await?
165            .context("missing target client_id")?;
166        let mut tx = db::begin_immediate(&mut conn).await?;
167        crate::sync::shared_state::ensure_no_active_local_shared_capture(&mut tx).await?;
168        import::replace_from_export(&mut tx, export, &target_client_id).await?;
169        let report = integrity::database_report(&mut tx).await?;
170        ensure_integrity_ok(&report)?;
171        tx.commit().await?;
172        Ok(report)
173    }
174
175    pub async fn database_integrity_report(&self) -> Result<IntegrityReport> {
176        let mut conn = self.acquire_reader().await?;
177        integrity::database_report(&mut conn).await
178    }
179
180    pub async fn attachment_integrity_checks(
181        &self,
182        blob_dir: &Path,
183        deep: bool,
184    ) -> Result<Vec<IntegrityCheck>> {
185        let mut conn = self.acquire_reader().await?;
186        integrity::attachment_integrity_checks(&mut conn, blob_dir, deep).await
187    }
188
189    pub async fn create_backup_archive(&self, blob_dir: &Path, output: &Path) -> Result<()> {
190        let _installation = self.backup_installation_guard()?;
191        let mut conn = self.acquire_writer().await?;
192        let hashes: Vec<String> = sqlx::query_scalar(
193            "SELECT sha256 FROM blob_inventory WHERE available = 1 ORDER BY sha256",
194        )
195        .fetch_all(&mut *conn)
196        .await?;
197        let mut leases = Vec::with_capacity(hashes.len());
198        for hash in hashes {
199            match crate::attachments::lifecycle::acquire_lease(
200                &mut conn,
201                &hash,
202                "backup",
203                &crate::attachments::lifecycle::SystemClock,
204            )
205            .await
206            {
207                Ok(lease) => leases.push(lease),
208                Err(error) => {
209                    for lease in leases {
210                        let _ =
211                            crate::attachments::lifecycle::release_lease(&mut conn, &lease).await;
212                    }
213                    return Err(error);
214                }
215            }
216        }
217        let backup_result = archive::create_backup_archive(&mut conn, blob_dir, output).await;
218        for lease in leases {
219            crate::attachments::lifecycle::release_lease(&mut conn, &lease).await?;
220        }
221        backup_result
222    }
223}
224
225pub fn is_backup_archive(path: &Path) -> Result<bool> {
226    archive::is_archive_path(path)
227}
228
229pub async fn restore_backup_archive(
230    db_path: &Path,
231    blob_dir: &Path,
232    source: &Path,
233) -> Result<PathBuf> {
234    archive::restore_backup_archive(db_path, blob_dir, source).await
235}
236
237#[cfg(test)]
238mod tests {
239    use super::write_export_file;
240    use std::fs;
241
242    #[cfg(unix)]
243    #[test]
244    fn export_file_is_owner_only_under_permissive_umask() {
245        use crate::private_fs::test_umask::{Umask, mode};
246
247        let _umask = Umask::set(0o022);
248        let temp = tempfile::tempdir().unwrap();
249        let output = temp.path().join("new-parent").join("export.json");
250        write_export_file(&output, b"{}\n").unwrap();
251
252        assert_eq!(fs::read(&output).unwrap(), b"{}\n");
253        assert_eq!(mode(&output), 0o600);
254        assert_eq!(mode(output.parent().unwrap()), 0o700);
255    }
256
257    #[cfg(unix)]
258    #[test]
259    fn export_file_replaces_existing_regular_file_and_remains_owner_only() {
260        use std::os::unix::fs::PermissionsExt;
261
262        use crate::private_fs::test_umask::{Umask, mode};
263
264        let _umask = Umask::set(0o022);
265        let temp = tempfile::tempdir().unwrap();
266        let output = temp.path().join("export.json");
267        write_export_file(&output, b"first export").unwrap();
268        fs::set_permissions(&output, fs::Permissions::from_mode(0o644)).unwrap();
269
270        write_export_file(&output, b"second export").unwrap();
271
272        assert_eq!(fs::read(&output).unwrap(), b"second export");
273        assert_eq!(mode(&output), 0o600);
274    }
275
276    #[cfg(unix)]
277    #[test]
278    fn export_file_refuses_symlink_without_following_it() {
279        use std::os::unix::fs::symlink;
280
281        let temp = tempfile::tempdir().unwrap();
282        let target = temp.path().join("victim");
283        let output = temp.path().join("export.json");
284        fs::write(&target, b"sentinel").unwrap();
285        symlink(&target, &output).unwrap();
286
287        assert!(write_export_file(&output, b"export").is_err());
288        assert_eq!(fs::read(&target).unwrap(), b"sentinel");
289        assert!(
290            fs::symlink_metadata(&output)
291                .unwrap()
292                .file_type()
293                .is_symlink()
294        );
295    }
296}
aven_core/data_safety.rs

aven_core/
data_safety.rs
