1use crate::choices::{TaskPriority, TaskSource, TaskStatus};
2use crate::ids::{MetadataFieldId, ProjectId, TaskId, WorkspaceId};
3use crate::recurrence::{
4 RecurrenceDuePolicy, RecurrenceFrequency, RecurrenceOutcome, RecurrenceProjectionState,
5 RecurrenceRule, RecurrenceSchedule, RecurrenceSeriesId, RecurrenceSeriesState, TimeZoneId,
6 WeekdaySet, derive_occurrence_identity, is_slot,
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 = 3;
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 #[serde(default)]
58 pub metadata_fields: Vec<MetadataFieldRow>,
59 #[serde(default)]
60 pub metadata_field_id_aliases: Vec<MetadataFieldIdAliasRow>,
61 pub tasks: Vec<TaskRow>,
62 #[serde(default)]
63 pub task_metadata: Vec<TaskMetadataRow>,
64 pub task_labels: Vec<TaskLabelRow>,
65 pub notes: Vec<NoteRow>,
66 pub task_dependencies: Vec<TaskDependencyRow>,
67 pub task_epic_links: Vec<TaskEpicLinkRow>,
68 #[serde(default)]
69 pub task_related_links: Vec<TaskRelatedLinkRow>,
70 #[serde(default)]
71 pub task_attachments: Vec<TaskAttachmentRow>,
72 #[serde(default)]
73 pub blob_inventory: Vec<BlobInventoryExportRow>,
74 #[serde(default)]
75 pub recurrence_series: Vec<RecurrenceSeriesRow>,
76 #[serde(default)]
77 pub recurrence_series_labels: Vec<RecurrenceSeriesLabelRow>,
78 #[serde(default)]
79 pub recurrence_series_metadata: Vec<RecurrenceSeriesMetadataRow>,
80 #[serde(default)]
81 pub recurrence_occurrences: Vec<RecurrenceOccurrenceRow>,
82 #[serde(default)]
83 pub recurrence_pause_intervals: Vec<RecurrencePauseIntervalRow>,
84 pub changes: Vec<ChangeRow>,
85 pub field_versions: Vec<FieldVersionRow>,
86 pub conflicts: Vec<ConflictRow>,
87 pub meta: Vec<MetaRow>,
88}
89
90#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
91pub struct WorkspaceRow {
92 pub id: WorkspaceId,
93 pub name: String,
94 pub key: String,
95 pub created_at: String,
96 pub updated_at: String,
97 pub archived: i64,
98}
99
100#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
101pub struct ProjectRow {
102 pub id: ProjectId,
103 pub workspace_id: WorkspaceId,
104 pub key: String,
105 pub name: String,
106 pub prefix: String,
107 pub created_at: String,
108 pub updated_at: String,
109 pub deleted: i64,
110}
111
112#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
113pub struct ProjectPathRow {
114 pub workspace_id: WorkspaceId,
115 pub project_id: ProjectId,
116 pub path: String,
117}
118
119#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
120pub struct ProjectIdAliasRow {
121 pub workspace_id: WorkspaceId,
122 pub remote_project_id: ProjectId,
123 pub local_project_id: ProjectId,
124}
125
126#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
127pub struct LabelRow {
128 pub workspace_id: WorkspaceId,
129 pub name: String,
130 pub created_at: String,
131}
132
133#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
134pub struct MetadataFieldRow {
135 pub id: MetadataFieldId,
136 pub workspace_id: WorkspaceId,
137 pub key: String,
138 pub created_at: String,
139 pub updated_at: String,
140}
141
142#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
143pub struct MetadataFieldIdAliasRow {
144 pub workspace_id: WorkspaceId,
145 pub remote_field_id: MetadataFieldId,
146 pub local_field_id: MetadataFieldId,
147}
148
149#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
150pub struct TaskMetadataRow {
151 pub workspace_id: WorkspaceId,
152 pub task_id: TaskId,
153 pub field_id: MetadataFieldId,
154 pub value: String,
155 pub created_at: String,
156 pub updated_at: String,
157}
158
159fn default_task_source() -> String {
160 TaskSource::Unknown.as_str().to_string()
161}
162
163#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
164pub struct TaskRow {
165 pub workspace_id: WorkspaceId,
166 pub id: TaskId,
167 pub title: String,
168 pub description: String,
169 pub project_id: ProjectId,
170 pub status: String,
171 pub priority: String,
172 #[serde(default = "default_task_source")]
173 pub source: String,
174 pub created_at: String,
175 pub updated_at: String,
176 pub queue_activity_at: String,
177 #[serde(default)]
178 pub available_at: String,
179 #[serde(default)]
180 pub due_on: String,
181 pub deleted: i64,
182 pub is_epic: i64,
183}
184
185#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
186pub struct TaskEpicLinkRow {
187 pub workspace_id: WorkspaceId,
188 pub child_task_id: TaskId,
189 pub epic_task_id: TaskId,
190 pub created_at: String,
191}
192
193#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
194pub struct TaskLabelRow {
195 pub workspace_id: WorkspaceId,
196 pub task_id: TaskId,
197 pub label: String,
198}
199
200#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
201pub struct NoteRow {
202 pub workspace_id: WorkspaceId,
203 pub id: String,
204 pub task_id: TaskId,
205 pub body: String,
206 pub created_at: String,
207 pub change_id: String,
208}
209
210#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
211pub struct TaskDependencyRow {
212 pub workspace_id: WorkspaceId,
213 pub task_id: TaskId,
214 pub depends_on_task_id: TaskId,
215 pub created_at: String,
216}
217
218#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
219pub struct TaskRelatedLinkRow {
220 pub workspace_id: WorkspaceId,
221 pub task_a_id: TaskId,
222 pub task_b_id: TaskId,
223 pub linked: i64,
224 pub last_change_id: String,
225}
226
227#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
228pub struct TaskAttachmentRow {
229 pub workspace_id: WorkspaceId,
230 pub attachment_id: String,
231 pub task_id: TaskId,
232 pub sha256: String,
233 pub byte_size: i64,
234 pub media_type: String,
235 pub filename: Option<String>,
236 pub alt_text: Option<String>,
237 pub width: Option<i64>,
238 pub height: Option<i64>,
239 pub created_at: String,
240 pub created_by_change_id: Option<String>,
241 pub deleted: i64,
242 pub deleted_at: Option<String>,
243 pub deleted_by_change_id: Option<String>,
244}
245
246#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
247pub struct BlobInventoryExportRow {
248 pub sha256: String,
249 pub byte_size: i64,
250 pub media_type: String,
251 pub available: i64,
252 pub first_seen_at: String,
253 pub last_verified_at: Option<String>,
254}
255
256#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
257pub struct RecurrenceSeriesRow {
258 pub workspace_id: WorkspaceId,
259 pub id: RecurrenceSeriesId,
260 pub title: String,
261 pub description: String,
262 pub project_id: ProjectId,
263 pub priority: String,
264 pub initial_status: String,
265 pub frequency: String,
266 pub interval: i64,
267 pub weekdays: String,
268 pub timezone: String,
269 pub start_on: String,
270 pub available_local_time: String,
271 pub due_policy: String,
272 pub state: String,
273 pub stopped_at: String,
274 pub created_at: String,
275 pub updated_at: String,
276 pub deleted: i64,
277}
278
279#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
280pub struct RecurrenceSeriesLabelRow {
281 pub workspace_id: WorkspaceId,
282 pub series_id: RecurrenceSeriesId,
283 pub label: String,
284}
285
286#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
287pub struct RecurrenceSeriesMetadataRow {
288 pub workspace_id: WorkspaceId,
289 pub series_id: RecurrenceSeriesId,
290 pub field_id: MetadataFieldId,
291 pub value: String,
292 pub created_at: String,
293 pub updated_at: String,
294}
295
296#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
297pub struct RecurrenceOccurrenceRow {
298 pub workspace_id: WorkspaceId,
299 pub series_id: RecurrenceSeriesId,
300 pub slot_on: String,
301 pub task_id: String,
302 pub outcome: String,
303 pub resolved_at: String,
304 pub outcome_change_id: String,
305 pub projection_state: String,
306 pub archived_at: String,
307}
308
309#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
310pub struct RecurrencePauseIntervalRow {
311 pub workspace_id: WorkspaceId,
312 pub id: String,
313 pub series_id: RecurrenceSeriesId,
314 pub paused_at: String,
315 pub resumed_at: String,
316 pub suspended_slot_on: String,
317 pub suspended_task_id: String,
318 pub created_by_change_id: String,
319 pub resolved_by_change_id: String,
320}
321
322#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
323pub struct ChangeRow {
324 pub change_id: String,
325 pub client_id: String,
326 pub local_seq: i64,
327 pub entity_type: String,
328 pub entity_id: String,
329 pub field: Option<String>,
330 pub op_type: String,
331 pub payload: String,
332 pub base_version: Option<String>,
333 pub created_at: String,
334 pub server_seq: Option<i64>,
335}
336
337#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
338pub struct FieldVersionRow {
339 pub workspace_id: WorkspaceId,
340 pub entity_type: String,
341 pub entity_id: String,
342 pub field: String,
343 pub version: String,
344}
345
346#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
347pub struct ConflictRow {
348 pub id: i64,
349 pub workspace_id: WorkspaceId,
350 pub entity_type: String,
351 pub entity_id: String,
352 pub task_id: String,
353 pub field: String,
354 pub base_version: Option<String>,
355 pub local_value: String,
356 pub remote_value: String,
357 pub local_change_id: Option<String>,
358 pub remote_change_id: String,
359 pub variant_a: String,
360 pub variant_b: String,
361 pub created_at: String,
362 pub resolved: i64,
363}
364
365#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
366pub struct MetaRow {
367 pub key: String,
368 pub value: String,
369}
370
371impl Database {
372 pub async fn export_data(&self, exported_at: String) -> Result<AvenExport> {
373 let mut conn = self.acquire_writer().await?;
374 let mut tx = db::begin_immediate(&mut conn).await?;
375 let schema_version = db::current_schema_version(&mut tx).await?;
376 let tables = ExportTables {
377 workspaces: scan_workspaces(&mut tx).await?,
378 projects: scan_projects(&mut tx).await?,
379 project_paths: scan_project_paths(&mut tx).await?,
380 project_id_aliases: scan_project_id_aliases(&mut tx).await?,
381 labels: scan_labels(&mut tx).await?,
382 metadata_fields: scan_metadata_fields(&mut tx).await?,
383 metadata_field_id_aliases: scan_metadata_field_id_aliases(&mut tx).await?,
384 tasks: scan_tasks(&mut tx).await?,
385 task_metadata: scan_task_metadata(&mut tx).await?,
386 task_labels: scan_task_labels(&mut tx).await?,
387 notes: scan_notes(&mut tx).await?,
388 task_dependencies: scan_task_dependencies(&mut tx).await?,
389 task_epic_links: scan_task_epic_links(&mut tx).await?,
390 task_related_links: scan_task_related_links(&mut tx).await?,
391 task_attachments: scan_task_attachments(&mut tx).await?,
392 blob_inventory: scan_blob_inventory(&mut tx).await?,
393 recurrence_series: scan_recurrence_series(&mut tx).await?,
394 recurrence_series_labels: scan_recurrence_series_labels(&mut tx).await?,
395 recurrence_series_metadata: scan_recurrence_series_metadata(&mut tx).await?,
396 recurrence_occurrences: scan_recurrence_occurrences(&mut tx).await?,
397 recurrence_pause_intervals: scan_recurrence_pause_intervals(&mut tx).await?,
398 changes: scan_changes(&mut tx).await?,
399 field_versions: scan_field_versions(&mut tx).await?,
400 conflicts: scan_conflicts(&mut tx).await?,
401 meta: scan_meta(&mut tx).await?,
402 };
403 let version = if tables.task_related_links.is_empty() {
404 2
405 } else {
406 EXPORT_VERSION
407 };
408 tx.commit().await?;
409 Ok(AvenExport {
410 format: EXPORT_FORMAT.to_string(),
411 version,
412 exported_at,
413 schema_version,
414 blobs_included: false,
415 tables,
416 })
417 }
418
419 pub async fn validate_import_data(&self, export: &AvenExport) -> Result<()> {
420 let mut conn = self.acquire_reader().await?;
421 ensure_supported_export(&mut conn, export).await?;
422 validate_export_snapshot(export)
423 }
424
425 pub async fn import_data(&self, export: &AvenExport) -> Result<IntegrityReport> {
426 let mut conn = self.acquire_writer().await?;
427 ensure_supported_export(&mut conn, export).await?;
428 validate_export_snapshot(export)?;
429 let target_client_id = db::get_meta(&mut conn, "client_id")
430 .await?
431 .context("missing target client_id")?;
432 let mut tx = db::begin_immediate(&mut conn).await?;
433 replace_from_export(&mut tx, export, &target_client_id).await?;
434 let report = database_integrity_report_with_connection(&mut tx).await?;
435 ensure_integrity_ok(&report)?;
436 tx.commit().await?;
437 Ok(report)
438 }
439
440 pub async fn database_integrity_report(&self) -> Result<IntegrityReport> {
441 let mut conn = self.acquire_reader().await?;
442 database_integrity_report_with_connection(&mut conn).await
443 }
444
445 pub async fn attachment_integrity_checks(
446 &self,
447 blob_dir: &Path,
448 deep: bool,
449 ) -> Result<Vec<IntegrityCheck>> {
450 let mut conn = self.acquire_reader().await?;
451 integrity::attachment_integrity_checks(&mut conn, blob_dir, deep).await
452 }
453
454 pub async fn create_backup_archive(&self, blob_dir: &Path, output: &Path) -> Result<()> {
455 let mut conn = self.acquire_writer().await?;
456 let hashes: Vec<String> = sqlx::query_scalar(
457 "SELECT sha256 FROM blob_inventory WHERE available = 1 ORDER BY sha256",
458 )
459 .fetch_all(&mut *conn)
460 .await?;
461 let mut leases = Vec::with_capacity(hashes.len());
462 for hash in hashes {
463 match crate::attachments::lifecycle::acquire_lease(
464 &mut conn,
465 &hash,
466 "backup",
467 &crate::attachments::lifecycle::SystemClock,
468 )
469 .await
470 {
471 Ok(lease) => leases.push(lease),
472 Err(error) => {
473 for lease in leases {
474 let _ =
475 crate::attachments::lifecycle::release_lease(&mut conn, &lease).await;
476 }
477 return Err(error);
478 }
479 }
480 }
481 let backup_result = archive::create_backup_archive(&mut conn, blob_dir, output).await;
482 for lease in leases {
483 crate::attachments::lifecycle::release_lease(&mut conn, &lease).await?;
484 }
485 backup_result
486 }
487}
488
489pub fn is_backup_archive(path: &Path) -> Result<bool> {
490 archive::is_archive_path(path)
491}
492
493pub async fn restore_backup_archive(
494 db_path: &Path,
495 blob_dir: &Path,
496 source: &Path,
497) -> Result<PathBuf> {
498 archive::restore_backup_archive(db_path, blob_dir, source).await
499}
500
501async fn scan_workspaces(conn: &mut SqliteConnection) -> Result<Vec<WorkspaceRow>> {
502 tables::scan_rows(
503 conn,
504 "SELECT id, name, key, created_at, updated_at, archived FROM workspaces",
505 )
506 .await
507}
508
509async fn scan_projects(conn: &mut SqliteConnection) -> Result<Vec<ProjectRow>> {
510 tables::scan_rows(
511 conn,
512 "SELECT id, workspace_id, key, name, prefix, created_at, updated_at, deleted FROM projects",
513 )
514 .await
515}
516
517async fn scan_project_paths(conn: &mut SqliteConnection) -> Result<Vec<ProjectPathRow>> {
518 tables::scan_rows(
519 conn,
520 "SELECT workspace_id, project_id, path FROM project_paths",
521 )
522 .await
523}
524
525async fn scan_project_id_aliases(conn: &mut SqliteConnection) -> Result<Vec<ProjectIdAliasRow>> {
526 tables::scan_rows(
527 conn,
528 "SELECT workspace_id, remote_project_id, local_project_id FROM project_id_aliases",
529 )
530 .await
531}
532
533async fn scan_labels(conn: &mut SqliteConnection) -> Result<Vec<LabelRow>> {
534 tables::scan_rows(conn, "SELECT workspace_id, name, created_at FROM labels").await
535}
536
537async fn scan_metadata_fields(conn: &mut SqliteConnection) -> Result<Vec<MetadataFieldRow>> {
538 tables::scan_rows(
539 conn,
540 "SELECT id, workspace_id, key, created_at, updated_at FROM metadata_fields",
541 )
542 .await
543}
544
545async fn scan_metadata_field_id_aliases(
546 conn: &mut SqliteConnection,
547) -> Result<Vec<MetadataFieldIdAliasRow>> {
548 tables::scan_rows(
549 conn,
550 "SELECT workspace_id, remote_field_id, local_field_id FROM metadata_field_id_aliases",
551 )
552 .await
553}
554
555async fn scan_task_metadata(conn: &mut SqliteConnection) -> Result<Vec<TaskMetadataRow>> {
556 tables::scan_rows(
557 conn,
558 "SELECT workspace_id, task_id, field_id, value, created_at, updated_at FROM task_metadata",
559 )
560 .await
561}
562
563async fn scan_tasks(conn: &mut SqliteConnection) -> Result<Vec<TaskRow>> {
564 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
565}
566
567async fn scan_task_labels(conn: &mut SqliteConnection) -> Result<Vec<TaskLabelRow>> {
568 tables::scan_rows(conn, "SELECT workspace_id, task_id, label FROM task_labels").await
569}
570
571async fn scan_notes(conn: &mut SqliteConnection) -> Result<Vec<NoteRow>> {
572 tables::scan_rows(
573 conn,
574 "SELECT workspace_id, id, task_id, body, created_at, change_id FROM notes",
575 )
576 .await
577}
578
579async fn scan_task_dependencies(conn: &mut SqliteConnection) -> Result<Vec<TaskDependencyRow>> {
580 tables::scan_rows(
581 conn,
582 "SELECT workspace_id, task_id, depends_on_task_id, created_at FROM task_dependencies",
583 )
584 .await
585}
586
587async fn scan_task_epic_links(conn: &mut SqliteConnection) -> Result<Vec<TaskEpicLinkRow>> {
588 tables::scan_rows(
589 conn,
590 "SELECT workspace_id, child_task_id, epic_task_id, created_at FROM task_epic_links",
591 )
592 .await
593}
594
595async fn scan_task_related_links(conn: &mut SqliteConnection) -> Result<Vec<TaskRelatedLinkRow>> {
596 tables::scan_rows(
597 conn,
598 "SELECT workspace_id, task_a_id, task_b_id, linked, last_change_id FROM task_related_links",
599 )
600 .await
601}
602
603async fn scan_task_attachments(conn: &mut SqliteConnection) -> Result<Vec<TaskAttachmentRow>> {
604 tables::scan_rows(
605 conn,
606 "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",
607 )
608 .await
609}
610
611async fn scan_blob_inventory(conn: &mut SqliteConnection) -> Result<Vec<BlobInventoryExportRow>> {
612 tables::scan_rows(
613 conn,
614 "SELECT sha256, byte_size, media_type, available, first_seen_at, last_verified_at FROM blob_inventory",
615 )
616 .await
617}
618
619async fn scan_recurrence_series(conn: &mut SqliteConnection) -> Result<Vec<RecurrenceSeriesRow>> {
620 tables::scan_rows(
621 conn,
622 "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",
623 )
624 .await
625}
626
627async fn scan_recurrence_series_labels(
628 conn: &mut SqliteConnection,
629) -> Result<Vec<RecurrenceSeriesLabelRow>> {
630 tables::scan_rows(
631 conn,
632 "SELECT workspace_id, series_id, label FROM recurrence_series_labels",
633 )
634 .await
635}
636
637async fn scan_recurrence_series_metadata(
638 conn: &mut SqliteConnection,
639) -> Result<Vec<RecurrenceSeriesMetadataRow>> {
640 tables::scan_rows(
641 conn,
642 "SELECT workspace_id, series_id, field_id, value, created_at, updated_at
643 FROM recurrence_series_metadata",
644 )
645 .await
646}
647
648async fn scan_recurrence_occurrences(
649 conn: &mut SqliteConnection,
650) -> Result<Vec<RecurrenceOccurrenceRow>> {
651 tables::scan_rows(
652 conn,
653 "SELECT workspace_id, series_id, slot_on, task_id, outcome, resolved_at, outcome_change_id, projection_state, archived_at FROM recurrence_occurrences",
654 )
655 .await
656}
657
658async fn scan_recurrence_pause_intervals(
659 conn: &mut SqliteConnection,
660) -> Result<Vec<RecurrencePauseIntervalRow>> {
661 tables::scan_rows(
662 conn,
663 "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",
664 )
665 .await
666}
667
668async fn scan_changes(conn: &mut SqliteConnection) -> Result<Vec<ChangeRow>> {
669 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
670}
671
672async fn scan_field_versions(conn: &mut SqliteConnection) -> Result<Vec<FieldVersionRow>> {
673 tables::scan_rows(
674 conn,
675 "SELECT workspace_id, entity_type, entity_id, field, version FROM field_versions",
676 )
677 .await
678}
679
680async fn scan_conflicts(conn: &mut SqliteConnection) -> Result<Vec<ConflictRow>> {
681 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
682}
683
684async fn scan_meta(conn: &mut SqliteConnection) -> Result<Vec<MetaRow>> {
685 tables::scan_rows(conn, "SELECT key, value FROM meta").await
686}
687
688async fn ensure_supported_export(_conn: &mut SqliteConnection, export: &AvenExport) -> Result<()> {
689 if export.format != EXPORT_FORMAT {
690 bail!("error export-format-unsupported format={}", export.format);
691 }
692 if !matches!(export.version, 1 | 2 | EXPORT_VERSION) {
693 bail!(
694 "error export-version-unsupported version={}",
695 export.version
696 );
697 }
698 Ok(())
699}
700
701fn validate_export_snapshot(export: &AvenExport) -> Result<()> {
702 ensure!(
703 export.version == EXPORT_VERSION
704 || (export.tables.task_related_links.is_empty()
705 && !export.tables.changes.iter().any(|change| matches!(
706 change.op_type.as_str(),
707 "related_add" | "related_remove"
708 ))),
709 "error invalid-export-snapshot related links require version {EXPORT_VERSION}; related changes are not supported in older versions"
710 );
711 let mut workspace_ids = HashSet::new();
712 for workspace in &export.tables.workspaces {
713 if workspace_ids.contains(&workspace.id) {
714 continue;
715 }
716 workspace_ids.insert(workspace.id.clone());
717 }
718
719 let mut project_ids: HashMap<WorkspaceId, HashSet<ProjectId>> = HashMap::new();
720 for project in &export.tables.projects {
721 if !workspace_ids.contains(&project.workspace_id) {
722 bail!(
723 "error invalid-export-snapshot project.workspace_id={} is missing",
724 project.workspace_id
725 );
726 }
727 project_ids
728 .entry(project.workspace_id.clone())
729 .or_default()
730 .insert(project.id.clone());
731 }
732
733 for path in &export.tables.project_paths {
734 let projects = project_ids.get(&path.workspace_id).ok_or_else(|| {
735 anyhow::Error::msg(format!(
736 "error invalid-export-snapshot project_path.workspace_id={} is missing",
737 path.workspace_id
738 ))
739 })?;
740 if !projects.contains(&path.project_id) {
741 bail!(
742 "error invalid-export-snapshot project_path.project_id={} is missing in workspace {}",
743 path.project_id,
744 path.workspace_id
745 );
746 }
747 }
748
749 let mut label_keys: HashSet<(WorkspaceId, String)> = HashSet::new();
750 for label in &export.tables.labels {
751 if !workspace_ids.contains(&label.workspace_id) {
752 bail!(
753 "error invalid-export-snapshot label.workspace_id={} is missing",
754 label.workspace_id
755 );
756 }
757 label_keys.insert((label.workspace_id.clone(), label.name.clone()));
758 }
759
760 let mut task_ids: HashMap<WorkspaceId, HashSet<TaskId>> = HashMap::new();
761 for task in &export.tables.tasks {
762 TaskSource::parse(&task.source)?;
763 if let Err(error) = crate::time_validation::validate_due_on_value(&task.due_on) {
764 bail!(
765 "error invalid-export-snapshot task.due_on={} is invalid: {error}",
766 task.due_on
767 );
768 }
769 let workspace_projects = project_ids.get(&task.workspace_id).ok_or_else(|| {
770 anyhow::Error::msg(format!(
771 "error invalid-export-snapshot task.workspace_id={} is missing",
772 task.workspace_id
773 ))
774 })?;
775 if !workspace_projects.contains(&task.project_id) {
776 bail!(
777 "error invalid-export-snapshot task.project_id={} is missing in workspace {}",
778 task.project_id,
779 task.workspace_id
780 );
781 }
782 task_ids
783 .entry(task.workspace_id.clone())
784 .or_default()
785 .insert(task.id.clone());
786 }
787
788 for task_label in &export.tables.task_labels {
789 let task_workspace = task_ids.get(&task_label.workspace_id).ok_or_else(|| {
790 anyhow::Error::msg(format!(
791 "error invalid-export-snapshot task_label.workspace_id={} is missing",
792 task_label.workspace_id
793 ))
794 })?;
795 if !task_workspace.contains(&task_label.task_id) {
796 bail!(
797 "error invalid-export-snapshot task_label.task_id={} is missing in workspace {}",
798 task_label.task_id,
799 task_label.workspace_id
800 );
801 }
802 if !label_keys.contains(&(task_label.workspace_id.clone(), task_label.label.clone())) {
803 bail!(
804 "error invalid-export-snapshot task_label.label={} is missing in workspace {}",
805 task_label.label,
806 task_label.workspace_id
807 );
808 }
809 }
810
811 for note in &export.tables.notes {
812 let task_workspace = task_ids.get(¬e.workspace_id).ok_or_else(|| {
813 anyhow::Error::msg(format!(
814 "error invalid-export-snapshot note.workspace_id={} is missing",
815 note.workspace_id
816 ))
817 })?;
818 if !task_workspace.contains(¬e.task_id) {
819 bail!(
820 "error invalid-export-snapshot note.task_id={} is missing in workspace {}",
821 note.task_id,
822 note.workspace_id
823 );
824 }
825 }
826
827 for dep in &export.tables.task_dependencies {
828 let tasks = task_ids.get(&dep.workspace_id).ok_or_else(|| {
829 anyhow::Error::msg(format!(
830 "error invalid-export-snapshot dependency.workspace_id={} is missing",
831 dep.workspace_id
832 ))
833 })?;
834 if !tasks.contains(&dep.task_id) || !tasks.contains(&dep.depends_on_task_id) {
835 bail!(
836 "error invalid-export-snapshot task_dependencies are missing tasks in workspace {}",
837 dep.workspace_id
838 );
839 }
840 }
841
842 for epic_link in &export.tables.task_epic_links {
843 let tasks = task_ids.get(&epic_link.workspace_id).ok_or_else(|| {
844 anyhow::Error::msg(format!(
845 "error invalid-export-snapshot epic_link.workspace_id={} is missing",
846 epic_link.workspace_id
847 ))
848 })?;
849 if !tasks.contains(&epic_link.child_task_id) || !tasks.contains(&epic_link.epic_task_id) {
850 bail!(
851 "error invalid-export-snapshot task_epic_links are missing tasks in workspace {}",
852 epic_link.workspace_id
853 );
854 }
855 }
856
857 let changes_by_id = export
858 .tables
859 .changes
860 .iter()
861 .map(|change| (change.change_id.as_str(), change))
862 .collect::<HashMap<_, _>>();
863 for link in &export.tables.task_related_links {
864 ensure!(
865 link.task_a_id < link.task_b_id,
866 "error invalid-export-snapshot related pair is not canonical"
867 );
868 ensure!(
869 matches!(link.linked, 0 | 1),
870 "error invalid-export-snapshot related linked value is invalid"
871 );
872 let tasks = task_ids
873 .get(&link.workspace_id)
874 .context("error invalid-export-snapshot related workspace is missing")?;
875 ensure!(
876 tasks.contains(&link.task_a_id) && tasks.contains(&link.task_b_id),
877 "error invalid-export-snapshot related endpoints are missing"
878 );
879 let change = changes_by_id
880 .get(link.last_change_id.as_str())
881 .context("error invalid-export-snapshot related change is missing")?;
882 let payload: Value = serde_json::from_str(&change.payload)?;
883 let related_id: TaskId = payload
884 .get("related_task_id")
885 .and_then(Value::as_str)
886 .context("error invalid-export-snapshot related payload is invalid")?
887 .parse()?;
888 let initiating_id: TaskId = change.entity_id.parse()?;
889 let (change_a, change_b) =
890 crate::operations::canonical_related_pair(&initiating_id, &related_id)?;
891 ensure!(
892 change.entity_type == "task"
893 && change.field.as_deref() == Some("related")
894 && payload.get("workspace_id").and_then(Value::as_str)
895 == Some(link.workspace_id.as_str())
896 && change_a == &link.task_a_id
897 && change_b == &link.task_b_id
898 && ((link.linked == 1 && change.op_type == "related_add")
899 || (link.linked == 0 && change.op_type == "related_remove")),
900 "error invalid-export-snapshot related change does not match state"
901 );
902 }
903
904 let mut inventory = HashMap::new();
905 for blob in &export.tables.blob_inventory {
906 crate::attachments::validate_sha256(&blob.sha256)?;
907 crate::attachments::validate_media_type(&blob.media_type)?;
908 crate::attachments::validate_blob_size(usize::try_from(blob.byte_size).unwrap_or(0))?;
909 if blob.available != 0 && blob.available != 1 {
910 bail!("error invalid-export-snapshot blob_inventory.available invalid");
911 }
912 if inventory
913 .insert(
914 blob.sha256.clone(),
915 (blob.byte_size, blob.media_type.as_str()),
916 )
917 .is_some()
918 {
919 bail!("error invalid-export-snapshot blob_inventory.sha256 duplicate");
920 }
921 }
922
923 for attachment in &export.tables.task_attachments {
924 crate::attachments::validate_attachment_id(&attachment.attachment_id)?;
925 crate::attachments::validate_sha256(&attachment.sha256)?;
926 crate::attachments::validate_media_type(&attachment.media_type)?;
927 crate::attachments::validate_blob_size(usize::try_from(attachment.byte_size).unwrap_or(0))?;
928 crate::attachments::validate_filename(attachment.filename.as_deref())?;
929 crate::attachments::validate_alt_text(attachment.alt_text.as_deref())?;
930 crate::attachments::validate_dimensions(attachment.width, attachment.height)?;
931 let tasks = task_ids.get(&attachment.workspace_id).ok_or_else(|| {
932 anyhow::Error::msg(format!(
933 "error invalid-export-snapshot attachment.workspace_id={} is missing",
934 attachment.workspace_id
935 ))
936 })?;
937 if !tasks.contains(&attachment.task_id) {
938 bail!(
939 "error invalid-export-snapshot attachment.task_id={} is missing in workspace {}",
940 attachment.task_id,
941 attachment.workspace_id
942 );
943 }
944 let Some((inventory_size, inventory_media_type)) = inventory.get(&attachment.sha256) else {
945 bail!("error invalid-export-snapshot attachment inventory missing");
946 };
947 if *inventory_size != attachment.byte_size || *inventory_media_type != attachment.media_type
948 {
949 bail!("error invalid-export-snapshot attachment inventory metadata mismatch");
950 }
951 if attachment.deleted != 0 && attachment.deleted != 1 {
952 bail!(
953 "error invalid-export-snapshot attachment.deleted={} for attachment {}",
954 attachment.deleted,
955 attachment.attachment_id
956 );
957 }
958 }
959
960 for alias in &export.tables.project_id_aliases {
961 let workspace_projects = project_ids.get(&alias.workspace_id).ok_or_else(|| {
962 anyhow::Error::msg(format!(
963 "error invalid-export-snapshot project_alias.workspace_id={} is missing",
964 alias.workspace_id
965 ))
966 })?;
967 if !workspace_projects.contains(&alias.local_project_id) {
968 bail!(
969 "error invalid-export-snapshot local_project_id={} is missing in workspace {}",
970 alias.local_project_id,
971 alias.workspace_id
972 );
973 }
974 }
975
976 let mut metadata_ids = HashSet::new();
977 let mut metadata_keys = HashSet::new();
978 for field in &export.tables.metadata_fields {
979 ensure!(
980 workspace_ids.contains(&field.workspace_id),
981 "error invalid-export-snapshot metadata_field.workspace_id is missing"
982 );
983 let normalized = crate::metadata::normalize_metadata_key(&field.key)?;
984 ensure!(
985 normalized == field.key,
986 "error invalid-export-snapshot metadata field key is noncanonical"
987 );
988 ensure!(
989 metadata_ids.insert((field.workspace_id.clone(), field.id.clone())),
990 "error invalid-export-snapshot metadata field identity is duplicated"
991 );
992 ensure!(
993 metadata_keys.insert((field.workspace_id.clone(), field.key.clone())),
994 "error invalid-export-snapshot metadata field key is duplicated"
995 );
996 }
997 let mut metadata_aliases = HashSet::new();
998 for alias in &export.tables.metadata_field_id_aliases {
999 ensure!(
1000 metadata_ids.contains(&(alias.workspace_id.clone(), alias.local_field_id.clone())),
1001 "error invalid-export-snapshot metadata alias target is missing"
1002 );
1003 ensure!(
1004 metadata_aliases.insert((alias.workspace_id.clone(), alias.remote_field_id.clone())),
1005 "error invalid-export-snapshot metadata alias identity is duplicated"
1006 );
1007 }
1008 let mut task_metadata_keys = HashSet::new();
1009 let mut task_metadata_usage: HashMap<(WorkspaceId, TaskId), (usize, usize)> = HashMap::new();
1010 for value in &export.tables.task_metadata {
1011 ensure!(
1012 task_ids
1013 .get(&value.workspace_id)
1014 .is_some_and(|tasks| tasks.contains(&value.task_id)),
1015 "error invalid-export-snapshot task metadata task is missing"
1016 );
1017 ensure!(
1018 metadata_ids.contains(&(value.workspace_id.clone(), value.field_id.clone())),
1019 "error invalid-export-snapshot task metadata field is missing"
1020 );
1021 ensure!(
1022 value.value.len() <= crate::metadata::MAX_METADATA_VALUE_BYTES,
1023 "error invalid-export-snapshot task metadata value is too large"
1024 );
1025 ensure!(
1026 task_metadata_keys.insert((
1027 value.workspace_id.clone(),
1028 value.task_id.clone(),
1029 value.field_id.clone(),
1030 )),
1031 "error invalid-export-snapshot task metadata identity is duplicated"
1032 );
1033 let usage = task_metadata_usage
1034 .entry((value.workspace_id.clone(), value.task_id.clone()))
1035 .or_default();
1036 usage.0 += 1;
1037 usage.1 += value.value.len();
1038 }
1039 ensure!(
1040 task_metadata_usage.values().all(|(count, bytes)| {
1041 *count <= crate::metadata::MAX_METADATA_VALUES
1042 && *bytes <= crate::metadata::MAX_METADATA_TOTAL_BYTES
1043 }),
1044 "error invalid-export-snapshot task metadata limits exceeded"
1045 );
1046
1047 let series_ids = export
1048 .tables
1049 .recurrence_series
1050 .iter()
1051 .map(|series| (series.workspace_id.clone(), series.id.clone()))
1052 .collect::<HashSet<_>>();
1053 let mut series_metadata_keys = HashSet::new();
1054 let mut series_metadata_usage: HashMap<(WorkspaceId, RecurrenceSeriesId), (usize, usize)> =
1055 HashMap::new();
1056 for value in &export.tables.recurrence_series_metadata {
1057 ensure!(
1058 series_ids.contains(&(value.workspace_id.clone(), value.series_id.clone())),
1059 "error invalid-export-snapshot recurrence metadata series is missing"
1060 );
1061 ensure!(
1062 metadata_ids.contains(&(value.workspace_id.clone(), value.field_id.clone())),
1063 "error invalid-export-snapshot recurrence metadata field is missing"
1064 );
1065 ensure!(
1066 value.value.len() <= crate::metadata::MAX_METADATA_VALUE_BYTES,
1067 "error invalid-export-snapshot recurrence metadata value is too large"
1068 );
1069 ensure!(
1070 series_metadata_keys.insert((
1071 value.workspace_id.clone(),
1072 value.series_id.clone(),
1073 value.field_id.clone(),
1074 )),
1075 "error invalid-export-snapshot recurrence metadata identity is duplicated"
1076 );
1077 let usage = series_metadata_usage
1078 .entry((value.workspace_id.clone(), value.series_id.clone()))
1079 .or_default();
1080 usage.0 += 1;
1081 usage.1 += value.value.len();
1082 }
1083 ensure!(
1084 series_metadata_usage.values().all(|(count, bytes)| {
1085 *count <= crate::metadata::MAX_METADATA_VALUES
1086 && *bytes <= crate::metadata::MAX_METADATA_TOTAL_BYTES
1087 }),
1088 "error invalid-export-snapshot recurrence metadata limits exceeded"
1089 );
1090
1091 if has_recurrence_data(export) {
1092 validate_recurrence_snapshot(export, &workspace_ids, &project_ids, &label_keys, &task_ids)?;
1093 }
1094
1095 Ok(())
1096}
1097
1098fn has_recurrence_data(export: &AvenExport) -> bool {
1099 !export.tables.recurrence_series.is_empty()
1100 || !export.tables.recurrence_series_labels.is_empty()
1101 || !export.tables.recurrence_series_metadata.is_empty()
1102 || !export.tables.recurrence_occurrences.is_empty()
1103 || !export.tables.recurrence_pause_intervals.is_empty()
1104}
1105
1106fn validate_recurrence_snapshot(
1107 export: &AvenExport,
1108 workspace_ids: &HashSet<WorkspaceId>,
1109 project_ids: &HashMap<WorkspaceId, HashSet<ProjectId>>,
1110 label_keys: &HashSet<(WorkspaceId, String)>,
1111 task_ids: &HashMap<WorkspaceId, HashSet<TaskId>>,
1112) -> Result<()> {
1113 struct ValidatedSeries<'a> {
1114 row: &'a RecurrenceSeriesRow,
1115 schedule: RecurrenceSchedule,
1116 state: RecurrenceSeriesState,
1117 created_at: DateTime<chrono::FixedOffset>,
1118 stopped_at: Option<DateTime<chrono::FixedOffset>>,
1119 }
1120
1121 let task_rows = export
1122 .tables
1123 .tasks
1124 .iter()
1125 .map(|task| ((task.workspace_id.clone(), task.id.clone()), task))
1126 .collect::<HashMap<_, _>>();
1127 let mut change_rows = HashMap::new();
1128 for change in &export.tables.changes {
1129 ensure!(
1130 change_rows
1131 .insert(change.change_id.as_str(), change)
1132 .is_none(),
1133 "error invalid-export-snapshot change.change_id={} is duplicated",
1134 change.change_id
1135 );
1136 }
1137 let field_versions = export
1138 .tables
1139 .field_versions
1140 .iter()
1141 .map(|row| {
1142 (
1143 (
1144 row.workspace_id.clone(),
1145 row.entity_type.as_str(),
1146 row.entity_id.as_str(),
1147 row.field.as_str(),
1148 ),
1149 row.version.as_str(),
1150 )
1151 })
1152 .collect::<HashMap<_, _>>();
1153
1154 let mut series_by_id = HashMap::new();
1155 for row in &export.tables.recurrence_series {
1156 ensure!(
1157 workspace_ids.contains(&row.workspace_id),
1158 "error invalid-export-snapshot recurrence_series.workspace_id={} is missing",
1159 row.workspace_id
1160 );
1161 ensure!(
1162 project_ids
1163 .get(&row.workspace_id)
1164 .is_some_and(|projects| projects.contains(&row.project_id)),
1165 "error invalid-export-snapshot recurrence_series.project_id={} is missing in workspace {}",
1166 row.project_id,
1167 row.workspace_id
1168 );
1169 TaskPriority::parse(&row.priority).context("invalid recurrence series priority")?;
1170 let initial_status =
1171 TaskStatus::parse(&row.initial_status).context("invalid recurrence initial status")?;
1172 ensure!(
1173 initial_status.is_open(),
1174 "error invalid-export-snapshot recurrence initial status must be open"
1175 );
1176 let frequency = RecurrenceFrequency::parse(&row.frequency)?;
1177 let interval = u32::try_from(row.interval).context("invalid recurrence interval")?;
1178 let weekdays = row
1179 .weekdays
1180 .parse::<WeekdaySet>()
1181 .map_err(anyhow::Error::msg)?;
1182 let rule = RecurrenceRule::new(frequency, interval, weekdays)?;
1183 let timezone = row.timezone.parse::<TimeZoneId>()?;
1184 let start_on = row
1185 .start_on
1186 .parse::<NaiveDate>()
1187 .context("invalid recurrence start date")?;
1188 let available_local_time = optional_import_text(&row.available_local_time)
1189 .map(|value| value.parse::<NaiveTime>())
1190 .transpose()
1191 .context("invalid recurrence availability time")?;
1192 let due_policy = RecurrenceDuePolicy::parse(&row.due_policy)?;
1193 let state = RecurrenceSeriesState::parse(&row.state)?;
1194 let stopped_at = optional_import_text(&row.stopped_at)
1195 .map(DateTime::parse_from_rfc3339)
1196 .transpose()
1197 .context("invalid recurrence stop time")?;
1198 ensure!(
1199 matches!(state, RecurrenceSeriesState::Stopped) == stopped_at.is_some(),
1200 "error invalid-export-snapshot recurrence stopped state and stop time disagree"
1201 );
1202 let created_at = DateTime::parse_from_rfc3339(&row.created_at)
1203 .context("invalid recurrence creation time")?;
1204 DateTime::parse_from_rfc3339(&row.updated_at).context("invalid recurrence update time")?;
1205 ensure!(
1206 matches!(row.deleted, 0 | 1),
1207 "error invalid-export-snapshot recurrence deleted value must be zero or one"
1208 );
1209 let key = (row.workspace_id.clone(), row.id.clone());
1210 ensure!(
1211 !series_by_id.contains_key(&key),
1212 "error invalid-export-snapshot recurrence series identity is duplicated"
1213 );
1214 series_by_id.insert(
1215 key,
1216 ValidatedSeries {
1217 row,
1218 schedule: RecurrenceSchedule::new(
1219 rule,
1220 timezone,
1221 start_on,
1222 available_local_time,
1223 due_policy,
1224 ),
1225 state,
1226 created_at,
1227 stopped_at,
1228 },
1229 );
1230 }
1231
1232 let mut series_label_keys = HashSet::new();
1233 for row in &export.tables.recurrence_series_labels {
1234 let series_key = (row.workspace_id.clone(), row.series_id.clone());
1235 ensure!(
1236 series_by_id.contains_key(&series_key),
1237 "error invalid-export-snapshot recurrence series label has no series"
1238 );
1239 ensure!(
1240 label_keys.contains(&(row.workspace_id.clone(), row.label.clone())),
1241 "error invalid-export-snapshot recurrence series label={} is missing",
1242 row.label
1243 );
1244 ensure!(
1245 series_label_keys.insert((row.workspace_id.clone(), row.series_id.clone(), &row.label)),
1246 "error invalid-export-snapshot recurrence series label is duplicated"
1247 );
1248 }
1249
1250 let mut occurrence_keys = HashSet::new();
1251 let mut occurrence_tasks = HashSet::new();
1252 let mut projected_series = HashSet::new();
1253 for row in &export.tables.recurrence_occurrences {
1254 let series_key = (row.workspace_id.clone(), row.series_id.clone());
1255 let series = series_by_id
1256 .get(&series_key)
1257 .context("error invalid-export-snapshot recurrence occurrence has no series")?;
1258 let slot_on = row
1259 .slot_on
1260 .parse::<NaiveDate>()
1261 .context("invalid recurrence slot date")?;
1262 ensure!(
1263 occurrence_keys.insert((row.workspace_id.clone(), row.series_id.clone(), slot_on)),
1264 "error invalid-export-snapshot recurrence occurrence identity is duplicated"
1265 );
1266 ensure!(
1267 is_slot(&series.schedule.rule, series.schedule.start_on, slot_on),
1268 "error invalid-export-snapshot recurrence slot={} is outside the series lattice",
1269 slot_on
1270 );
1271 let creation_date = series
1272 .created_at
1273 .with_timezone(&series.schedule.timezone.timezone())
1274 .date_naive();
1275 ensure!(
1276 slot_on >= creation_date,
1277 "error invalid-export-snapshot recurrence slot={} precedes the series lifecycle",
1278 slot_on
1279 );
1280 let projection_state = RecurrenceProjectionState::parse(&row.projection_state)?;
1281 let outcome = optional_import_text(&row.outcome)
1282 .map(RecurrenceOutcome::parse)
1283 .transpose()?;
1284 let task_id = optional_import_text(&row.task_id)
1285 .map(str::parse::<TaskId>)
1286 .transpose()?;
1287 let resolved_at = optional_import_text(&row.resolved_at);
1288 let outcome_change_id = optional_import_text(&row.outcome_change_id);
1289 let archived_at = optional_import_text(&row.archived_at);
1290 let valid_shape = match projection_state {
1291 RecurrenceProjectionState::Projected => {
1292 task_id.is_some()
1293 && outcome.is_none()
1294 && resolved_at.is_none()
1295 && outcome_change_id.is_none()
1296 && archived_at.is_none()
1297 }
1298 RecurrenceProjectionState::Resolved => {
1299 task_id.is_some()
1300 && outcome.is_some()
1301 && resolved_at.is_some()
1302 && outcome_change_id.is_some()
1303 && archived_at.is_none()
1304 }
1305 RecurrenceProjectionState::Archived => {
1306 task_id.is_some()
1307 && outcome.is_none()
1308 && resolved_at.is_none()
1309 && outcome_change_id.is_none()
1310 && archived_at.is_some()
1311 }
1312 };
1313 ensure!(
1314 valid_shape,
1315 "error invalid-export-snapshot recurrence occurrence state and fields disagree"
1316 );
1317 if matches!(projection_state, RecurrenceProjectionState::Projected) {
1318 ensure!(
1319 projected_series.insert(series_key.clone()),
1320 "error invalid-export-snapshot recurrence projection is not unique"
1321 );
1322 }
1323 for value in [resolved_at, archived_at].into_iter().flatten() {
1324 let activity_at = DateTime::parse_from_rfc3339(value)
1325 .context("invalid recurrence occurrence timestamp")?;
1326 if let Some(stopped_at) = series.stopped_at {
1327 ensure!(
1328 activity_at <= stopped_at,
1329 "error invalid-export-snapshot recurrence activity exceeds the stop boundary"
1330 );
1331 }
1332 }
1333
1334 if let Some(task_id) = task_id {
1335 ensure!(
1336 task_ids
1337 .get(&row.workspace_id)
1338 .is_some_and(|tasks| tasks.contains(&task_id)),
1339 "error invalid-export-snapshot recurrence task={} is missing",
1340 task_id
1341 );
1342 ensure!(
1343 occurrence_tasks.insert((row.workspace_id.clone(), task_id.clone())),
1344 "error invalid-export-snapshot recurrence task link is duplicated"
1345 );
1346 let identity = derive_occurrence_identity(
1347 &row.workspace_id,
1348 &row.series_id,
1349 &series.schedule,
1350 slot_on,
1351 )?;
1352 ensure!(
1353 identity.task_id == task_id,
1354 "error invalid-export-snapshot recurrence deterministic task identity mismatch slot={slot_on}"
1355 );
1356 let task = task_rows
1357 .get(&(row.workspace_id.clone(), task_id.clone()))
1358 .context("error invalid-export-snapshot recurrence task row is missing")?;
1359 ensure!(
1360 task.created_at == identity.created_at,
1361 "error invalid-export-snapshot recurrence deterministic task timestamp mismatch slot={slot_on}"
1362 );
1363 match outcome {
1364 Some(RecurrenceOutcome::Completed) => ensure!(
1365 task.status == "done",
1366 "error invalid-export-snapshot recurrence completed outcome requires done task"
1367 ),
1368 Some(RecurrenceOutcome::Skipped) => ensure!(
1369 task.status == "canceled",
1370 "error invalid-export-snapshot recurrence skipped outcome requires canceled task"
1371 ),
1372 None if matches!(projection_state, RecurrenceProjectionState::Projected) => {
1373 ensure!(
1374 TaskStatus::parse(&task.status)?.is_open(),
1375 "error invalid-export-snapshot recurrence projected task must be open"
1376 );
1377 }
1378 None => {}
1379 }
1380 validate_deterministic_change(
1381 change_rows.get(identity.task_change_id.as_str()).copied(),
1382 &identity,
1383 slot_on,
1384 false,
1385 )?;
1386 validate_deterministic_change(
1387 change_rows
1388 .get(identity.occurrence_change_id.as_str())
1389 .copied(),
1390 &identity,
1391 slot_on,
1392 true,
1393 )?;
1394 for field in TaskField::VERSIONED {
1395 let version = field_versions
1396 .get(&(
1397 row.workspace_id.clone(),
1398 "task",
1399 task_id.as_str(),
1400 field.as_str(),
1401 ))
1402 .context(
1403 "error invalid-export-snapshot recurrence task field version is missing",
1404 )?;
1405 if *version == identity.field_version_seeds.task {
1406 continue;
1407 }
1408 ensure!(
1409 change_rows.contains_key(version),
1410 "error invalid-export-snapshot recurrence task field version has no change"
1411 );
1412 }
1413 }
1414
1415 if let Some(change_id) = outcome_change_id {
1416 let change = change_rows
1417 .get(change_id)
1418 .context("error invalid-export-snapshot recurrence outcome change is missing")?;
1419 ensure!(
1420 change.entity_type == "recurrence_series"
1421 && change.entity_id == row.series_id.as_str()
1422 && change.field.as_deref() == Some("outcome")
1423 && change.op_type == "resolve_recurrence_occurrence",
1424 "error invalid-export-snapshot recurrence outcome change identity mismatch"
1425 );
1426 let payload: Value = serde_json::from_str(&change.payload)
1427 .context("invalid recurrence outcome change payload")?;
1428 ensure!(
1429 payload.get("slot_on").and_then(Value::as_str) == Some(row.slot_on.as_str())
1430 && payload.get("outcome").and_then(Value::as_str)
1431 == outcome.map(RecurrenceOutcome::as_str)
1432 && payload.get("resolved_at").and_then(Value::as_str) == resolved_at,
1433 "error invalid-export-snapshot recurrence outcome change payload mismatch"
1434 );
1435 }
1436 }
1437
1438 let mut pause_ids = HashSet::new();
1439 let mut pauses_by_series: HashMap<_, Vec<_>> = HashMap::new();
1440 for row in &export.tables.recurrence_pause_intervals {
1441 let series_key = (row.workspace_id.clone(), row.series_id.clone());
1442 let series = series_by_id
1443 .get(&series_key)
1444 .context("error invalid-export-snapshot recurrence pause has no series")?;
1445 ensure!(
1446 pause_ids.insert((row.workspace_id.clone(), row.id.as_str())),
1447 "error invalid-export-snapshot recurrence pause identity is duplicated"
1448 );
1449 let paused_at = DateTime::parse_from_rfc3339(&row.paused_at)
1450 .context("invalid recurrence pause time")?;
1451 let resumed_at = optional_import_text(&row.resumed_at)
1452 .map(DateTime::parse_from_rfc3339)
1453 .transpose()
1454 .context("invalid recurrence resume time")?;
1455 ensure!(
1456 resumed_at.is_none_or(|resumed| resumed > paused_at),
1457 "error invalid-export-snapshot recurrence pause interval is inverted"
1458 );
1459 ensure!(
1460 paused_at >= series.created_at,
1461 "error invalid-export-snapshot recurrence pause precedes the series lifecycle"
1462 );
1463 if let Some(stopped_at) = series.stopped_at {
1464 ensure!(
1465 paused_at <= stopped_at && resumed_at.is_none_or(|resumed| resumed <= stopped_at),
1466 "error invalid-export-snapshot recurrence pause exceeds the stop boundary"
1467 );
1468 }
1469 ensure!(
1470 resumed_at.is_some() != row.resolved_by_change_id.is_empty(),
1471 "error invalid-export-snapshot recurrence pause resolution fields disagree"
1472 );
1473 ensure!(
1474 row.suspended_slot_on.is_empty() == row.suspended_task_id.is_empty(),
1475 "error invalid-export-snapshot recurrence suspended task fields disagree"
1476 );
1477 if !row.suspended_slot_on.is_empty() {
1478 let slot_on = row.suspended_slot_on.parse::<NaiveDate>()?;
1479 let task_id = row.suspended_task_id.parse::<TaskId>()?;
1480 ensure!(
1481 occurrence_keys.contains(&(
1482 row.workspace_id.clone(),
1483 row.series_id.clone(),
1484 slot_on
1485 )) && occurrence_tasks.contains(&(row.workspace_id.clone(), task_id)),
1486 "error invalid-export-snapshot recurrence suspended task link is missing"
1487 );
1488 }
1489 for change_id in [
1490 Some(row.created_by_change_id.as_str()),
1491 optional_import_text(&row.resolved_by_change_id),
1492 ]
1493 .into_iter()
1494 .flatten()
1495 {
1496 let change = change_rows
1497 .get(change_id)
1498 .context("error invalid-export-snapshot recurrence pause change is missing")?;
1499 ensure!(
1500 change.entity_type == "recurrence_series"
1501 && change.entity_id == row.series_id.as_str(),
1502 "error invalid-export-snapshot recurrence pause change identity mismatch"
1503 );
1504 }
1505 pauses_by_series
1506 .entry(series_key)
1507 .or_default()
1508 .push((paused_at, resumed_at));
1509 }
1510 for pauses in pauses_by_series.values_mut() {
1511 pauses.sort_by_key(|(paused_at, _)| *paused_at);
1512 for pair in pauses.windows(2) {
1513 ensure!(
1514 pair[0].1.is_some_and(|resumed_at| resumed_at <= pair[1].0),
1515 "error invalid-export-snapshot recurrence pause intervals overlap"
1516 );
1517 }
1518 }
1519
1520 for series in series_by_id.values() {
1521 let has_lifecycle_conflict = export.tables.conflicts.iter().any(|conflict| {
1522 conflict.workspace_id == series.row.workspace_id
1523 && conflict.entity_type == "recurrence_series"
1524 && conflict.entity_id == series.row.id.as_str()
1525 && conflict.field == "state"
1526 && conflict.resolved == 0
1527 });
1528 let has_open_pause = pauses_by_series
1529 .get(&(series.row.workspace_id.clone(), series.row.id.clone()))
1530 .is_some_and(|pauses| pauses.iter().any(|(_, resumed_at)| resumed_at.is_none()));
1531 if !has_lifecycle_conflict {
1532 ensure!(
1533 matches!(series.state, RecurrenceSeriesState::Paused) == has_open_pause,
1534 "error invalid-export-snapshot recurrence state and open pause disagree"
1535 );
1536 }
1537 }
1538
1539 Ok(())
1540}
1541
1542fn validate_deterministic_change(
1543 change: Option<&ChangeRow>,
1544 identity: &crate::recurrence::RecurrenceOccurrenceIdentity,
1545 slot_on: NaiveDate,
1546 projection: bool,
1547) -> Result<()> {
1548 let change = change
1549 .context("error invalid-export-snapshot recurrence deterministic change is missing")?;
1550 let (entity_type, entity_id, field, op_type, created_at) = if projection {
1551 (
1552 "recurrence_series",
1553 identity.occurrence_link.series_id.as_str(),
1554 Some("projection"),
1555 "project_recurrence_occurrence",
1556 identity.occurrence_link.projected_at.as_str(),
1557 )
1558 } else {
1559 (
1560 "task",
1561 identity.task_id.as_str(),
1562 None,
1563 "create_task",
1564 identity.created_at.as_str(),
1565 )
1566 };
1567 ensure!(
1568 change.entity_type == entity_type
1569 && change.entity_id == entity_id
1570 && change.field.as_deref() == field
1571 && change.op_type == op_type
1572 && change.created_at == created_at,
1573 "error invalid-export-snapshot recurrence deterministic change identity mismatch"
1574 );
1575 let payload: Value = serde_json::from_str(&change.payload)
1576 .context("invalid recurrence deterministic change payload")?;
1577 for (key, expected) in [
1578 ("task_id", identity.task_id.as_str()),
1579 ("series_id", identity.occurrence_link.series_id.as_str()),
1580 ("task_change_id", identity.task_change_id.as_str()),
1581 (
1582 "occurrence_change_id",
1583 identity.occurrence_change_id.as_str(),
1584 ),
1585 (
1586 "task_field_version_seed",
1587 identity.field_version_seeds.task.as_str(),
1588 ),
1589 (
1590 "occurrence_field_version_seed",
1591 identity.field_version_seeds.occurrence.as_str(),
1592 ),
1593 ] {
1594 ensure!(
1595 payload.get(key).and_then(Value::as_str) == Some(expected),
1596 "error invalid-export-snapshot recurrence deterministic payload field={key} mismatch"
1597 );
1598 }
1599 ensure!(
1600 payload.get("slot_on").and_then(Value::as_str) == Some(&slot_on.to_string()),
1601 "error invalid-export-snapshot recurrence deterministic payload slot mismatch"
1602 );
1603 if projection {
1604 ensure!(
1605 payload.get("projected_at").and_then(Value::as_str)
1606 == Some(identity.occurrence_link.projected_at.as_str()),
1607 "error invalid-export-snapshot recurrence deterministic projection link mismatch"
1608 );
1609 }
1610 Ok(())
1611}
1612
1613fn optional_import_text(value: &str) -> Option<&str> {
1614 (!value.is_empty()).then_some(value)
1615}
1616
1617async fn replace_from_export(
1618 tx: &mut SqliteConnection,
1619 export: &AvenExport,
1620 target_client_id: &str,
1621) -> Result<()> {
1622 let recurrence_data = has_recurrence_data(export);
1623 let delete_order = [
1624 "DELETE FROM recurrence_pause_intervals",
1625 "DELETE FROM recurrence_occurrences",
1626 "DELETE FROM recurrence_series_metadata",
1627 "DELETE FROM recurrence_series_labels",
1628 "DELETE FROM recurrence_series",
1629 "DELETE FROM task_metadata",
1630 "DELETE FROM task_attachments",
1631 "DELETE FROM blob_inventory",
1632 "DELETE FROM task_related_links",
1633 "DELETE FROM task_epic_links",
1634 "DELETE FROM task_dependencies",
1635 "DELETE FROM task_labels",
1636 "DELETE FROM notes",
1637 "DELETE FROM conflicts",
1638 "DELETE FROM field_versions",
1639 "DELETE FROM changes",
1640 "DELETE FROM project_paths",
1641 "DELETE FROM project_id_aliases",
1642 "DELETE FROM metadata_field_id_aliases",
1643 "DELETE FROM tasks",
1644 "DELETE FROM metadata_fields",
1645 "DELETE FROM labels",
1646 "DELETE FROM projects",
1647 "DELETE FROM workspaces",
1648 "DELETE FROM meta",
1649 ];
1650 for sql in delete_order {
1651 sqlx::query(sql).execute(&mut *tx).await?;
1652 }
1653
1654 db::set_meta(tx, "client_id", target_client_id).await?;
1655 db::set_meta(tx, "sync_cursor", "0").await?;
1656 let local_seq = export
1657 .tables
1658 .changes
1659 .iter()
1660 .map(|row| row.local_seq)
1661 .max()
1662 .unwrap_or(0);
1663 db::set_meta(tx, "local_seq", &local_seq.to_string()).await?;
1664
1665 for meta in &export.tables.meta {
1666 if matches!(
1667 meta.key.as_str(),
1668 "client_id" | "sync_server_url" | "sync_cursor" | "local_seq"
1669 ) {
1670 continue;
1671 }
1672 db::set_meta(tx, &meta.key, &meta.value).await?;
1673 }
1674
1675 let suppressed_attachment_changes = export
1676 .tables
1677 .changes
1678 .iter()
1679 .filter(|change| {
1680 change.server_seq.is_none()
1681 && change.field.as_deref() == Some("attachments")
1682 && matches!(
1683 change.op_type.as_str(),
1684 "attachment_add" | "attachment_delete"
1685 )
1686 })
1687 .map(|change| change.change_id.as_str())
1688 .collect::<HashSet<_>>();
1689 let mut attachments = export.tables.task_attachments.clone();
1690 for attachment in &mut attachments {
1691 if attachment
1692 .created_by_change_id
1693 .as_deref()
1694 .is_some_and(|id| suppressed_attachment_changes.contains(id))
1695 {
1696 attachment.created_by_change_id = None;
1697 }
1698 if attachment
1699 .deleted_by_change_id
1700 .as_deref()
1701 .is_some_and(|id| suppressed_attachment_changes.contains(id))
1702 {
1703 attachment.deleted_by_change_id = None;
1704 }
1705 }
1706 let changes = export
1707 .tables
1708 .changes
1709 .iter()
1710 .filter(|change| !suppressed_attachment_changes.contains(change.change_id.as_str()))
1711 .filter(|change| {
1712 recurrence_data
1713 || (change.entity_type != "recurrence_series"
1714 && !change.op_type.contains("recurrence"))
1715 })
1716 .cloned()
1717 .collect::<Vec<_>>();
1718 let field_versions = export
1719 .tables
1720 .field_versions
1721 .iter()
1722 .filter(|row| recurrence_data || row.entity_type != "recurrence_series")
1723 .cloned()
1724 .collect::<Vec<_>>();
1725 let conflicts = export
1726 .tables
1727 .conflicts
1728 .iter()
1729 .filter(|row| recurrence_data || row.entity_type != "recurrence_series")
1730 .cloned()
1731 .collect::<Vec<_>>();
1732
1733 tables::import_workspaces(tx, &export.tables.workspaces).await?;
1734 tables::import_projects(tx, &export.tables.projects).await?;
1735 tables::import_project_id_aliases(tx, &export.tables.project_id_aliases).await?;
1736 tables::import_project_paths(tx, &export.tables.project_paths).await?;
1737 tables::import_labels(tx, &export.tables.labels).await?;
1738 tables::import_metadata_fields(tx, &export.tables.metadata_fields).await?;
1739 tables::import_metadata_field_id_aliases(tx, &export.tables.metadata_field_id_aliases).await?;
1740 tables::import_tasks(tx, &export.tables.tasks).await?;
1741 tables::import_task_metadata(tx, &export.tables.task_metadata).await?;
1742 tables::import_task_labels(tx, &export.tables.task_labels).await?;
1743 tables::import_notes(tx, &export.tables.notes).await?;
1744 tables::import_task_dependencies(tx, &export.tables.task_dependencies).await?;
1745 tables::import_task_epic_links(tx, &export.tables.task_epic_links).await?;
1746 tables::import_blob_inventory(tx, &export.tables.blob_inventory).await?;
1747 tables::import_task_attachments(tx, &attachments).await?;
1748 if recurrence_data {
1749 tables::import_recurrence_series(tx, &export.tables.recurrence_series).await?;
1750 tables::import_recurrence_series_labels(tx, &export.tables.recurrence_series_labels)
1751 .await?;
1752 tables::import_recurrence_series_metadata(tx, &export.tables.recurrence_series_metadata)
1753 .await?;
1754 tables::import_recurrence_occurrences(tx, &export.tables.recurrence_occurrences).await?;
1755 tables::import_recurrence_pause_intervals(tx, &export.tables.recurrence_pause_intervals)
1756 .await?;
1757 }
1758 tables::import_changes(tx, &changes).await?;
1759 tables::import_task_related_links(tx, &export.tables.task_related_links).await?;
1760 tables::import_field_versions(tx, &field_versions).await?;
1761 tables::import_conflicts(tx, &conflicts).await?;
1762
1763 Ok(())
1764}
1765
1766async fn database_integrity_report_with_connection(
1767 conn: &mut SqliteConnection,
1768) -> Result<IntegrityReport> {
1769 let quick_check_value: String = query_scalar("PRAGMA quick_check")
1770 .fetch_one(&mut *conn)
1771 .await?;
1772 let mut checks = Vec::new();
1773 checks.push(count_check(
1774 conn,
1775 "task projects",
1776 "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",
1777 )
1778 .await?);
1779 checks.push(count_check(
1780 conn,
1781 "project paths",
1782 "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",
1783 )
1784 .await?);
1785 checks.push(count_check(
1786 conn,
1787 "project aliases",
1788 "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",
1789 )
1790 .await?);
1791 checks.push(count_check(
1792 conn,
1793 "metadata field workspaces",
1794 "SELECT count(*) FROM metadata_fields f LEFT JOIN workspaces w ON w.id = f.workspace_id WHERE w.id IS NULL",
1795 )
1796 .await?);
1797 checks.push(count_check(
1798 conn,
1799 "metadata aliases",
1800 "SELECT count(*) FROM metadata_field_id_aliases a LEFT JOIN metadata_fields f ON f.workspace_id = a.workspace_id AND f.id = a.local_field_id WHERE f.id IS NULL",
1801 )
1802 .await?);
1803 checks.push(count_check(
1804 conn,
1805 "task metadata tasks",
1806 "SELECT count(*) FROM task_metadata m LEFT JOIN tasks t ON t.workspace_id = m.workspace_id AND t.id = m.task_id WHERE t.id IS NULL",
1807 )
1808 .await?);
1809 checks.push(count_check(
1810 conn,
1811 "task metadata fields",
1812 "SELECT count(*) FROM task_metadata m LEFT JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id WHERE f.id IS NULL",
1813 )
1814 .await?);
1815 checks.push(
1816 count_check(
1817 conn,
1818 "task metadata value size",
1819 "SELECT count(*) FROM task_metadata WHERE length(CAST(value AS BLOB)) > 4096",
1820 )
1821 .await?,
1822 );
1823 checks.push(count_check(
1824 conn,
1825 "task metadata aggregate limits",
1826 "SELECT count(*) FROM (SELECT workspace_id, task_id FROM task_metadata GROUP BY workspace_id, task_id HAVING count(*) > 128 OR sum(length(CAST(value AS BLOB))) > 32768)",
1827 )
1828 .await?);
1829 checks.push(count_check(
1830 conn,
1831 "recurrence metadata series",
1832 "SELECT count(*) FROM recurrence_series_metadata m LEFT JOIN recurrence_series s ON s.workspace_id = m.workspace_id AND s.id = m.series_id WHERE s.id IS NULL",
1833 )
1834 .await?);
1835 checks.push(count_check(
1836 conn,
1837 "recurrence metadata fields",
1838 "SELECT count(*) FROM recurrence_series_metadata m LEFT JOIN metadata_fields f ON f.workspace_id = m.workspace_id AND f.id = m.field_id WHERE f.id IS NULL",
1839 )
1840 .await?);
1841 checks.push(count_check(
1842 conn,
1843 "recurrence metadata value size",
1844 "SELECT count(*) FROM recurrence_series_metadata WHERE length(CAST(value AS BLOB)) > 4096",
1845 )
1846 .await?);
1847 checks.push(count_check(
1848 conn,
1849 "recurrence metadata aggregate limits",
1850 "SELECT count(*) FROM (SELECT workspace_id, series_id FROM recurrence_series_metadata GROUP BY workspace_id, series_id HAVING count(*) > 128 OR sum(length(CAST(value AS BLOB))) > 32768)",
1851 )
1852 .await?);
1853 let invalid_metadata_keys = sqlx::query_scalar::<_, String>("SELECT key FROM metadata_fields")
1854 .fetch_all(&mut *conn)
1855 .await?
1856 .into_iter()
1857 .filter(|key| match crate::metadata::normalize_metadata_key(key) {
1858 Ok(normalized) => normalized != key.as_str(),
1859 Err(_) => true,
1860 })
1861 .count();
1862 checks.push(IntegrityCheck {
1863 label: "metadata field keys",
1864 ok: invalid_metadata_keys == 0,
1865 value: invalid_metadata_keys.to_string(),
1866 });
1867 checks.push(count_check(
1868 conn,
1869 "task label tasks",
1870 "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",
1871 )
1872 .await?);
1873 checks.push(count_check(
1874 conn,
1875 "task label labels",
1876 "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",
1877 )
1878 .await?);
1879 checks.push(count_check(
1880 conn,
1881 "notes",
1882 "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",
1883 )
1884 .await?);
1885 checks.push(count_check(
1886 conn,
1887 "note changes",
1888 "SELECT count(*) FROM notes n LEFT JOIN changes c ON c.change_id = n.change_id WHERE c.change_id IS NULL",
1889 )
1890 .await?);
1891 checks.push(count_check(
1892 conn,
1893 "dependency tasks",
1894 "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",
1895 )
1896 .await?);
1897 checks.push(count_check(
1898 conn,
1899 "dependency targets",
1900 "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",
1901 )
1902 .await?);
1903 checks.push(count_check(
1904 conn,
1905 "related link first endpoints",
1906 "SELECT count(*) FROM task_related_links r LEFT JOIN tasks t ON t.workspace_id = r.workspace_id AND t.id = r.task_a_id WHERE t.id IS NULL",
1907 )
1908 .await?);
1909 checks.push(count_check(
1910 conn,
1911 "related link second endpoints",
1912 "SELECT count(*) FROM task_related_links r LEFT JOIN tasks t ON t.workspace_id = r.workspace_id AND t.id = r.task_b_id WHERE t.id IS NULL",
1913 )
1914 .await?);
1915 checks.push(count_check(
1916 conn,
1917 "related link canonical pairs",
1918 "SELECT count(*) FROM task_related_links WHERE task_a_id >= task_b_id OR linked NOT IN (0, 1)",
1919 )
1920 .await?);
1921 checks.push(
1922 count_check(
1923 conn,
1924 "related link changes",
1925 "SELECT count(*) FROM task_related_links r
1926 WHERE NOT EXISTS (
1927 SELECT 1 FROM changes c
1928 WHERE c.change_id = r.last_change_id
1929 AND c.entity_type = 'task' AND c.field = 'related'
1930 AND c.op_type = CASE r.linked WHEN 1 THEN 'related_add' ELSE 'related_remove' END
1931 AND CASE WHEN json_valid(c.payload) THEN json_extract(c.payload, '$.workspace_id') END = r.workspace_id
1932 AND CASE WHEN json_valid(c.payload) THEN json_extract(c.payload, '$.related_task_id') END IS NOT NULL
1933 AND min(c.entity_id, CASE WHEN json_valid(c.payload) THEN json_extract(c.payload, '$.related_task_id') END) = r.task_a_id
1934 AND max(c.entity_id, CASE WHEN json_valid(c.payload) THEN json_extract(c.payload, '$.related_task_id') END) = r.task_b_id
1935 )",
1936 )
1937 .await?,
1938 );
1939 checks.push(count_check(
1940 conn,
1941 "epic link children",
1942 "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",
1943 )
1944 .await?);
1945 checks.push(count_check(
1946 conn,
1947 "epic link parents",
1948 "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",
1949 )
1950 .await?);
1951 checks.push(count_check(
1952 conn,
1953 "epic link parent flags",
1954 "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",
1955 )
1956 .await?);
1957 checks.push(count_check(
1958 conn,
1959 "conflict tasks",
1960 "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",
1961 )
1962 .await?);
1963 checks.push(count_check(
1964 conn,
1965 "task due dates",
1966 "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)",
1967 )
1968 .await?);
1969 checks.push(count_check(
1970 conn,
1971 "field version tasks",
1972 "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",
1973 )
1974 .await?);
1975 checks.push(count_check(
1976 conn,
1977 "field version changes",
1978 "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))",
1979 )
1980 .await?);
1981 checks.extend(integrity::recurrence_integrity_checks(conn).await?);
1982 push_meta_checks(conn, &mut checks).await?;
1983
1984 Ok(IntegrityReport {
1985 quick_check_ok: quick_check_value == "ok",
1986 quick_check_value,
1987 checks,
1988 })
1989}
1990
1991pub(crate) fn ensure_integrity_ok(report: &IntegrityReport) -> Result<()> {
1992 let mut bad = vec![];
1993 if !report.quick_check_ok {
1994 bad.push("quick check");
1995 }
1996 for check in &report.checks {
1997 if !check.ok && check.label != "recurrence projection gaps" {
1998 bad.push(check.label);
1999 }
2000 }
2001 if bad.is_empty() {
2002 return Ok(());
2003 }
2004 bail!("error data-integrity-failed checks={}", bad.join(", "))
2005}
2006
2007async fn count_check(
2008 conn: &mut SqliteConnection,
2009 label: &'static str,
2010 query: &'static str,
2011) -> Result<IntegrityCheck> {
2012 let count: i64 = query_scalar(query).fetch_one(&mut *conn).await?;
2013 Ok(IntegrityCheck {
2014 label,
2015 ok: count == 0,
2016 value: format!("{count} orphaned"),
2017 })
2018}
2019
2020async fn push_meta_checks(
2021 conn: &mut SqliteConnection,
2022 checks: &mut Vec<IntegrityCheck>,
2023) -> Result<()> {
2024 let local_seq = db::get_meta(conn, "local_seq").await?;
2025 let local_seq_check = match local_seq {
2026 Some(raw) => match raw.parse::<i64>() {
2027 Ok(value) => {
2028 let max_seq: i64 = query_scalar("SELECT COALESCE(MAX(local_seq), 0) FROM changes")
2029 .fetch_one(&mut *conn)
2030 .await?;
2031 let ok = value >= max_seq;
2032 IntegrityCheck {
2033 label: "meta local_seq",
2034 ok,
2035 value: value.to_string(),
2036 }
2037 }
2038 Err(error) => IntegrityCheck {
2039 label: "meta local_seq",
2040 ok: false,
2041 value: error.to_string(),
2042 },
2043 },
2044 None => IntegrityCheck {
2045 label: "meta local_seq",
2046 ok: false,
2047 value: "missing".to_string(),
2048 },
2049 };
2050 checks.push(local_seq_check);
2051
2052 let sync_cursor = db::get_meta(conn, "sync_cursor").await?;
2053 let sync_cursor_ok = match sync_cursor {
2054 Some(raw) => match raw.parse::<i64>() {
2055 Ok(_) => IntegrityCheck {
2056 label: "sync cursor",
2057 ok: true,
2058 value: raw,
2059 },
2060 Err(error) => IntegrityCheck {
2061 label: "sync cursor",
2062 ok: false,
2063 value: error.to_string(),
2064 },
2065 },
2066 None => IntegrityCheck {
2067 label: "sync cursor",
2068 ok: false,
2069 value: "missing".to_string(),
2070 },
2071 };
2072 checks.push(sync_cursor_ok);
2073
2074 Ok(())
2075}