aven_core/
data_safety.rs

1use anyhow::{Context, Result};
2use std::path::{Path, PathBuf};
3
4use crate::db::{self, Database};
5
6mod archive;
7mod export_types;
8mod import;
9mod integrity;
10mod scan;
11mod tables;
12mod validation;
13
14pub use export_types::{
15    AvenExport, BlobInventoryExportRow, ChangeRow, ConflictRow, ExportTables, FieldVersionRow,
16    LabelRow, MetaRow, MetadataFieldIdAliasRow, MetadataFieldRow, NoteRow, ProjectIdAliasRow,
17    ProjectPathRow, ProjectRow, RecurrenceOccurrenceRow, RecurrencePauseIntervalRow,
18    RecurrenceSeriesLabelRow, RecurrenceSeriesMetadataRow, RecurrenceSeriesRow, TaskAttachmentRow,
19    TaskDependencyRow, TaskEpicLinkRow, TaskLabelRow, TaskMetadataRow, TaskRelatedLinkRow, TaskRow,
20    WorkspaceRow,
21};
22use export_types::{EXPORT_FORMAT, EXPORT_VERSION};
23
24#[derive(Debug, Clone)]
25pub struct IntegrityReport {
26    pub quick_check_ok: bool,
27    pub quick_check_value: String,
28    pub checks: Vec<IntegrityCheck>,
29}
30
31#[derive(Debug, Clone)]
32pub struct IntegrityCheck {
33    pub label: &'static str,
34    pub ok: bool,
35    pub value: String,
36}
37
38pub(crate) fn ensure_integrity_ok(report: &IntegrityReport) -> Result<()> {
39    integrity::ensure_ok(report)
40}
41
42impl Database {
43    pub async fn export_data(&self, exported_at: String) -> Result<AvenExport> {
44        let mut conn = self.acquire_writer().await?;
45        let mut tx = db::begin_immediate(&mut conn).await?;
46        let schema_version = db::current_schema_version(&mut tx).await?;
47        let tables = ExportTables {
48            workspaces: scan::scan_workspaces(&mut tx).await?,
49            projects: scan::scan_projects(&mut tx).await?,
50            project_paths: scan::scan_project_paths(&mut tx).await?,
51            project_id_aliases: scan::scan_project_id_aliases(&mut tx).await?,
52            labels: scan::scan_labels(&mut tx).await?,
53            metadata_fields: scan::scan_metadata_fields(&mut tx).await?,
54            metadata_field_id_aliases: scan::scan_metadata_field_id_aliases(&mut tx).await?,
55            tasks: scan::scan_tasks(&mut tx).await?,
56            task_metadata: scan::scan_task_metadata(&mut tx).await?,
57            task_labels: scan::scan_task_labels(&mut tx).await?,
58            notes: scan::scan_notes(&mut tx).await?,
59            task_dependencies: scan::scan_task_dependencies(&mut tx).await?,
60            task_epic_links: scan::scan_task_epic_links(&mut tx).await?,
61            task_related_links: scan::scan_task_related_links(&mut tx).await?,
62            task_attachments: scan::scan_task_attachments(&mut tx).await?,
63            blob_inventory: scan::scan_blob_inventory(&mut tx).await?,
64            recurrence_series: scan::scan_recurrence_series(&mut tx).await?,
65            recurrence_series_labels: scan::scan_recurrence_series_labels(&mut tx).await?,
66            recurrence_series_metadata: scan::scan_recurrence_series_metadata(&mut tx).await?,
67            recurrence_occurrences: scan::scan_recurrence_occurrences(&mut tx).await?,
68            recurrence_pause_intervals: scan::scan_recurrence_pause_intervals(&mut tx).await?,
69            changes: scan::scan_changes(&mut tx).await?,
70            field_versions: scan::scan_field_versions(&mut tx).await?,
71            conflicts: scan::scan_conflicts(&mut tx).await?,
72            meta: scan::scan_meta(&mut tx).await?,
73        };
74        let version = if tables.task_related_links.is_empty() {
75            2
76        } else {
77            EXPORT_VERSION
78        };
79        tx.commit().await?;
80        Ok(AvenExport {
81            format: EXPORT_FORMAT.to_string(),
82            version,
83            exported_at,
84            schema_version,
85            blobs_included: false,
86            tables,
87        })
88    }
89
90    pub async fn validate_import_data(&self, export: &AvenExport) -> Result<()> {
91        let mut conn = self.acquire_reader().await?;
92        validation::ensure_supported_export(&mut conn, export).await?;
93        validation::validate_export_snapshot(export)
94    }
95
96    pub async fn import_data(&self, export: &AvenExport) -> Result<IntegrityReport> {
97        let mut conn = self.acquire_writer().await?;
98        validation::ensure_supported_export(&mut conn, export).await?;
99        validation::validate_export_snapshot(export)?;
100        let target_client_id = db::get_meta(&mut conn, "client_id")
101            .await?
102            .context("missing target client_id")?;
103        let mut tx = db::begin_immediate(&mut conn).await?;
104        import::replace_from_export(&mut tx, export, &target_client_id).await?;
105        let report = integrity::database_report(&mut tx).await?;
106        ensure_integrity_ok(&report)?;
107        tx.commit().await?;
108        Ok(report)
109    }
110
111    pub async fn database_integrity_report(&self) -> Result<IntegrityReport> {
112        let mut conn = self.acquire_reader().await?;
113        integrity::database_report(&mut conn).await
114    }
115
116    pub async fn attachment_integrity_checks(
117        &self,
118        blob_dir: &Path,
119        deep: bool,
120    ) -> Result<Vec<IntegrityCheck>> {
121        let mut conn = self.acquire_reader().await?;
122        integrity::attachment_integrity_checks(&mut conn, blob_dir, deep).await
123    }
124
125    pub async fn create_backup_archive(&self, blob_dir: &Path, output: &Path) -> Result<()> {
126        let mut conn = self.acquire_writer().await?;
127        let hashes: Vec<String> = sqlx::query_scalar(
128            "SELECT sha256 FROM blob_inventory WHERE available = 1 ORDER BY sha256",
129        )
130        .fetch_all(&mut *conn)
131        .await?;
132        let mut leases = Vec::with_capacity(hashes.len());
133        for hash in hashes {
134            match crate::attachments::lifecycle::acquire_lease(
135                &mut conn,
136                &hash,
137                "backup",
138                &crate::attachments::lifecycle::SystemClock,
139            )
140            .await
141            {
142                Ok(lease) => leases.push(lease),
143                Err(error) => {
144                    for lease in leases {
145                        let _ =
146                            crate::attachments::lifecycle::release_lease(&mut conn, &lease).await;
147                    }
148                    return Err(error);
149                }
150            }
151        }
152        let backup_result = archive::create_backup_archive(&mut conn, blob_dir, output).await;
153        for lease in leases {
154            crate::attachments::lifecycle::release_lease(&mut conn, &lease).await?;
155        }
156        backup_result
157    }
158}
159
160pub fn is_backup_archive(path: &Path) -> Result<bool> {
161    archive::is_archive_path(path)
162}
163
164pub async fn restore_backup_archive(
165    db_path: &Path,
166    blob_dir: &Path,
167    source: &Path,
168) -> Result<PathBuf> {
169    archive::restore_backup_archive(db_path, blob_dir, source).await
170}
aven_core/data_safety.rs

aven_core/
data_safety.rs
