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