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