1use crate::choices::{TaskPriority, TaskSource, TaskStatus};
2use crate::ids::{ProjectId, TaskId, WorkspaceId};
3use crate::recurrence::{
4 RecurrenceDuePolicy, RecurrenceFrequency, RecurrenceOutcome, RecurrenceProjectionState,
5 RecurrenceRule, RecurrenceSchedule, RecurrenceSeriesId, RecurrenceSeriesState, TimeZoneId,
6 WeekdaySet, derive_occurrence_identity, is_slot, slot_values,
7};
8use crate::task_fields::TaskField;
9use anyhow::{Context, Result, bail, ensure};
10use chrono::{DateTime, NaiveDate, NaiveTime};
11use serde::{Deserialize, Serialize};
12use serde_json::Value;
13use sqlx::{SqliteConnection, query_scalar};
14use std::collections::{HashMap, HashSet};
15use std::path::{Path, PathBuf};
16
17mod archive;
18mod integrity;
19mod tables;
20use crate::db::{self, Database};
21
22const EXPORT_FORMAT: &str = "aven-export";
23const EXPORT_VERSION: i64 = 1;
24
25#[derive(Debug, Clone)]
26pub struct IntegrityReport {
27 pub quick_check_ok: bool,
28 pub quick_check_value: String,
29 pub checks: Vec<IntegrityCheck>,
30}
31
32#[derive(Debug, Clone)]
33pub struct IntegrityCheck {
34 pub label: &'static str,
35 pub ok: bool,
36 pub value: String,
37}
38
39#[derive(Debug, Serialize, Deserialize)]
40pub struct AvenExport {
41 pub format: String,
42 pub version: i64,
43 pub exported_at: String,
44 pub schema_version: i64,
45 #[serde(default)]
46 pub blobs_included: bool,
47 pub tables: ExportTables,
48}
49
50#[derive(Debug, Serialize, Deserialize)]
51pub struct ExportTables {
52 pub workspaces: Vec<WorkspaceRow>,
53 pub projects: Vec<ProjectRow>,
54 pub project_paths: Vec<ProjectPathRow>,
55 pub project_id_aliases: Vec<ProjectIdAliasRow>,
56 pub labels: Vec<LabelRow>,
57 pub tasks: Vec<TaskRow>,
58 pub task_labels: Vec<TaskLabelRow>,
59 pub notes: Vec<NoteRow>,
60 pub task_dependencies: Vec<TaskDependencyRow>,
61 pub task_epic_links: Vec<TaskEpicLinkRow>,
62 #[serde(default)]
63 pub task_attachments: Vec<TaskAttachmentRow>,
64 #[serde(default)]
65 pub blob_inventory: Vec<BlobInventoryExportRow>,
66 #[serde(default)]
67 pub recurrence_series: Vec<RecurrenceSeriesRow>,
68 #[serde(default)]
69 pub recurrence_series_labels: Vec<RecurrenceSeriesLabelRow>,
70 #[serde(default)]
71 pub recurrence_occurrences: Vec<RecurrenceOccurrenceRow>,
72 #[serde(default)]
73 pub recurrence_pause_intervals: Vec<RecurrencePauseIntervalRow>,
74 pub changes: Vec<ChangeRow>,
75 pub field_versions: Vec<FieldVersionRow>,
76 pub conflicts: Vec<ConflictRow>,
77 pub meta: Vec<MetaRow>,
78}
79
80#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
81pub struct WorkspaceRow {
82 pub id: WorkspaceId,
83 pub name: String,
84 pub key: String,
85 pub created_at: String,
86 pub updated_at: String,
87 pub archived: i64,
88}
89
90#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
91pub struct ProjectRow {
92 pub id: ProjectId,
93 pub workspace_id: WorkspaceId,
94 pub key: String,
95 pub name: String,
96 pub prefix: String,
97 pub created_at: String,
98 pub updated_at: String,
99 pub deleted: i64,
100}
101
102#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
103pub struct ProjectPathRow {
104 pub workspace_id: WorkspaceId,
105 pub project_id: ProjectId,
106 pub path: String,
107}
108
109#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
110pub struct ProjectIdAliasRow {
111 pub workspace_id: WorkspaceId,
112 pub remote_project_id: ProjectId,
113 pub local_project_id: ProjectId,
114}
115
116#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
117pub struct LabelRow {
118 pub workspace_id: WorkspaceId,
119 pub name: String,
120 pub created_at: String,
121}
122
123fn default_task_source() -> String {
124 TaskSource::Unknown.as_str().to_string()
125}
126
127#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
128pub struct TaskRow {
129 pub workspace_id: WorkspaceId,
130 pub id: TaskId,
131 pub title: String,
132 pub description: String,
133 pub project_id: ProjectId,
134 pub status: String,
135 pub priority: String,
136 #[serde(default = "default_task_source")]
137 pub source: String,
138 pub created_at: String,
139 pub updated_at: String,
140 pub queue_activity_at: String,
141 #[serde(default)]
142 pub available_at: String,
143 #[serde(default)]
144 pub due_on: String,
145 pub deleted: i64,
146 pub is_epic: i64,
147}
148
149#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
150pub struct TaskEpicLinkRow {
151 pub workspace_id: WorkspaceId,
152 pub child_task_id: TaskId,
153 pub epic_task_id: TaskId,
154 pub created_at: String,
155}
156
157#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
158pub struct TaskLabelRow {
159 pub workspace_id: WorkspaceId,
160 pub task_id: TaskId,
161 pub label: String,
162}
163
164#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
165pub struct NoteRow {
166 pub workspace_id: WorkspaceId,
167 pub id: String,
168 pub task_id: TaskId,
169 pub body: String,
170 pub created_at: String,
171 pub change_id: String,
172}
173
174#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
175pub struct TaskDependencyRow {
176 pub workspace_id: WorkspaceId,
177 pub task_id: TaskId,
178 pub depends_on_task_id: TaskId,
179 pub created_at: String,
180}
181
182#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
183pub struct TaskAttachmentRow {
184 pub workspace_id: WorkspaceId,
185 pub attachment_id: String,
186 pub task_id: TaskId,
187 pub sha256: String,
188 pub byte_size: i64,
189 pub media_type: String,
190 pub filename: Option<String>,
191 pub alt_text: Option<String>,
192 pub width: Option<i64>,
193 pub height: Option<i64>,
194 pub created_at: String,
195 pub created_by_change_id: Option<String>,
196 pub deleted: i64,
197 pub deleted_at: Option<String>,
198 pub deleted_by_change_id: Option<String>,
199}
200
201#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
202pub struct BlobInventoryExportRow {
203 pub sha256: String,
204 pub byte_size: i64,
205 pub media_type: String,
206 pub available: i64,
207 pub first_seen_at: String,
208 pub last_verified_at: Option<String>,
209}
210
211#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
212pub struct RecurrenceSeriesRow {
213 pub workspace_id: WorkspaceId,
214 pub id: RecurrenceSeriesId,
215 pub title: String,
216 pub description: String,
217 pub project_id: ProjectId,
218 pub priority: String,
219 pub initial_status: String,
220 pub frequency: String,
221 pub interval: i64,
222 pub weekdays: String,
223 pub timezone: String,
224 pub start_on: String,
225 pub available_local_time: String,
226 pub due_policy: String,
227 pub state: String,
228 pub stopped_at: String,
229 pub created_at: String,
230 pub updated_at: String,
231 pub deleted: i64,
232}
233
234#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
235pub struct RecurrenceSeriesLabelRow {
236 pub workspace_id: WorkspaceId,
237 pub series_id: RecurrenceSeriesId,
238 pub label: String,
239}
240
241#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
242pub struct RecurrenceOccurrenceRow {
243 pub workspace_id: WorkspaceId,
244 pub series_id: RecurrenceSeriesId,
245 pub slot_on: String,
246 pub task_id: String,
247 pub outcome: String,
248 pub resolved_at: String,
249 pub outcome_change_id: String,
250 pub projection_state: String,
251 pub archived_at: String,
252}
253
254#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
255pub struct RecurrencePauseIntervalRow {
256 pub workspace_id: WorkspaceId,
257 pub id: String,
258 pub series_id: RecurrenceSeriesId,
259 pub paused_at: String,
260 pub resumed_at: String,
261 pub suspended_slot_on: String,
262 pub suspended_task_id: String,
263 pub created_by_change_id: String,
264 pub resolved_by_change_id: String,
265}
266
267#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
268pub struct ChangeRow {
269 pub change_id: String,
270 pub client_id: String,
271 pub local_seq: i64,
272 pub entity_type: String,
273 pub entity_id: String,
274 pub field: Option<String>,
275 pub op_type: String,
276 pub payload: String,
277 pub base_version: Option<String>,
278 pub created_at: String,
279 pub server_seq: Option<i64>,
280}
281
282#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
283pub struct FieldVersionRow {
284 pub workspace_id: WorkspaceId,
285 pub entity_type: String,
286 pub entity_id: String,
287 pub field: String,
288 pub version: String,
289}
290
291#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
292pub struct ConflictRow {
293 pub id: i64,
294 pub workspace_id: WorkspaceId,
295 pub entity_type: String,
296 pub entity_id: String,
297 pub task_id: String,
298 pub field: String,
299 pub base_version: Option<String>,
300 pub local_value: String,
301 pub remote_value: String,
302 pub local_change_id: Option<String>,
303 pub remote_change_id: String,
304 pub variant_a: String,
305 pub variant_b: String,
306 pub created_at: String,
307 pub resolved: i64,
308}
309
310#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
311pub struct MetaRow {
312 pub key: String,
313 pub value: String,
314}
315
316impl Database {
317 pub async fn export_data(&self, exported_at: String) -> Result<AvenExport> {
318 let mut conn = self.acquire().await?;
319 let schema_version = db::current_schema_version(&mut conn).await?;
320 Ok(AvenExport {
321 format: EXPORT_FORMAT.to_string(),
322 version: EXPORT_VERSION,
323 exported_at,
324 schema_version,
325 blobs_included: false,
326 tables: ExportTables {
327 workspaces: scan_workspaces(&mut conn).await?,
328 projects: scan_projects(&mut conn).await?,
329 project_paths: scan_project_paths(&mut conn).await?,
330 project_id_aliases: scan_project_id_aliases(&mut conn).await?,
331 labels: scan_labels(&mut conn).await?,
332 tasks: scan_tasks(&mut conn).await?,
333 task_labels: scan_task_labels(&mut conn).await?,
334 notes: scan_notes(&mut conn).await?,
335 task_dependencies: scan_task_dependencies(&mut conn).await?,
336 task_epic_links: scan_task_epic_links(&mut conn).await?,
337 task_attachments: scan_task_attachments(&mut conn).await?,
338 blob_inventory: scan_blob_inventory(&mut conn).await?,
339 recurrence_series: scan_recurrence_series(&mut conn).await?,
340 recurrence_series_labels: scan_recurrence_series_labels(&mut conn).await?,
341 recurrence_occurrences: scan_recurrence_occurrences(&mut conn).await?,
342 recurrence_pause_intervals: scan_recurrence_pause_intervals(&mut conn).await?,
343 changes: scan_changes(&mut conn).await?,
344 field_versions: scan_field_versions(&mut conn).await?,
345 conflicts: scan_conflicts(&mut conn).await?,
346 meta: scan_meta(&mut conn).await?,
347 },
348 })
349 }
350
351 pub async fn validate_import_data(&self, export: &AvenExport) -> Result<()> {
352 let mut conn = self.acquire().await?;
353 ensure_supported_export(&mut conn, export).await?;
354 validate_export_snapshot(export)
355 }
356
357 pub async fn import_data(&self, export: &AvenExport) -> Result<IntegrityReport> {
358 let mut conn = self.acquire().await?;
359 ensure_supported_export(&mut conn, export).await?;
360 validate_export_snapshot(export)?;
361 let target_client_id = db::get_meta(&mut conn, "client_id")
362 .await?
363 .context("missing target client_id")?;
364 let mut tx = db::begin_immediate(&mut conn).await?;
365 replace_from_export(&mut tx, export, &target_client_id).await?;
366 let report = database_integrity_report_with_connection(&mut tx).await?;
367 ensure_integrity_ok(&report)?;
368 tx.commit().await?;
369 Ok(report)
370 }
371
372 pub async fn database_integrity_report(&self) -> Result<IntegrityReport> {
373 let mut conn = self.acquire().await?;
374 database_integrity_report_with_connection(&mut conn).await
375 }
376
377 pub async fn attachment_integrity_checks(
378 &self,
379 blob_dir: &Path,
380 deep: bool,
381 ) -> Result<Vec<IntegrityCheck>> {
382 let mut conn = self.acquire().await?;
383 integrity::attachment_integrity_checks(&mut conn, blob_dir, deep).await
384 }
385
386 pub async fn create_backup_archive(
387 &self,
388 db_path: &Path,
389 blob_dir: &Path,
390 output: &Path,
391 ) -> Result<()> {
392 let mut conn = self.acquire().await?;
393 let hashes: Vec<String> = sqlx::query_scalar(
394 "SELECT sha256 FROM blob_inventory WHERE available = 1 ORDER BY sha256",
395 )
396 .fetch_all(&mut *conn)
397 .await?;
398 let mut leases = Vec::with_capacity(hashes.len());
399 for hash in hashes {
400 match crate::attachments::lifecycle::acquire_lease(
401 &mut conn,
402 &hash,
403 "backup",
404 &crate::attachments::lifecycle::SystemClock,
405 )
406 .await
407 {
408 Ok(lease) => leases.push(lease),
409 Err(error) => {
410 for lease in leases {
411 let _ =
412 crate::attachments::lifecycle::release_lease(&mut conn, &lease).await;
413 }
414 return Err(error);
415 }
416 }
417 }
418 let backup_result =
419 archive::create_backup_archive(&mut conn, db_path, blob_dir, output).await;
420 for lease in leases {
421 crate::attachments::lifecycle::release_lease(&mut conn, &lease).await?;
422 }
423 backup_result
424 }
425}
426
427pub fn is_backup_archive(path: &Path) -> Result<bool> {
428 archive::is_archive_path(path)
429}
430
431pub async fn restore_backup_archive(
432 db_path: &Path,
433 blob_dir: &Path,
434 source: &Path,
435) -> Result<PathBuf> {
436 archive::restore_backup_archive(db_path, blob_dir, source).await
437}
438
439async fn scan_workspaces(conn: &mut SqliteConnection) -> Result<Vec<WorkspaceRow>> {
440 tables::scan_rows(
441 conn,
442 "SELECT id, name, key, created_at, updated_at, archived FROM workspaces",
443 )
444 .await
445}
446
447async fn scan_projects(conn: &mut SqliteConnection) -> Result<Vec<ProjectRow>> {
448 tables::scan_rows(
449 conn,
450 "SELECT id, workspace_id, key, name, prefix, created_at, updated_at, deleted FROM projects",
451 )
452 .await
453}
454
455async fn scan_project_paths(conn: &mut SqliteConnection) -> Result<Vec<ProjectPathRow>> {
456 tables::scan_rows(
457 conn,
458 "SELECT workspace_id, project_id, path FROM project_paths",
459 )
460 .await
461}
462
463async fn scan_project_id_aliases(conn: &mut SqliteConnection) -> Result<Vec<ProjectIdAliasRow>> {
464 tables::scan_rows(
465 conn,
466 "SELECT workspace_id, remote_project_id, local_project_id FROM project_id_aliases",
467 )
468 .await
469}
470
471async fn scan_labels(conn: &mut SqliteConnection) -> Result<Vec<LabelRow>> {
472 tables::scan_rows(conn, "SELECT workspace_id, name, created_at FROM labels").await
473}
474
475async fn scan_tasks(conn: &mut SqliteConnection) -> Result<Vec<TaskRow>> {
476 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
477}
478
479async fn scan_task_labels(conn: &mut SqliteConnection) -> Result<Vec<TaskLabelRow>> {
480 tables::scan_rows(conn, "SELECT workspace_id, task_id, label FROM task_labels").await
481}
482
483async fn scan_notes(conn: &mut SqliteConnection) -> Result<Vec<NoteRow>> {
484 tables::scan_rows(
485 conn,
486 "SELECT workspace_id, id, task_id, body, created_at, change_id FROM notes",
487 )
488 .await
489}
490
491async fn scan_task_dependencies(conn: &mut SqliteConnection) -> Result<Vec<TaskDependencyRow>> {
492 tables::scan_rows(
493 conn,
494 "SELECT workspace_id, task_id, depends_on_task_id, created_at FROM task_dependencies",
495 )
496 .await
497}
498
499async fn scan_task_epic_links(conn: &mut SqliteConnection) -> Result<Vec<TaskEpicLinkRow>> {
500 tables::scan_rows(
501 conn,
502 "SELECT workspace_id, child_task_id, epic_task_id, created_at FROM task_epic_links",
503 )
504 .await
505}
506
507async fn scan_task_attachments(conn: &mut SqliteConnection) -> Result<Vec<TaskAttachmentRow>> {
508 tables::scan_rows(
509 conn,
510 "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",
511 )
512 .await
513}
514
515async fn scan_blob_inventory(conn: &mut SqliteConnection) -> Result<Vec<BlobInventoryExportRow>> {
516 tables::scan_rows(
517 conn,
518 "SELECT sha256, byte_size, media_type, available, first_seen_at, last_verified_at FROM blob_inventory",
519 )
520 .await
521}
522
523async fn scan_recurrence_series(conn: &mut SqliteConnection) -> Result<Vec<RecurrenceSeriesRow>> {
524 tables::scan_rows(
525 conn,
526 "SELECT workspace_id, id, title, description, project_id, priority, initial_status, frequency, interval, weekdays, timezone, start_on, available_local_time, due_policy, state, stopped_at, created_at, updated_at, deleted FROM recurrence_series",
527 )
528 .await
529}
530
531async fn scan_recurrence_series_labels(
532 conn: &mut SqliteConnection,
533) -> Result<Vec<RecurrenceSeriesLabelRow>> {
534 tables::scan_rows(
535 conn,
536 "SELECT workspace_id, series_id, label FROM recurrence_series_labels",
537 )
538 .await
539}
540
541async fn scan_recurrence_occurrences(
542 conn: &mut SqliteConnection,
543) -> Result<Vec<RecurrenceOccurrenceRow>> {
544 tables::scan_rows(
545 conn,
546 "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at, outcome_change_id, projection_state, archived_at FROM recurrence_occurrences",
547 )
548 .await
549}
550
551async fn scan_recurrence_pause_intervals(
552 conn: &mut SqliteConnection,
553) -> Result<Vec<RecurrencePauseIntervalRow>> {
554 tables::scan_rows(
555 conn,
556 "SELECT workspace_id, id, series_id, paused_at, resumed_at, suspended_slot_on, suspended_task_id, created_by_change_id, resolved_by_change_id FROM recurrence_pause_intervals",
557 )
558 .await
559}
560
561async fn scan_changes(conn: &mut SqliteConnection) -> Result<Vec<ChangeRow>> {
562 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
563}
564
565async fn scan_field_versions(conn: &mut SqliteConnection) -> Result<Vec<FieldVersionRow>> {
566 tables::scan_rows(
567 conn,
568 "SELECT workspace_id, entity_type, entity_id, field, version FROM field_versions",
569 )
570 .await
571}
572
573async fn scan_conflicts(conn: &mut SqliteConnection) -> Result<Vec<ConflictRow>> {
574 tables::scan_rows(conn, "SELECT id, workspace_id, entity_type, entity_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
575}
576
577async fn scan_meta(conn: &mut SqliteConnection) -> Result<Vec<MetaRow>> {
578 tables::scan_rows(conn, "SELECT key, value FROM meta").await
579}
580
581async fn ensure_supported_export(_conn: &mut SqliteConnection, export: &AvenExport) -> Result<()> {
582 if export.format != EXPORT_FORMAT {
583 bail!("error export-format-unsupported format={}", export.format);
584 }
585 if export.version != EXPORT_VERSION {
586 bail!(
587 "error export-version-unsupported version={}",
588 export.version
589 );
590 }
591 Ok(())
592}
593
594fn validate_export_snapshot(export: &AvenExport) -> Result<()> {
595 let mut workspace_ids = HashSet::new();
596 for workspace in &export.tables.workspaces {
597 if workspace_ids.contains(&workspace.id) {
598 continue;
599 }
600 workspace_ids.insert(workspace.id.clone());
601 }
602
603 let mut project_ids: HashMap<WorkspaceId, HashSet<ProjectId>> = HashMap::new();
604 for project in &export.tables.projects {
605 if !workspace_ids.contains(&project.workspace_id) {
606 bail!(
607 "error invalid-export-snapshot project.workspace_id={} is missing",
608 project.workspace_id
609 );
610 }
611 project_ids
612 .entry(project.workspace_id.clone())
613 .or_default()
614 .insert(project.id.clone());
615 }
616
617 for path in &export.tables.project_paths {
618 let projects = project_ids.get(&path.workspace_id).ok_or_else(|| {
619 anyhow::Error::msg(format!(
620 "error invalid-export-snapshot project_path.workspace_id={} is missing",
621 path.workspace_id
622 ))
623 })?;
624 if !projects.contains(&path.project_id) {
625 bail!(
626 "error invalid-export-snapshot project_path.project_id={} is missing in workspace {}",
627 path.project_id,
628 path.workspace_id
629 );
630 }
631 }
632
633 let mut label_keys: HashSet<(WorkspaceId, String)> = HashSet::new();
634 for label in &export.tables.labels {
635 if !workspace_ids.contains(&label.workspace_id) {
636 bail!(
637 "error invalid-export-snapshot label.workspace_id={} is missing",
638 label.workspace_id
639 );
640 }
641 label_keys.insert((label.workspace_id.clone(), label.name.clone()));
642 }
643
644 let mut task_ids: HashMap<WorkspaceId, HashSet<TaskId>> = HashMap::new();
645 for task in &export.tables.tasks {
646 TaskSource::parse(&task.source)?;
647 if let Err(error) = crate::time_validation::validate_due_on_value(&task.due_on) {
648 bail!(
649 "error invalid-export-snapshot task.due_on={} is invalid: {error}",
650 task.due_on
651 );
652 }
653 let workspace_projects = project_ids.get(&task.workspace_id).ok_or_else(|| {
654 anyhow::Error::msg(format!(
655 "error invalid-export-snapshot task.workspace_id={} is missing",
656 task.workspace_id
657 ))
658 })?;
659 if !workspace_projects.contains(&task.project_id) {
660 bail!(
661 "error invalid-export-snapshot task.project_id={} is missing in workspace {}",
662 task.project_id,
663 task.workspace_id
664 );
665 }
666 task_ids
667 .entry(task.workspace_id.clone())
668 .or_default()
669 .insert(task.id.clone());
670 }
671
672 for task_label in &export.tables.task_labels {
673 let task_workspace = task_ids.get(&task_label.workspace_id).ok_or_else(|| {
674 anyhow::Error::msg(format!(
675 "error invalid-export-snapshot task_label.workspace_id={} is missing",
676 task_label.workspace_id
677 ))
678 })?;
679 if !task_workspace.contains(&task_label.task_id) {
680 bail!(
681 "error invalid-export-snapshot task_label.task_id={} is missing in workspace {}",
682 task_label.task_id,
683 task_label.workspace_id
684 );
685 }
686 if !label_keys.contains(&(task_label.workspace_id.clone(), task_label.label.clone())) {
687 bail!(
688 "error invalid-export-snapshot task_label.label={} is missing in workspace {}",
689 task_label.label,
690 task_label.workspace_id
691 );
692 }
693 }
694
695 for note in &export.tables.notes {
696 let task_workspace = task_ids.get(¬e.workspace_id).ok_or_else(|| {
697 anyhow::Error::msg(format!(
698 "error invalid-export-snapshot note.workspace_id={} is missing",
699 note.workspace_id
700 ))
701 })?;
702 if !task_workspace.contains(¬e.task_id) {
703 bail!(
704 "error invalid-export-snapshot note.task_id={} is missing in workspace {}",
705 note.task_id,
706 note.workspace_id
707 );
708 }
709 }
710
711 for dep in &export.tables.task_dependencies {
712 let tasks = task_ids.get(&dep.workspace_id).ok_or_else(|| {
713 anyhow::Error::msg(format!(
714 "error invalid-export-snapshot dependency.workspace_id={} is missing",
715 dep.workspace_id
716 ))
717 })?;
718 if !tasks.contains(&dep.task_id) || !tasks.contains(&dep.depends_on_task_id) {
719 bail!(
720 "error invalid-export-snapshot task_dependencies are missing tasks in workspace {}",
721 dep.workspace_id
722 );
723 }
724 }
725
726 for epic_link in &export.tables.task_epic_links {
727 let tasks = task_ids.get(&epic_link.workspace_id).ok_or_else(|| {
728 anyhow::Error::msg(format!(
729 "error invalid-export-snapshot epic_link.workspace_id={} is missing",
730 epic_link.workspace_id
731 ))
732 })?;
733 if !tasks.contains(&epic_link.child_task_id) || !tasks.contains(&epic_link.epic_task_id) {
734 bail!(
735 "error invalid-export-snapshot task_epic_links are missing tasks in workspace {}",
736 epic_link.workspace_id
737 );
738 }
739 }
740
741 let mut inventory = HashMap::new();
742 for blob in &export.tables.blob_inventory {
743 crate::attachments::validate_sha256(&blob.sha256)?;
744 crate::attachments::validate_media_type(&blob.media_type)?;
745 crate::attachments::validate_blob_size(usize::try_from(blob.byte_size).unwrap_or(0))?;
746 if blob.available != 0 && blob.available != 1 {
747 bail!("error invalid-export-snapshot blob_inventory.available invalid");
748 }
749 if inventory
750 .insert(
751 blob.sha256.clone(),
752 (blob.byte_size, blob.media_type.as_str()),
753 )
754 .is_some()
755 {
756 bail!("error invalid-export-snapshot blob_inventory.sha256 duplicate");
757 }
758 }
759
760 for attachment in &export.tables.task_attachments {
761 crate::attachments::validate_attachment_id(&attachment.attachment_id)?;
762 crate::attachments::validate_sha256(&attachment.sha256)?;
763 crate::attachments::validate_media_type(&attachment.media_type)?;
764 crate::attachments::validate_blob_size(usize::try_from(attachment.byte_size).unwrap_or(0))?;
765 crate::attachments::validate_filename(attachment.filename.as_deref())?;
766 crate::attachments::validate_alt_text(attachment.alt_text.as_deref())?;
767 crate::attachments::validate_dimensions(attachment.width, attachment.height)?;
768 let tasks = task_ids.get(&attachment.workspace_id).ok_or_else(|| {
769 anyhow::Error::msg(format!(
770 "error invalid-export-snapshot attachment.workspace_id={} is missing",
771 attachment.workspace_id
772 ))
773 })?;
774 if !tasks.contains(&attachment.task_id) {
775 bail!(
776 "error invalid-export-snapshot attachment.task_id={} is missing in workspace {}",
777 attachment.task_id,
778 attachment.workspace_id
779 );
780 }
781 let Some((inventory_size, inventory_media_type)) = inventory.get(&attachment.sha256) else {
782 bail!("error invalid-export-snapshot attachment inventory missing");
783 };
784 if *inventory_size != attachment.byte_size || *inventory_media_type != attachment.media_type
785 {
786 bail!("error invalid-export-snapshot attachment inventory metadata mismatch");
787 }
788 if attachment.deleted != 0 && attachment.deleted != 1 {
789 bail!(
790 "error invalid-export-snapshot attachment.deleted={} for attachment {}",
791 attachment.deleted,
792 attachment.attachment_id
793 );
794 }
795 }
796
797 for alias in &export.tables.project_id_aliases {
798 let workspace_projects = project_ids.get(&alias.workspace_id).ok_or_else(|| {
799 anyhow::Error::msg(format!(
800 "error invalid-export-snapshot project_alias.workspace_id={} is missing",
801 alias.workspace_id
802 ))
803 })?;
804 if !workspace_projects.contains(&alias.local_project_id) {
805 bail!(
806 "error invalid-export-snapshot local_project_id={} is missing in workspace {}",
807 alias.local_project_id,
808 alias.workspace_id
809 );
810 }
811 }
812
813 if has_recurrence_data(export) {
814 validate_recurrence_snapshot(export, &workspace_ids, &project_ids, &label_keys, &task_ids)?;
815 }
816
817 Ok(())
818}
819
820fn has_recurrence_data(export: &AvenExport) -> bool {
821 !export.tables.recurrence_series.is_empty()
822 || !export.tables.recurrence_series_labels.is_empty()
823 || !export.tables.recurrence_occurrences.is_empty()
824 || !export.tables.recurrence_pause_intervals.is_empty()
825}
826
827fn validate_recurrence_snapshot(
828 export: &AvenExport,
829 workspace_ids: &HashSet<WorkspaceId>,
830 project_ids: &HashMap<WorkspaceId, HashSet<ProjectId>>,
831 label_keys: &HashSet<(WorkspaceId, String)>,
832 task_ids: &HashMap<WorkspaceId, HashSet<TaskId>>,
833) -> Result<()> {
834 struct ValidatedSeries<'a> {
835 row: &'a RecurrenceSeriesRow,
836 schedule: RecurrenceSchedule,
837 state: RecurrenceSeriesState,
838 created_at: DateTime<chrono::FixedOffset>,
839 stopped_at: Option<DateTime<chrono::FixedOffset>>,
840 }
841
842 let task_rows = export
843 .tables
844 .tasks
845 .iter()
846 .map(|task| ((task.workspace_id.clone(), task.id.clone()), task))
847 .collect::<HashMap<_, _>>();
848 let mut change_rows = HashMap::new();
849 for change in &export.tables.changes {
850 ensure!(
851 change_rows
852 .insert(change.change_id.as_str(), change)
853 .is_none(),
854 "error invalid-export-snapshot change.change_id={} is duplicated",
855 change.change_id
856 );
857 }
858 let field_versions = export
859 .tables
860 .field_versions
861 .iter()
862 .map(|row| {
863 (
864 (
865 row.workspace_id.clone(),
866 row.entity_type.as_str(),
867 row.entity_id.as_str(),
868 row.field.as_str(),
869 ),
870 row.version.as_str(),
871 )
872 })
873 .collect::<HashMap<_, _>>();
874
875 let mut series_by_id = HashMap::new();
876 for row in &export.tables.recurrence_series {
877 ensure!(
878 workspace_ids.contains(&row.workspace_id),
879 "error invalid-export-snapshot recurrence_series.workspace_id={} is missing",
880 row.workspace_id
881 );
882 ensure!(
883 project_ids
884 .get(&row.workspace_id)
885 .is_some_and(|projects| projects.contains(&row.project_id)),
886 "error invalid-export-snapshot recurrence_series.project_id={} is missing in workspace {}",
887 row.project_id,
888 row.workspace_id
889 );
890 TaskPriority::parse(&row.priority).context("invalid recurrence series priority")?;
891 let initial_status =
892 TaskStatus::parse(&row.initial_status).context("invalid recurrence initial status")?;
893 ensure!(
894 initial_status.is_open(),
895 "error invalid-export-snapshot recurrence initial status must be open"
896 );
897 let frequency = RecurrenceFrequency::parse(&row.frequency)?;
898 let interval = u32::try_from(row.interval).context("invalid recurrence interval")?;
899 let weekdays = row
900 .weekdays
901 .parse::<WeekdaySet>()
902 .map_err(anyhow::Error::msg)?;
903 let rule = RecurrenceRule::new(frequency, interval, weekdays)?;
904 let timezone = row.timezone.parse::<TimeZoneId>()?;
905 let start_on = row
906 .start_on
907 .parse::<NaiveDate>()
908 .context("invalid recurrence start date")?;
909 let available_local_time = optional_import_text(&row.available_local_time)
910 .map(|value| value.parse::<NaiveTime>())
911 .transpose()
912 .context("invalid recurrence availability time")?;
913 let due_policy = RecurrenceDuePolicy::parse(&row.due_policy)?;
914 let state = RecurrenceSeriesState::parse(&row.state)?;
915 let stopped_at = optional_import_text(&row.stopped_at)
916 .map(DateTime::parse_from_rfc3339)
917 .transpose()
918 .context("invalid recurrence stop time")?;
919 ensure!(
920 matches!(state, RecurrenceSeriesState::Stopped) == stopped_at.is_some(),
921 "error invalid-export-snapshot recurrence stopped state and stop time disagree"
922 );
923 let created_at = DateTime::parse_from_rfc3339(&row.created_at)
924 .context("invalid recurrence creation time")?;
925 DateTime::parse_from_rfc3339(&row.updated_at).context("invalid recurrence update time")?;
926 ensure!(
927 matches!(row.deleted, 0 | 1),
928 "error invalid-export-snapshot recurrence deleted value must be zero or one"
929 );
930 let key = (row.workspace_id.clone(), row.id.clone());
931 ensure!(
932 !series_by_id.contains_key(&key),
933 "error invalid-export-snapshot recurrence series identity is duplicated"
934 );
935 series_by_id.insert(
936 key,
937 ValidatedSeries {
938 row,
939 schedule: RecurrenceSchedule::new(
940 rule,
941 timezone,
942 start_on,
943 available_local_time,
944 due_policy,
945 ),
946 state,
947 created_at,
948 stopped_at,
949 },
950 );
951 }
952
953 let mut series_label_keys = HashSet::new();
954 for row in &export.tables.recurrence_series_labels {
955 let series_key = (row.workspace_id.clone(), row.series_id.clone());
956 ensure!(
957 series_by_id.contains_key(&series_key),
958 "error invalid-export-snapshot recurrence series label has no series"
959 );
960 ensure!(
961 label_keys.contains(&(row.workspace_id.clone(), row.label.clone())),
962 "error invalid-export-snapshot recurrence series label={} is missing",
963 row.label
964 );
965 ensure!(
966 series_label_keys.insert((row.workspace_id.clone(), row.series_id.clone(), &row.label)),
967 "error invalid-export-snapshot recurrence series label is duplicated"
968 );
969 }
970
971 let mut occurrence_keys = HashSet::new();
972 let mut occurrence_tasks = HashSet::new();
973 let mut projected_series = HashSet::new();
974 for row in &export.tables.recurrence_occurrences {
975 let series_key = (row.workspace_id.clone(), row.series_id.clone());
976 let series = series_by_id
977 .get(&series_key)
978 .context("error invalid-export-snapshot recurrence occurrence has no series")?;
979 let slot_on = row
980 .slot_on
981 .parse::<NaiveDate>()
982 .context("invalid recurrence slot date")?;
983 ensure!(
984 occurrence_keys.insert((row.workspace_id.clone(), row.series_id.clone(), slot_on)),
985 "error invalid-export-snapshot recurrence occurrence identity is duplicated"
986 );
987 ensure!(
988 is_slot(&series.schedule.rule, series.schedule.start_on, slot_on),
989 "error invalid-export-snapshot recurrence slot={} is outside the series lattice",
990 slot_on
991 );
992 let slot = slot_values(&series.schedule, slot_on)?;
993 let boundary = DateTime::parse_from_rfc3339(&slot.boundary_at)?;
994 let creation_date = series
995 .created_at
996 .with_timezone(&series.schedule.timezone.timezone())
997 .date_naive();
998 ensure!(
999 slot_on >= creation_date,
1000 "error invalid-export-snapshot recurrence slot={} precedes the series lifecycle",
1001 slot_on
1002 );
1003 if let Some(stopped_at) = series.stopped_at {
1004 ensure!(
1005 boundary <= stopped_at,
1006 "error invalid-export-snapshot recurrence slot={} exceeds the stop boundary",
1007 slot_on
1008 );
1009 }
1010
1011 let projection_state = RecurrenceProjectionState::parse(&row.projection_state)?;
1012 let outcome = optional_import_text(&row.outcome)
1013 .map(RecurrenceOutcome::parse)
1014 .transpose()?;
1015 let task_id = optional_import_text(&row.task_id)
1016 .map(str::parse::<TaskId>)
1017 .transpose()?;
1018 let resolved_at = optional_import_text(&row.resolved_at);
1019 let outcome_change_id = optional_import_text(&row.outcome_change_id);
1020 let archived_at = optional_import_text(&row.archived_at);
1021 let valid_shape = match projection_state {
1022 RecurrenceProjectionState::Projected => {
1023 task_id.is_some()
1024 && outcome.is_none()
1025 && resolved_at.is_none()
1026 && outcome_change_id.is_none()
1027 && archived_at.is_none()
1028 }
1029 RecurrenceProjectionState::Resolved => {
1030 task_id.is_some()
1031 && outcome.is_some()
1032 && resolved_at.is_some()
1033 && outcome_change_id.is_some()
1034 && archived_at.is_none()
1035 }
1036 RecurrenceProjectionState::Archived => {
1037 task_id.is_some()
1038 && outcome.is_none()
1039 && resolved_at.is_none()
1040 && outcome_change_id.is_none()
1041 && archived_at.is_some()
1042 }
1043 };
1044 ensure!(
1045 valid_shape,
1046 "error invalid-export-snapshot recurrence occurrence state and fields disagree"
1047 );
1048 if matches!(projection_state, RecurrenceProjectionState::Projected) {
1049 ensure!(
1050 projected_series.insert(series_key.clone()),
1051 "error invalid-export-snapshot recurrence projection is not unique"
1052 );
1053 }
1054 for value in [resolved_at, archived_at].into_iter().flatten() {
1055 DateTime::parse_from_rfc3339(value)
1056 .context("invalid recurrence occurrence timestamp")?;
1057 }
1058
1059 if let Some(task_id) = task_id {
1060 ensure!(
1061 task_ids
1062 .get(&row.workspace_id)
1063 .is_some_and(|tasks| tasks.contains(&task_id)),
1064 "error invalid-export-snapshot recurrence task={} is missing",
1065 task_id
1066 );
1067 ensure!(
1068 occurrence_tasks.insert((row.workspace_id.clone(), task_id.clone())),
1069 "error invalid-export-snapshot recurrence task link is duplicated"
1070 );
1071 let identity = derive_occurrence_identity(
1072 &row.workspace_id,
1073 &row.series_id,
1074 &series.schedule,
1075 slot_on,
1076 )?;
1077 ensure!(
1078 identity.task_id == task_id,
1079 "error invalid-export-snapshot recurrence deterministic task identity mismatch slot={slot_on}"
1080 );
1081 let task = task_rows
1082 .get(&(row.workspace_id.clone(), task_id.clone()))
1083 .context("error invalid-export-snapshot recurrence task row is missing")?;
1084 ensure!(
1085 task.created_at == identity.created_at,
1086 "error invalid-export-snapshot recurrence deterministic task timestamp mismatch slot={slot_on}"
1087 );
1088 match outcome {
1089 Some(RecurrenceOutcome::Completed) => ensure!(
1090 task.status == "done",
1091 "error invalid-export-snapshot recurrence completed outcome requires done task"
1092 ),
1093 Some(RecurrenceOutcome::Skipped) => ensure!(
1094 task.status == "canceled",
1095 "error invalid-export-snapshot recurrence skipped outcome requires canceled task"
1096 ),
1097 None if matches!(projection_state, RecurrenceProjectionState::Projected) => {
1098 ensure!(
1099 TaskStatus::parse(&task.status)?.is_open(),
1100 "error invalid-export-snapshot recurrence projected task must be open"
1101 );
1102 }
1103 None => {}
1104 }
1105 validate_deterministic_change(
1106 change_rows.get(identity.task_change_id.as_str()).copied(),
1107 &identity,
1108 slot_on,
1109 false,
1110 )?;
1111 validate_deterministic_change(
1112 change_rows
1113 .get(identity.occurrence_change_id.as_str())
1114 .copied(),
1115 &identity,
1116 slot_on,
1117 true,
1118 )?;
1119 for field in TaskField::VERSIONED {
1120 let version = field_versions
1121 .get(&(
1122 row.workspace_id.clone(),
1123 "task",
1124 task_id.as_str(),
1125 field.as_str(),
1126 ))
1127 .context(
1128 "error invalid-export-snapshot recurrence task field version is missing",
1129 )?;
1130 if *version == identity.field_version_seeds.task {
1131 continue;
1132 }
1133 ensure!(
1134 change_rows.contains_key(version),
1135 "error invalid-export-snapshot recurrence task field version has no change"
1136 );
1137 }
1138 }
1139
1140 if let Some(change_id) = outcome_change_id {
1141 let change = change_rows
1142 .get(change_id)
1143 .context("error invalid-export-snapshot recurrence outcome change is missing")?;
1144 ensure!(
1145 change.entity_type == "recurrence_series"
1146 && change.entity_id == row.series_id.as_str()
1147 && change.field.as_deref() == Some("outcome")
1148 && change.op_type == "resolve_recurrence_occurrence",
1149 "error invalid-export-snapshot recurrence outcome change identity mismatch"
1150 );
1151 let payload: Value = serde_json::from_str(&change.payload)
1152 .context("invalid recurrence outcome change payload")?;
1153 ensure!(
1154 payload.get("slot_on").and_then(Value::as_str) == Some(row.slot_on.as_str())
1155 && payload.get("outcome").and_then(Value::as_str)
1156 == outcome.map(RecurrenceOutcome::as_str)
1157 && payload.get("resolved_at").and_then(Value::as_str) == resolved_at,
1158 "error invalid-export-snapshot recurrence outcome change payload mismatch"
1159 );
1160 }
1161 }
1162
1163 let mut pause_ids = HashSet::new();
1164 let mut pauses_by_series: HashMap<_, Vec<_>> = HashMap::new();
1165 for row in &export.tables.recurrence_pause_intervals {
1166 let series_key = (row.workspace_id.clone(), row.series_id.clone());
1167 let series = series_by_id
1168 .get(&series_key)
1169 .context("error invalid-export-snapshot recurrence pause has no series")?;
1170 ensure!(
1171 pause_ids.insert((row.workspace_id.clone(), row.id.as_str())),
1172 "error invalid-export-snapshot recurrence pause identity is duplicated"
1173 );
1174 let paused_at = DateTime::parse_from_rfc3339(&row.paused_at)
1175 .context("invalid recurrence pause time")?;
1176 let resumed_at = optional_import_text(&row.resumed_at)
1177 .map(DateTime::parse_from_rfc3339)
1178 .transpose()
1179 .context("invalid recurrence resume time")?;
1180 ensure!(
1181 resumed_at.is_none_or(|resumed| resumed > paused_at),
1182 "error invalid-export-snapshot recurrence pause interval is inverted"
1183 );
1184 ensure!(
1185 paused_at >= series.created_at,
1186 "error invalid-export-snapshot recurrence pause precedes the series lifecycle"
1187 );
1188 if let Some(stopped_at) = series.stopped_at {
1189 ensure!(
1190 paused_at <= stopped_at && resumed_at.is_none_or(|resumed| resumed <= stopped_at),
1191 "error invalid-export-snapshot recurrence pause exceeds the stop boundary"
1192 );
1193 }
1194 ensure!(
1195 resumed_at.is_some() != row.resolved_by_change_id.is_empty(),
1196 "error invalid-export-snapshot recurrence pause resolution fields disagree"
1197 );
1198 ensure!(
1199 row.suspended_slot_on.is_empty() == row.suspended_task_id.is_empty(),
1200 "error invalid-export-snapshot recurrence suspended task fields disagree"
1201 );
1202 if !row.suspended_slot_on.is_empty() {
1203 let slot_on = row.suspended_slot_on.parse::<NaiveDate>()?;
1204 let task_id = row.suspended_task_id.parse::<TaskId>()?;
1205 ensure!(
1206 occurrence_keys.contains(&(
1207 row.workspace_id.clone(),
1208 row.series_id.clone(),
1209 slot_on
1210 )) && occurrence_tasks.contains(&(row.workspace_id.clone(), task_id)),
1211 "error invalid-export-snapshot recurrence suspended task link is missing"
1212 );
1213 }
1214 for change_id in [
1215 Some(row.created_by_change_id.as_str()),
1216 optional_import_text(&row.resolved_by_change_id),
1217 ]
1218 .into_iter()
1219 .flatten()
1220 {
1221 let change = change_rows
1222 .get(change_id)
1223 .context("error invalid-export-snapshot recurrence pause change is missing")?;
1224 ensure!(
1225 change.entity_type == "recurrence_series"
1226 && change.entity_id == row.series_id.as_str(),
1227 "error invalid-export-snapshot recurrence pause change identity mismatch"
1228 );
1229 }
1230 pauses_by_series
1231 .entry(series_key)
1232 .or_default()
1233 .push((paused_at, resumed_at));
1234 }
1235 for pauses in pauses_by_series.values_mut() {
1236 pauses.sort_by_key(|(paused_at, _)| *paused_at);
1237 for pair in pauses.windows(2) {
1238 ensure!(
1239 pair[0].1.is_some_and(|resumed_at| resumed_at <= pair[1].0),
1240 "error invalid-export-snapshot recurrence pause intervals overlap"
1241 );
1242 }
1243 }
1244
1245 for series in series_by_id.values() {
1246 let has_lifecycle_conflict = export.tables.conflicts.iter().any(|conflict| {
1247 conflict.workspace_id == series.row.workspace_id
1248 && conflict.entity_type == "recurrence_series"
1249 && conflict.entity_id == series.row.id.as_str()
1250 && conflict.field == "state"
1251 && conflict.resolved == 0
1252 });
1253 let has_open_pause = pauses_by_series
1254 .get(&(series.row.workspace_id.clone(), series.row.id.clone()))
1255 .is_some_and(|pauses| pauses.iter().any(|(_, resumed_at)| resumed_at.is_none()));
1256 if !has_lifecycle_conflict {
1257 ensure!(
1258 matches!(series.state, RecurrenceSeriesState::Paused) == has_open_pause,
1259 "error invalid-export-snapshot recurrence state and open pause disagree"
1260 );
1261 }
1262 }
1263
1264 Ok(())
1265}
1266
1267fn validate_deterministic_change(
1268 change: Option<&ChangeRow>,
1269 identity: &crate::recurrence::RecurrenceOccurrenceIdentity,
1270 slot_on: NaiveDate,
1271 projection: bool,
1272) -> Result<()> {
1273 let change = change
1274 .context("error invalid-export-snapshot recurrence deterministic change is missing")?;
1275 let (entity_type, entity_id, field, op_type, created_at) = if projection {
1276 (
1277 "recurrence_series",
1278 identity.occurrence_link.series_id.as_str(),
1279 Some("projection"),
1280 "project_recurrence_occurrence",
1281 identity.occurrence_link.projected_at.as_str(),
1282 )
1283 } else {
1284 (
1285 "task",
1286 identity.task_id.as_str(),
1287 None,
1288 "create_task",
1289 identity.created_at.as_str(),
1290 )
1291 };
1292 ensure!(
1293 change.entity_type == entity_type
1294 && change.entity_id == entity_id
1295 && change.field.as_deref() == field
1296 && change.op_type == op_type
1297 && change.created_at == created_at,
1298 "error invalid-export-snapshot recurrence deterministic change identity mismatch"
1299 );
1300 let payload: Value = serde_json::from_str(&change.payload)
1301 .context("invalid recurrence deterministic change payload")?;
1302 for (key, expected) in [
1303 ("task_id", identity.task_id.as_str()),
1304 ("series_id", identity.occurrence_link.series_id.as_str()),
1305 ("task_change_id", identity.task_change_id.as_str()),
1306 (
1307 "occurrence_change_id",
1308 identity.occurrence_change_id.as_str(),
1309 ),
1310 (
1311 "task_field_version_seed",
1312 identity.field_version_seeds.task.as_str(),
1313 ),
1314 (
1315 "occurrence_field_version_seed",
1316 identity.field_version_seeds.occurrence.as_str(),
1317 ),
1318 ] {
1319 ensure!(
1320 payload.get(key).and_then(Value::as_str) == Some(expected),
1321 "error invalid-export-snapshot recurrence deterministic payload field={key} mismatch"
1322 );
1323 }
1324 ensure!(
1325 payload.get("slot_on").and_then(Value::as_str) == Some(&slot_on.to_string()),
1326 "error invalid-export-snapshot recurrence deterministic payload slot mismatch"
1327 );
1328 if projection {
1329 ensure!(
1330 payload.get("projected_at").and_then(Value::as_str)
1331 == Some(identity.occurrence_link.projected_at.as_str()),
1332 "error invalid-export-snapshot recurrence deterministic projection link mismatch"
1333 );
1334 }
1335 Ok(())
1336}
1337
1338fn optional_import_text(value: &str) -> Option<&str> {
1339 (!value.is_empty()).then_some(value)
1340}
1341
1342async fn replace_from_export(
1343 tx: &mut SqliteConnection,
1344 export: &AvenExport,
1345 target_client_id: &str,
1346) -> Result<()> {
1347 let recurrence_data = has_recurrence_data(export);
1348 let delete_order = [
1349 "DELETE FROM recurrence_pause_intervals",
1350 "DELETE FROM recurrence_occurrences",
1351 "DELETE FROM recurrence_series_labels",
1352 "DELETE FROM recurrence_series",
1353 "DELETE FROM task_attachments",
1354 "DELETE FROM blob_inventory",
1355 "DELETE FROM task_epic_links",
1356 "DELETE FROM task_dependencies",
1357 "DELETE FROM task_labels",
1358 "DELETE FROM notes",
1359 "DELETE FROM conflicts",
1360 "DELETE FROM field_versions",
1361 "DELETE FROM changes",
1362 "DELETE FROM project_paths",
1363 "DELETE FROM project_id_aliases",
1364 "DELETE FROM tasks",
1365 "DELETE FROM labels",
1366 "DELETE FROM projects",
1367 "DELETE FROM workspaces",
1368 "DELETE FROM meta",
1369 ];
1370 for sql in delete_order {
1371 sqlx::query(sql).execute(&mut *tx).await?;
1372 }
1373
1374 db::set_meta(tx, "client_id", target_client_id).await?;
1375 db::set_meta(tx, "sync_cursor", "0").await?;
1376 let local_seq = export
1377 .tables
1378 .changes
1379 .iter()
1380 .map(|row| row.local_seq)
1381 .max()
1382 .unwrap_or(0);
1383 db::set_meta(tx, "local_seq", &local_seq.to_string()).await?;
1384
1385 for meta in &export.tables.meta {
1386 if matches!(
1387 meta.key.as_str(),
1388 "client_id" | "sync_server_url" | "sync_cursor" | "local_seq"
1389 ) {
1390 continue;
1391 }
1392 db::set_meta(tx, &meta.key, &meta.value).await?;
1393 }
1394
1395 let suppressed_attachment_changes = export
1396 .tables
1397 .changes
1398 .iter()
1399 .filter(|change| {
1400 change.server_seq.is_none()
1401 && change.field.as_deref() == Some("attachments")
1402 && matches!(
1403 change.op_type.as_str(),
1404 "attachment_add" | "attachment_delete"
1405 )
1406 })
1407 .map(|change| change.change_id.as_str())
1408 .collect::<HashSet<_>>();
1409 let mut attachments = export.tables.task_attachments.clone();
1410 for attachment in &mut attachments {
1411 if attachment
1412 .created_by_change_id
1413 .as_deref()
1414 .is_some_and(|id| suppressed_attachment_changes.contains(id))
1415 {
1416 attachment.created_by_change_id = None;
1417 }
1418 if attachment
1419 .deleted_by_change_id
1420 .as_deref()
1421 .is_some_and(|id| suppressed_attachment_changes.contains(id))
1422 {
1423 attachment.deleted_by_change_id = None;
1424 }
1425 }
1426 let changes = export
1427 .tables
1428 .changes
1429 .iter()
1430 .filter(|change| !suppressed_attachment_changes.contains(change.change_id.as_str()))
1431 .filter(|change| {
1432 recurrence_data
1433 || (change.entity_type != "recurrence_series"
1434 && !change.op_type.contains("recurrence"))
1435 })
1436 .cloned()
1437 .collect::<Vec<_>>();
1438 let field_versions = export
1439 .tables
1440 .field_versions
1441 .iter()
1442 .filter(|row| recurrence_data || row.entity_type != "recurrence_series")
1443 .cloned()
1444 .collect::<Vec<_>>();
1445 let conflicts = export
1446 .tables
1447 .conflicts
1448 .iter()
1449 .filter(|row| recurrence_data || row.entity_type != "recurrence_series")
1450 .cloned()
1451 .collect::<Vec<_>>();
1452
1453 tables::import_workspaces(tx, &export.tables.workspaces).await?;
1454 tables::import_projects(tx, &export.tables.projects).await?;
1455 tables::import_project_id_aliases(tx, &export.tables.project_id_aliases).await?;
1456 tables::import_project_paths(tx, &export.tables.project_paths).await?;
1457 tables::import_labels(tx, &export.tables.labels).await?;
1458 tables::import_tasks(tx, &export.tables.tasks).await?;
1459 tables::import_task_labels(tx, &export.tables.task_labels).await?;
1460 tables::import_notes(tx, &export.tables.notes).await?;
1461 tables::import_task_dependencies(tx, &export.tables.task_dependencies).await?;
1462 tables::import_task_epic_links(tx, &export.tables.task_epic_links).await?;
1463 tables::import_blob_inventory(tx, &export.tables.blob_inventory).await?;
1464 tables::import_task_attachments(tx, &attachments).await?;
1465 if recurrence_data {
1466 tables::import_recurrence_series(tx, &export.tables.recurrence_series).await?;
1467 tables::import_recurrence_series_labels(tx, &export.tables.recurrence_series_labels)
1468 .await?;
1469 tables::import_recurrence_occurrences(tx, &export.tables.recurrence_occurrences).await?;
1470 tables::import_recurrence_pause_intervals(tx, &export.tables.recurrence_pause_intervals)
1471 .await?;
1472 }
1473 tables::import_changes(tx, &changes).await?;
1474 tables::import_field_versions(tx, &field_versions).await?;
1475 tables::import_conflicts(tx, &conflicts).await?;
1476
1477 Ok(())
1478}
1479
1480async fn database_integrity_report_with_connection(
1481 conn: &mut SqliteConnection,
1482) -> Result<IntegrityReport> {
1483 let quick_check_value: String = query_scalar("PRAGMA quick_check")
1484 .fetch_one(&mut *conn)
1485 .await?;
1486 let mut checks = Vec::new();
1487 checks.push(count_check(
1488 conn,
1489 "task projects",
1490 "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",
1491 )
1492 .await?);
1493 checks.push(count_check(
1494 conn,
1495 "project paths",
1496 "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",
1497 )
1498 .await?);
1499 checks.push(count_check(
1500 conn,
1501 "project aliases",
1502 "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",
1503 )
1504 .await?);
1505 checks.push(count_check(
1506 conn,
1507 "task label tasks",
1508 "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",
1509 )
1510 .await?);
1511 checks.push(count_check(
1512 conn,
1513 "task label labels",
1514 "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",
1515 )
1516 .await?);
1517 checks.push(count_check(
1518 conn,
1519 "notes",
1520 "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",
1521 )
1522 .await?);
1523 checks.push(count_check(
1524 conn,
1525 "note changes",
1526 "SELECT count(*) FROM notes n LEFT JOIN changes c ON c.change_id = n.change_id WHERE c.change_id IS NULL",
1527 )
1528 .await?);
1529 checks.push(count_check(
1530 conn,
1531 "dependency tasks",
1532 "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",
1533 )
1534 .await?);
1535 checks.push(count_check(
1536 conn,
1537 "dependency targets",
1538 "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",
1539 )
1540 .await?);
1541 checks.push(count_check(
1542 conn,
1543 "epic link children",
1544 "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",
1545 )
1546 .await?);
1547 checks.push(count_check(
1548 conn,
1549 "epic link parents",
1550 "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",
1551 )
1552 .await?);
1553 checks.push(count_check(
1554 conn,
1555 "epic link parent flags",
1556 "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",
1557 )
1558 .await?);
1559 checks.push(count_check(
1560 conn,
1561 "conflict tasks",
1562 "SELECT count(*) FROM conflicts c LEFT JOIN tasks t ON t.workspace_id = c.workspace_id AND t.id = c.entity_id WHERE c.resolved = 0 AND c.entity_type = 'task' AND t.id IS NULL",
1563 )
1564 .await?);
1565 checks.push(count_check(
1566 conn,
1567 "task due dates",
1568 "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)",
1569 )
1570 .await?);
1571 checks.push(count_check(
1572 conn,
1573 "field version tasks",
1574 "SELECT count(*) FROM field_versions fv LEFT JOIN tasks t ON t.workspace_id = fv.workspace_id AND t.id = fv.entity_id WHERE fv.entity_type = 'task' AND t.id IS NULL",
1575 )
1576 .await?);
1577 checks.push(count_check(
1578 conn,
1579 "field version changes",
1580 "SELECT count(*) FROM field_versions fv LEFT JOIN changes c ON c.change_id = fv.version WHERE c.change_id IS NULL AND NOT (fv.entity_type = 'task' AND EXISTS (SELECT 1 FROM recurrence_occurrences o WHERE o.workspace_id = fv.workspace_id AND o.task_id = fv.entity_id))",
1581 )
1582 .await?);
1583 checks.extend(integrity::recurrence_integrity_checks(conn).await?);
1584 push_meta_checks(conn, &mut checks).await?;
1585
1586 Ok(IntegrityReport {
1587 quick_check_ok: quick_check_value == "ok",
1588 quick_check_value,
1589 checks,
1590 })
1591}
1592
1593pub(crate) fn ensure_integrity_ok(report: &IntegrityReport) -> Result<()> {
1594 let mut bad = vec![];
1595 if !report.quick_check_ok {
1596 bad.push("quick check");
1597 }
1598 for check in &report.checks {
1599 if !check.ok && check.label != "recurrence projection gaps" {
1600 bad.push(check.label);
1601 }
1602 }
1603 if bad.is_empty() {
1604 return Ok(());
1605 }
1606 bail!("error data-integrity-failed checks={}", bad.join(", "))
1607}
1608
1609async fn count_check(
1610 conn: &mut SqliteConnection,
1611 label: &'static str,
1612 query: &'static str,
1613) -> Result<IntegrityCheck> {
1614 let count: i64 = query_scalar(query).fetch_one(&mut *conn).await?;
1615 Ok(IntegrityCheck {
1616 label,
1617 ok: count == 0,
1618 value: format!("{count} orphaned"),
1619 })
1620}
1621
1622async fn push_meta_checks(
1623 conn: &mut SqliteConnection,
1624 checks: &mut Vec<IntegrityCheck>,
1625) -> Result<()> {
1626 let local_seq = db::get_meta(conn, "local_seq").await?;
1627 let local_seq_check = match local_seq {
1628 Some(raw) => match raw.parse::<i64>() {
1629 Ok(value) => {
1630 let max_seq: i64 = query_scalar("SELECT COALESCE(MAX(local_seq), 0) FROM changes")
1631 .fetch_one(&mut *conn)
1632 .await?;
1633 let ok = value >= max_seq;
1634 IntegrityCheck {
1635 label: "meta local_seq",
1636 ok,
1637 value: value.to_string(),
1638 }
1639 }
1640 Err(error) => IntegrityCheck {
1641 label: "meta local_seq",
1642 ok: false,
1643 value: error.to_string(),
1644 },
1645 },
1646 None => IntegrityCheck {
1647 label: "meta local_seq",
1648 ok: false,
1649 value: "missing".to_string(),
1650 },
1651 };
1652 checks.push(local_seq_check);
1653
1654 let sync_cursor = db::get_meta(conn, "sync_cursor").await?;
1655 let sync_cursor_ok = match sync_cursor {
1656 Some(raw) => match raw.parse::<i64>() {
1657 Ok(_) => IntegrityCheck {
1658 label: "sync cursor",
1659 ok: true,
1660 value: raw,
1661 },
1662 Err(error) => IntegrityCheck {
1663 label: "sync cursor",
1664 ok: false,
1665 value: error.to_string(),
1666 },
1667 },
1668 None => IntegrityCheck {
1669 label: "sync cursor",
1670 ok: false,
1671 value: "missing".to_string(),
1672 },
1673 };
1674 checks.push(sync_cursor_ok);
1675
1676 Ok(())
1677}