1use crate::choices::TaskSource;
2use crate::ids::{ProjectId, TaskId, WorkspaceId};
3use anyhow::{Context, Result, bail};
4use serde::{Deserialize, Serialize};
5use sqlx::{SqliteConnection, query_scalar};
6use std::collections::{HashMap, HashSet};
7use std::path::{Path, PathBuf};
8
9mod archive;
10mod integrity;
11mod tables;
12use crate::db::{self, Database};
13
14#[derive(Debug, Clone)]
15pub struct IntegrityReport {
16 pub quick_check_ok: bool,
17 pub quick_check_value: String,
18 pub checks: Vec<IntegrityCheck>,
19}
20
21#[derive(Debug, Clone)]
22pub struct IntegrityCheck {
23 pub label: &'static str,
24 pub ok: bool,
25 pub value: String,
26}
27
28#[derive(Debug, Serialize, Deserialize)]
29pub struct AvenExport {
30 pub format: String,
31 pub version: i64,
32 pub exported_at: String,
33 pub schema_version: i64,
34 #[serde(default)]
35 pub blobs_included: bool,
36 pub tables: ExportTables,
37}
38
39#[derive(Debug, Serialize, Deserialize)]
40pub struct ExportTables {
41 pub workspaces: Vec<WorkspaceRow>,
42 pub projects: Vec<ProjectRow>,
43 pub project_paths: Vec<ProjectPathRow>,
44 pub project_id_aliases: Vec<ProjectIdAliasRow>,
45 pub labels: Vec<LabelRow>,
46 pub tasks: Vec<TaskRow>,
47 pub task_labels: Vec<TaskLabelRow>,
48 pub notes: Vec<NoteRow>,
49 pub task_dependencies: Vec<TaskDependencyRow>,
50 pub task_epic_links: Vec<TaskEpicLinkRow>,
51 #[serde(default)]
52 pub task_attachments: Vec<TaskAttachmentRow>,
53 #[serde(default)]
54 pub blob_inventory: Vec<BlobInventoryExportRow>,
55 pub changes: Vec<ChangeRow>,
56 pub field_versions: Vec<FieldVersionRow>,
57 pub conflicts: Vec<ConflictRow>,
58 pub meta: Vec<MetaRow>,
59}
60
61#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
62pub struct WorkspaceRow {
63 pub id: WorkspaceId,
64 pub name: String,
65 pub key: String,
66 pub created_at: String,
67 pub updated_at: String,
68 pub archived: i64,
69}
70
71#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
72pub struct ProjectRow {
73 pub id: ProjectId,
74 pub workspace_id: WorkspaceId,
75 pub key: String,
76 pub name: String,
77 pub prefix: String,
78 pub created_at: String,
79 pub updated_at: String,
80 pub deleted: i64,
81}
82
83#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
84pub struct ProjectPathRow {
85 pub workspace_id: WorkspaceId,
86 pub project_id: ProjectId,
87 pub path: String,
88}
89
90#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
91pub struct ProjectIdAliasRow {
92 pub workspace_id: WorkspaceId,
93 pub remote_project_id: ProjectId,
94 pub local_project_id: ProjectId,
95}
96
97#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
98pub struct LabelRow {
99 pub workspace_id: WorkspaceId,
100 pub name: String,
101 pub created_at: String,
102}
103
104fn default_task_source() -> String {
105 TaskSource::Unknown.as_str().to_string()
106}
107
108#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
109pub struct TaskRow {
110 pub workspace_id: WorkspaceId,
111 pub id: TaskId,
112 pub title: String,
113 pub description: String,
114 pub project_id: ProjectId,
115 pub status: String,
116 pub priority: String,
117 #[serde(default = "default_task_source")]
118 pub source: String,
119 pub created_at: String,
120 pub updated_at: String,
121 pub queue_activity_at: String,
122 #[serde(default)]
123 pub available_at: String,
124 #[serde(default)]
125 pub due_on: String,
126 pub deleted: i64,
127 pub is_epic: i64,
128}
129
130#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
131pub struct TaskEpicLinkRow {
132 pub workspace_id: WorkspaceId,
133 pub child_task_id: TaskId,
134 pub epic_task_id: TaskId,
135 pub created_at: String,
136}
137
138#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
139pub struct TaskLabelRow {
140 pub workspace_id: WorkspaceId,
141 pub task_id: TaskId,
142 pub label: String,
143}
144
145#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
146pub struct NoteRow {
147 pub workspace_id: WorkspaceId,
148 pub id: String,
149 pub task_id: TaskId,
150 pub body: String,
151 pub created_at: String,
152 pub change_id: String,
153}
154
155#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
156pub struct TaskDependencyRow {
157 pub workspace_id: WorkspaceId,
158 pub task_id: TaskId,
159 pub depends_on_task_id: TaskId,
160 pub created_at: String,
161}
162
163#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
164pub struct TaskAttachmentRow {
165 pub workspace_id: WorkspaceId,
166 pub attachment_id: String,
167 pub task_id: TaskId,
168 pub sha256: String,
169 pub byte_size: i64,
170 pub media_type: String,
171 pub filename: Option<String>,
172 pub alt_text: Option<String>,
173 pub width: Option<i64>,
174 pub height: Option<i64>,
175 pub created_at: String,
176 pub created_by_change_id: Option<String>,
177 pub deleted: i64,
178 pub deleted_at: Option<String>,
179 pub deleted_by_change_id: Option<String>,
180}
181
182#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
183pub struct BlobInventoryExportRow {
184 pub sha256: String,
185 pub byte_size: i64,
186 pub media_type: String,
187 pub available: i64,
188 pub first_seen_at: String,
189 pub last_verified_at: Option<String>,
190}
191
192#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
193pub struct ChangeRow {
194 pub change_id: String,
195 pub client_id: String,
196 pub local_seq: i64,
197 pub entity_type: String,
198 pub entity_id: String,
199 pub field: Option<String>,
200 pub op_type: String,
201 pub payload: String,
202 pub base_version: Option<String>,
203 pub created_at: String,
204 pub server_seq: Option<i64>,
205}
206
207#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
208pub struct FieldVersionRow {
209 pub entity_id: String,
210 pub field: String,
211 pub version: String,
212}
213
214#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
215pub struct ConflictRow {
216 pub id: i64,
217 pub workspace_id: WorkspaceId,
218 pub task_id: TaskId,
219 pub field: String,
220 pub base_version: Option<String>,
221 pub local_value: String,
222 pub remote_value: String,
223 pub local_change_id: Option<String>,
224 pub remote_change_id: String,
225 pub variant_a: String,
226 pub variant_b: String,
227 pub created_at: String,
228 pub resolved: i64,
229}
230
231#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
232pub struct MetaRow {
233 pub key: String,
234 pub value: String,
235}
236
237impl Database {
238 pub async fn export_data(&self, exported_at: String) -> Result<AvenExport> {
239 let mut conn = self.acquire().await?;
240 let schema_version = db::current_schema_version(&mut conn).await?;
241 Ok(AvenExport {
242 format: "aven-export".to_string(),
243 version: 1,
244 exported_at,
245 schema_version,
246 blobs_included: false,
247 tables: ExportTables {
248 workspaces: scan_workspaces(&mut conn).await?,
249 projects: scan_projects(&mut conn).await?,
250 project_paths: scan_project_paths(&mut conn).await?,
251 project_id_aliases: scan_project_id_aliases(&mut conn).await?,
252 labels: scan_labels(&mut conn).await?,
253 tasks: scan_tasks(&mut conn).await?,
254 task_labels: scan_task_labels(&mut conn).await?,
255 notes: scan_notes(&mut conn).await?,
256 task_dependencies: scan_task_dependencies(&mut conn).await?,
257 task_epic_links: scan_task_epic_links(&mut conn).await?,
258 task_attachments: scan_task_attachments(&mut conn).await?,
259 blob_inventory: scan_blob_inventory(&mut conn).await?,
260 changes: scan_changes(&mut conn).await?,
261 field_versions: scan_field_versions(&mut conn).await?,
262 conflicts: scan_conflicts(&mut conn).await?,
263 meta: scan_meta(&mut conn).await?,
264 },
265 })
266 }
267
268 pub async fn validate_import_data(&self, export: &AvenExport) -> Result<()> {
269 let mut conn = self.acquire().await?;
270 ensure_supported_export(&mut conn, export).await?;
271 validate_export_snapshot(export)
272 }
273
274 pub async fn import_data(&self, export: &AvenExport) -> Result<IntegrityReport> {
275 let mut conn = self.acquire().await?;
276 ensure_supported_export(&mut conn, export).await?;
277 validate_export_snapshot(export)?;
278 let target_client_id = db::get_meta(&mut conn, "client_id")
279 .await?
280 .context("missing target client_id")?;
281 let mut tx = db::begin_immediate(&mut conn).await?;
282 replace_from_export(&mut tx, export, &target_client_id).await?;
283 let report = database_integrity_report_with_connection(&mut tx).await?;
284 ensure_integrity_ok(&report)?;
285 tx.commit().await?;
286 Ok(report)
287 }
288
289 pub async fn database_integrity_report(&self) -> Result<IntegrityReport> {
290 let mut conn = self.acquire().await?;
291 database_integrity_report_with_connection(&mut conn).await
292 }
293
294 pub async fn attachment_integrity_checks(
295 &self,
296 blob_dir: &Path,
297 deep: bool,
298 ) -> Result<Vec<IntegrityCheck>> {
299 let mut conn = self.acquire().await?;
300 integrity::attachment_integrity_checks(&mut conn, blob_dir, deep).await
301 }
302
303 pub async fn create_backup_archive(
304 &self,
305 db_path: &Path,
306 blob_dir: &Path,
307 output: &Path,
308 ) -> Result<()> {
309 let mut conn = self.acquire().await?;
310 let hashes: Vec<String> = sqlx::query_scalar(
311 "SELECT sha256 FROM blob_inventory WHERE available = 1 ORDER BY sha256",
312 )
313 .fetch_all(&mut *conn)
314 .await?;
315 let mut leases = Vec::with_capacity(hashes.len());
316 for hash in hashes {
317 match crate::attachments::lifecycle::acquire_lease(
318 &mut conn,
319 &hash,
320 "backup",
321 &crate::attachments::lifecycle::SystemClock,
322 )
323 .await
324 {
325 Ok(lease) => leases.push(lease),
326 Err(error) => {
327 for lease in leases {
328 let _ =
329 crate::attachments::lifecycle::release_lease(&mut conn, &lease).await;
330 }
331 return Err(error);
332 }
333 }
334 }
335 let backup_result =
336 archive::create_backup_archive(&mut conn, db_path, blob_dir, output).await;
337 for lease in leases {
338 crate::attachments::lifecycle::release_lease(&mut conn, &lease).await?;
339 }
340 backup_result
341 }
342}
343
344pub fn is_backup_archive(path: &Path) -> Result<bool> {
345 archive::is_archive_path(path)
346}
347
348pub async fn restore_backup_archive(
349 db_path: &Path,
350 blob_dir: &Path,
351 source: &Path,
352) -> Result<PathBuf> {
353 archive::restore_backup_archive(db_path, blob_dir, source).await
354}
355
356async fn scan_workspaces(conn: &mut SqliteConnection) -> Result<Vec<WorkspaceRow>> {
357 tables::scan_rows(
358 conn,
359 "SELECT id, name, key, created_at, updated_at, archived FROM workspaces",
360 )
361 .await
362}
363
364async fn scan_projects(conn: &mut SqliteConnection) -> Result<Vec<ProjectRow>> {
365 tables::scan_rows(
366 conn,
367 "SELECT id, workspace_id, key, name, prefix, created_at, updated_at, deleted FROM projects",
368 )
369 .await
370}
371
372async fn scan_project_paths(conn: &mut SqliteConnection) -> Result<Vec<ProjectPathRow>> {
373 tables::scan_rows(
374 conn,
375 "SELECT workspace_id, project_id, path FROM project_paths",
376 )
377 .await
378}
379
380async fn scan_project_id_aliases(conn: &mut SqliteConnection) -> Result<Vec<ProjectIdAliasRow>> {
381 tables::scan_rows(
382 conn,
383 "SELECT workspace_id, remote_project_id, local_project_id FROM project_id_aliases",
384 )
385 .await
386}
387
388async fn scan_labels(conn: &mut SqliteConnection) -> Result<Vec<LabelRow>> {
389 tables::scan_rows(conn, "SELECT workspace_id, name, created_at FROM labels").await
390}
391
392async fn scan_tasks(conn: &mut SqliteConnection) -> Result<Vec<TaskRow>> {
393 tables::scan_rows(conn, "SELECT workspace_id, id, title, description, project_id, status, priority, source, created_at, updated_at, queue_activity_at, available_at, due_on, deleted, is_epic FROM tasks").await
394}
395
396async fn scan_task_labels(conn: &mut SqliteConnection) -> Result<Vec<TaskLabelRow>> {
397 tables::scan_rows(conn, "SELECT workspace_id, task_id, label FROM task_labels").await
398}
399
400async fn scan_notes(conn: &mut SqliteConnection) -> Result<Vec<NoteRow>> {
401 tables::scan_rows(
402 conn,
403 "SELECT workspace_id, id, task_id, body, created_at, change_id FROM notes",
404 )
405 .await
406}
407
408async fn scan_task_dependencies(conn: &mut SqliteConnection) -> Result<Vec<TaskDependencyRow>> {
409 tables::scan_rows(
410 conn,
411 "SELECT workspace_id, task_id, depends_on_task_id, created_at FROM task_dependencies",
412 )
413 .await
414}
415
416async fn scan_task_epic_links(conn: &mut SqliteConnection) -> Result<Vec<TaskEpicLinkRow>> {
417 tables::scan_rows(
418 conn,
419 "SELECT workspace_id, child_task_id, epic_task_id, created_at FROM task_epic_links",
420 )
421 .await
422}
423
424async fn scan_task_attachments(conn: &mut SqliteConnection) -> Result<Vec<TaskAttachmentRow>> {
425 tables::scan_rows(
426 conn,
427 "SELECT workspace_id, attachment_id, task_id, sha256, byte_size, media_type, filename, alt_text, width, height, created_at, created_by_change_id, deleted, deleted_at, deleted_by_change_id FROM task_attachments",
428 )
429 .await
430}
431
432async fn scan_blob_inventory(conn: &mut SqliteConnection) -> Result<Vec<BlobInventoryExportRow>> {
433 tables::scan_rows(
434 conn,
435 "SELECT sha256, byte_size, media_type, available, first_seen_at, last_verified_at FROM blob_inventory",
436 )
437 .await
438}
439
440async fn scan_changes(conn: &mut SqliteConnection) -> Result<Vec<ChangeRow>> {
441 tables::scan_rows(conn, "SELECT change_id, client_id, local_seq, entity_type, entity_id, field, op_type, payload, base_version, created_at, server_seq FROM changes").await
442}
443
444async fn scan_field_versions(conn: &mut SqliteConnection) -> Result<Vec<FieldVersionRow>> {
445 tables::scan_rows(conn, "SELECT entity_id, field, version FROM field_versions").await
446}
447
448async fn scan_conflicts(conn: &mut SqliteConnection) -> Result<Vec<ConflictRow>> {
449 tables::scan_rows(conn, "SELECT id, workspace_id, task_id, field, base_version, local_value, remote_value, local_change_id, remote_change_id, variant_a, variant_b, created_at, resolved FROM conflicts").await
450}
451
452async fn scan_meta(conn: &mut SqliteConnection) -> Result<Vec<MetaRow>> {
453 tables::scan_rows(conn, "SELECT key, value FROM meta").await
454}
455
456async fn ensure_supported_export(conn: &mut SqliteConnection, export: &AvenExport) -> Result<()> {
457 if export.format != "aven-export" {
458 bail!("error export-format-unsupported format={}", export.format);
459 }
460 if export.version != 1 {
461 bail!(
462 "error export-version-unsupported version={}",
463 export.version
464 );
465 }
466 let current = db::current_schema_version(conn).await?;
467 if export.schema_version != current {
468 bail!(
469 "error export-schema-unsupported expected={} actual={}",
470 current,
471 export.schema_version
472 );
473 }
474 Ok(())
475}
476
477fn validate_export_snapshot(export: &AvenExport) -> Result<()> {
478 let mut workspace_ids = HashSet::new();
479 for workspace in &export.tables.workspaces {
480 if workspace_ids.contains(&workspace.id) {
481 continue;
482 }
483 workspace_ids.insert(workspace.id.clone());
484 }
485
486 let mut project_ids: HashMap<WorkspaceId, HashSet<ProjectId>> = HashMap::new();
487 for project in &export.tables.projects {
488 if !workspace_ids.contains(&project.workspace_id) {
489 bail!(
490 "error invalid-export-snapshot project.workspace_id={} is missing",
491 project.workspace_id
492 );
493 }
494 project_ids
495 .entry(project.workspace_id.clone())
496 .or_default()
497 .insert(project.id.clone());
498 }
499
500 for path in &export.tables.project_paths {
501 let projects = project_ids.get(&path.workspace_id).ok_or_else(|| {
502 anyhow::Error::msg(format!(
503 "error invalid-export-snapshot project_path.workspace_id={} is missing",
504 path.workspace_id
505 ))
506 })?;
507 if !projects.contains(&path.project_id) {
508 bail!(
509 "error invalid-export-snapshot project_path.project_id={} is missing in workspace {}",
510 path.project_id,
511 path.workspace_id
512 );
513 }
514 }
515
516 let mut label_keys: HashSet<(WorkspaceId, String)> = HashSet::new();
517 for label in &export.tables.labels {
518 if !workspace_ids.contains(&label.workspace_id) {
519 bail!(
520 "error invalid-export-snapshot label.workspace_id={} is missing",
521 label.workspace_id
522 );
523 }
524 label_keys.insert((label.workspace_id.clone(), label.name.clone()));
525 }
526
527 let mut task_ids: HashMap<WorkspaceId, HashSet<TaskId>> = HashMap::new();
528 for task in &export.tables.tasks {
529 TaskSource::parse(&task.source)?;
530 if let Err(error) = crate::time_validation::validate_due_on_value(&task.due_on) {
531 bail!(
532 "error invalid-export-snapshot task.due_on={} is invalid: {error}",
533 task.due_on
534 );
535 }
536 let workspace_projects = project_ids.get(&task.workspace_id).ok_or_else(|| {
537 anyhow::Error::msg(format!(
538 "error invalid-export-snapshot task.workspace_id={} is missing",
539 task.workspace_id
540 ))
541 })?;
542 if !workspace_projects.contains(&task.project_id) {
543 bail!(
544 "error invalid-export-snapshot task.project_id={} is missing in workspace {}",
545 task.project_id,
546 task.workspace_id
547 );
548 }
549 task_ids
550 .entry(task.workspace_id.clone())
551 .or_default()
552 .insert(task.id.clone());
553 }
554
555 for task_label in &export.tables.task_labels {
556 let task_workspace = task_ids.get(&task_label.workspace_id).ok_or_else(|| {
557 anyhow::Error::msg(format!(
558 "error invalid-export-snapshot task_label.workspace_id={} is missing",
559 task_label.workspace_id
560 ))
561 })?;
562 if !task_workspace.contains(&task_label.task_id) {
563 bail!(
564 "error invalid-export-snapshot task_label.task_id={} is missing in workspace {}",
565 task_label.task_id,
566 task_label.workspace_id
567 );
568 }
569 if !label_keys.contains(&(task_label.workspace_id.clone(), task_label.label.clone())) {
570 bail!(
571 "error invalid-export-snapshot task_label.label={} is missing in workspace {}",
572 task_label.label,
573 task_label.workspace_id
574 );
575 }
576 }
577
578 for note in &export.tables.notes {
579 let task_workspace = task_ids.get(¬e.workspace_id).ok_or_else(|| {
580 anyhow::Error::msg(format!(
581 "error invalid-export-snapshot note.workspace_id={} is missing",
582 note.workspace_id
583 ))
584 })?;
585 if !task_workspace.contains(¬e.task_id) {
586 bail!(
587 "error invalid-export-snapshot note.task_id={} is missing in workspace {}",
588 note.task_id,
589 note.workspace_id
590 );
591 }
592 }
593
594 for dep in &export.tables.task_dependencies {
595 let tasks = task_ids.get(&dep.workspace_id).ok_or_else(|| {
596 anyhow::Error::msg(format!(
597 "error invalid-export-snapshot dependency.workspace_id={} is missing",
598 dep.workspace_id
599 ))
600 })?;
601 if !tasks.contains(&dep.task_id) || !tasks.contains(&dep.depends_on_task_id) {
602 bail!(
603 "error invalid-export-snapshot task_dependencies are missing tasks in workspace {}",
604 dep.workspace_id
605 );
606 }
607 }
608
609 for epic_link in &export.tables.task_epic_links {
610 let tasks = task_ids.get(&epic_link.workspace_id).ok_or_else(|| {
611 anyhow::Error::msg(format!(
612 "error invalid-export-snapshot epic_link.workspace_id={} is missing",
613 epic_link.workspace_id
614 ))
615 })?;
616 if !tasks.contains(&epic_link.child_task_id) || !tasks.contains(&epic_link.epic_task_id) {
617 bail!(
618 "error invalid-export-snapshot task_epic_links are missing tasks in workspace {}",
619 epic_link.workspace_id
620 );
621 }
622 }
623
624 let mut inventory = HashMap::new();
625 for blob in &export.tables.blob_inventory {
626 crate::attachments::validate_sha256(&blob.sha256)?;
627 crate::attachments::validate_media_type(&blob.media_type)?;
628 crate::attachments::validate_blob_size(usize::try_from(blob.byte_size).unwrap_or(0))?;
629 if blob.available != 0 && blob.available != 1 {
630 bail!("error invalid-export-snapshot blob_inventory.available invalid");
631 }
632 if inventory
633 .insert(
634 blob.sha256.clone(),
635 (blob.byte_size, blob.media_type.as_str()),
636 )
637 .is_some()
638 {
639 bail!("error invalid-export-snapshot blob_inventory.sha256 duplicate");
640 }
641 }
642
643 for attachment in &export.tables.task_attachments {
644 crate::attachments::validate_attachment_id(&attachment.attachment_id)?;
645 crate::attachments::validate_sha256(&attachment.sha256)?;
646 crate::attachments::validate_media_type(&attachment.media_type)?;
647 crate::attachments::validate_blob_size(usize::try_from(attachment.byte_size).unwrap_or(0))?;
648 crate::attachments::validate_filename(attachment.filename.as_deref())?;
649 crate::attachments::validate_alt_text(attachment.alt_text.as_deref())?;
650 crate::attachments::validate_dimensions(attachment.width, attachment.height)?;
651 let tasks = task_ids.get(&attachment.workspace_id).ok_or_else(|| {
652 anyhow::Error::msg(format!(
653 "error invalid-export-snapshot attachment.workspace_id={} is missing",
654 attachment.workspace_id
655 ))
656 })?;
657 if !tasks.contains(&attachment.task_id) {
658 bail!(
659 "error invalid-export-snapshot attachment.task_id={} is missing in workspace {}",
660 attachment.task_id,
661 attachment.workspace_id
662 );
663 }
664 let Some((inventory_size, inventory_media_type)) = inventory.get(&attachment.sha256) else {
665 bail!("error invalid-export-snapshot attachment inventory missing");
666 };
667 if *inventory_size != attachment.byte_size || *inventory_media_type != attachment.media_type
668 {
669 bail!("error invalid-export-snapshot attachment inventory metadata mismatch");
670 }
671 if attachment.deleted != 0 && attachment.deleted != 1 {
672 bail!(
673 "error invalid-export-snapshot attachment.deleted={} for attachment {}",
674 attachment.deleted,
675 attachment.attachment_id
676 );
677 }
678 }
679
680 for alias in &export.tables.project_id_aliases {
681 let workspace_projects = project_ids.get(&alias.workspace_id).ok_or_else(|| {
682 anyhow::Error::msg(format!(
683 "error invalid-export-snapshot project_alias.workspace_id={} is missing",
684 alias.workspace_id
685 ))
686 })?;
687 if !workspace_projects.contains(&alias.local_project_id) {
688 bail!(
689 "error invalid-export-snapshot local_project_id={} is missing in workspace {}",
690 alias.local_project_id,
691 alias.workspace_id
692 );
693 }
694 }
695
696 Ok(())
697}
698
699async fn replace_from_export(
700 tx: &mut SqliteConnection,
701 export: &AvenExport,
702 target_client_id: &str,
703) -> Result<()> {
704 let delete_order = [
705 "DELETE FROM task_attachments",
706 "DELETE FROM blob_inventory",
707 "DELETE FROM task_epic_links",
708 "DELETE FROM task_dependencies",
709 "DELETE FROM task_labels",
710 "DELETE FROM notes",
711 "DELETE FROM conflicts",
712 "DELETE FROM field_versions",
713 "DELETE FROM changes",
714 "DELETE FROM project_paths",
715 "DELETE FROM project_id_aliases",
716 "DELETE FROM tasks",
717 "DELETE FROM labels",
718 "DELETE FROM projects",
719 "DELETE FROM workspaces",
720 "DELETE FROM meta",
721 ];
722 for sql in delete_order {
723 sqlx::query(sql).execute(&mut *tx).await?;
724 }
725
726 db::set_meta(tx, "client_id", target_client_id).await?;
727 db::set_meta(tx, "sync_cursor", "0").await?;
728 let local_seq = export
729 .tables
730 .changes
731 .iter()
732 .map(|row| row.local_seq)
733 .max()
734 .unwrap_or(0);
735 db::set_meta(tx, "local_seq", &local_seq.to_string()).await?;
736
737 for meta in &export.tables.meta {
738 if matches!(
739 meta.key.as_str(),
740 "client_id" | "sync_server_url" | "sync_cursor" | "local_seq"
741 ) {
742 continue;
743 }
744 db::set_meta(tx, &meta.key, &meta.value).await?;
745 }
746
747 let suppressed_attachment_changes = export
748 .tables
749 .changes
750 .iter()
751 .filter(|change| {
752 change.server_seq.is_none()
753 && change.field.as_deref() == Some("attachments")
754 && matches!(
755 change.op_type.as_str(),
756 "attachment_add" | "attachment_delete"
757 )
758 })
759 .map(|change| change.change_id.as_str())
760 .collect::<HashSet<_>>();
761 let mut attachments = export.tables.task_attachments.clone();
762 for attachment in &mut attachments {
763 if attachment
764 .created_by_change_id
765 .as_deref()
766 .is_some_and(|id| suppressed_attachment_changes.contains(id))
767 {
768 attachment.created_by_change_id = None;
769 }
770 if attachment
771 .deleted_by_change_id
772 .as_deref()
773 .is_some_and(|id| suppressed_attachment_changes.contains(id))
774 {
775 attachment.deleted_by_change_id = None;
776 }
777 }
778 let changes = export
779 .tables
780 .changes
781 .iter()
782 .filter(|change| !suppressed_attachment_changes.contains(change.change_id.as_str()))
783 .cloned()
784 .collect::<Vec<_>>();
785
786 tables::import_workspaces(tx, &export.tables.workspaces).await?;
787 tables::import_projects(tx, &export.tables.projects).await?;
788 tables::import_project_id_aliases(tx, &export.tables.project_id_aliases).await?;
789 tables::import_project_paths(tx, &export.tables.project_paths).await?;
790 tables::import_labels(tx, &export.tables.labels).await?;
791 tables::import_tasks(tx, &export.tables.tasks).await?;
792 tables::import_task_labels(tx, &export.tables.task_labels).await?;
793 tables::import_notes(tx, &export.tables.notes).await?;
794 tables::import_task_dependencies(tx, &export.tables.task_dependencies).await?;
795 tables::import_task_epic_links(tx, &export.tables.task_epic_links).await?;
796 tables::import_blob_inventory(tx, &export.tables.blob_inventory).await?;
797 tables::import_task_attachments(tx, &attachments).await?;
798 tables::import_changes(tx, &changes).await?;
799 tables::import_field_versions(tx, &export.tables.field_versions).await?;
800 tables::import_conflicts(tx, &export.tables.conflicts).await?;
801
802 Ok(())
803}
804
805async fn database_integrity_report_with_connection(
806 conn: &mut SqliteConnection,
807) -> Result<IntegrityReport> {
808 let quick_check_value: String = query_scalar("PRAGMA quick_check")
809 .fetch_one(&mut *conn)
810 .await?;
811 let mut checks = Vec::new();
812 checks.push(count_check(
813 conn,
814 "task projects",
815 "SELECT count(*) FROM tasks t LEFT JOIN projects p ON p.workspace_id = t.workspace_id AND p.id = t.project_id WHERE p.id IS NULL",
816 )
817 .await?);
818 checks.push(count_check(
819 conn,
820 "project paths",
821 "SELECT count(*) FROM project_paths pp LEFT JOIN projects p ON p.workspace_id = pp.workspace_id AND p.id = pp.project_id WHERE p.id IS NULL",
822 )
823 .await?);
824 checks.push(count_check(
825 conn,
826 "project aliases",
827 "SELECT count(*) FROM project_id_aliases a LEFT JOIN projects p ON p.workspace_id = a.workspace_id AND p.id = a.local_project_id WHERE p.id IS NULL",
828 )
829 .await?);
830 checks.push(count_check(
831 conn,
832 "task label tasks",
833 "SELECT count(*) FROM task_labels tl LEFT JOIN tasks t ON t.workspace_id = tl.workspace_id AND t.id = tl.task_id WHERE t.id IS NULL",
834 )
835 .await?);
836 checks.push(count_check(
837 conn,
838 "task label labels",
839 "SELECT count(*) FROM task_labels tl LEFT JOIN labels l ON l.workspace_id = tl.workspace_id AND l.name = tl.label WHERE l.name IS NULL",
840 )
841 .await?);
842 checks.push(count_check(
843 conn,
844 "notes",
845 "SELECT count(*) FROM notes n LEFT JOIN tasks t ON t.workspace_id = n.workspace_id AND t.id = n.task_id WHERE t.id IS NULL",
846 )
847 .await?);
848 checks.push(count_check(
849 conn,
850 "note changes",
851 "SELECT count(*) FROM notes n LEFT JOIN changes c ON c.change_id = n.change_id WHERE c.change_id IS NULL",
852 )
853 .await?);
854 checks.push(count_check(
855 conn,
856 "dependency tasks",
857 "SELECT count(*) FROM task_dependencies d LEFT JOIN tasks t ON t.workspace_id = d.workspace_id AND t.id = d.task_id WHERE t.id IS NULL",
858 )
859 .await?);
860 checks.push(count_check(
861 conn,
862 "dependency targets",
863 "SELECT count(*) FROM task_dependencies d LEFT JOIN tasks t ON t.workspace_id = d.workspace_id AND t.id = d.depends_on_task_id WHERE t.id IS NULL",
864 )
865 .await?);
866 checks.push(count_check(
867 conn,
868 "epic link children",
869 "SELECT count(*) FROM task_epic_links l LEFT JOIN tasks t ON t.workspace_id = l.workspace_id AND t.id = l.child_task_id WHERE t.id IS NULL",
870 )
871 .await?);
872 checks.push(count_check(
873 conn,
874 "epic link parents",
875 "SELECT count(*) FROM task_epic_links l LEFT JOIN tasks t ON t.workspace_id = l.workspace_id AND t.id = l.epic_task_id WHERE t.id IS NULL",
876 )
877 .await?);
878 checks.push(count_check(
879 conn,
880 "epic link parent flags",
881 "SELECT count(*) FROM task_epic_links l JOIN tasks t ON t.workspace_id = l.workspace_id AND t.id = l.epic_task_id WHERE t.is_epic = 0",
882 )
883 .await?);
884 checks.push(count_check(
885 conn,
886 "conflict tasks",
887 "SELECT count(*) FROM conflicts c LEFT JOIN tasks t ON t.workspace_id = c.workspace_id AND t.id = c.task_id WHERE c.resolved = 0 AND t.id IS NULL",
888 )
889 .await?);
890 checks.push(count_check(
891 conn,
892 "task due dates",
893 "SELECT count(*) FROM tasks WHERE due_on != '' AND (length(due_on) != 10 OR substr(due_on, 5, 1) != '-' OR substr(due_on, 8, 1) != '-' OR date(due_on) IS NULL OR strftime('%Y-%m-%d', due_on) != due_on)",
894 )
895 .await?);
896 checks.push(count_check(
897 conn,
898 "field version tasks",
899 "SELECT count(*) FROM field_versions fv LEFT JOIN tasks t ON t.id = fv.entity_id WHERE t.id IS NULL AND fv.field IN ('title','description','status','priority','project','labels','available_at','due_on','deleted','is_epic')",
900 )
901 .await?);
902 checks.push(count_check(
903 conn,
904 "field version changes",
905 "SELECT count(*) FROM field_versions fv LEFT JOIN changes c ON c.change_id = fv.version WHERE c.change_id IS NULL",
906 )
907 .await?);
908 push_meta_checks(conn, &mut checks).await?;
909
910 Ok(IntegrityReport {
911 quick_check_ok: quick_check_value == "ok",
912 quick_check_value,
913 checks,
914 })
915}
916
917pub(crate) fn ensure_integrity_ok(report: &IntegrityReport) -> Result<()> {
918 let mut bad = vec![];
919 if !report.quick_check_ok {
920 bad.push("quick check");
921 }
922 for check in &report.checks {
923 if !check.ok {
924 bad.push(check.label);
925 }
926 }
927 if bad.is_empty() {
928 return Ok(());
929 }
930 bail!("error data-integrity-failed checks={}", bad.join(", "))
931}
932
933async fn count_check(
934 conn: &mut SqliteConnection,
935 label: &'static str,
936 query: &'static str,
937) -> Result<IntegrityCheck> {
938 let count: i64 = query_scalar(query).fetch_one(&mut *conn).await?;
939 Ok(IntegrityCheck {
940 label,
941 ok: count == 0,
942 value: format!("{count} orphaned"),
943 })
944}
945
946async fn push_meta_checks(
947 conn: &mut SqliteConnection,
948 checks: &mut Vec<IntegrityCheck>,
949) -> Result<()> {
950 let local_seq = db::get_meta(conn, "local_seq").await?;
951 let local_seq_check = match local_seq {
952 Some(raw) => match raw.parse::<i64>() {
953 Ok(value) => {
954 let max_seq: i64 = query_scalar("SELECT COALESCE(MAX(local_seq), 0) FROM changes")
955 .fetch_one(&mut *conn)
956 .await?;
957 let ok = value >= max_seq;
958 IntegrityCheck {
959 label: "meta local_seq",
960 ok,
961 value: value.to_string(),
962 }
963 }
964 Err(error) => IntegrityCheck {
965 label: "meta local_seq",
966 ok: false,
967 value: error.to_string(),
968 },
969 },
970 None => IntegrityCheck {
971 label: "meta local_seq",
972 ok: false,
973 value: "missing".to_string(),
974 },
975 };
976 checks.push(local_seq_check);
977
978 let sync_cursor = db::get_meta(conn, "sync_cursor").await?;
979 let sync_cursor_ok = match sync_cursor {
980 Some(raw) => match raw.parse::<i64>() {
981 Ok(_) => IntegrityCheck {
982 label: "sync cursor",
983 ok: true,
984 value: raw,
985 },
986 Err(error) => IntegrityCheck {
987 label: "sync cursor",
988 ok: false,
989 value: error.to_string(),
990 },
991 },
992 None => IntegrityCheck {
993 label: "sync cursor",
994 ok: false,
995 value: "missing".to_string(),
996 },
997 };
998 checks.push(sync_cursor_ok);
999
1000 Ok(())
1001}