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