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, slot_values,
7};
8use crate::task_fields::TaskField;
9use anyhow::{Context, Result, bail, ensure};
10use chrono::{DateTime, NaiveDate, NaiveTime};
11use serde::{Deserialize, Serialize};
12use serde_json::Value;
13use sqlx::{SqliteConnection, query_scalar};
14use std::collections::{HashMap, HashSet};
15use std::path::{Path, PathBuf};
16
17mod archive;
18mod integrity;
19mod tables;
20use crate::db::{self, Database};
21
22const EXPORT_FORMAT: &str = "aven-export";
23const EXPORT_VERSION: i64 = 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 slot = slot_values(&series.schedule, slot_on)?;
1188 let boundary = DateTime::parse_from_rfc3339(&slot.boundary_at)?;
1189 let creation_date = series
1190 .created_at
1191 .with_timezone(&series.schedule.timezone.timezone())
1192 .date_naive();
1193 ensure!(
1194 slot_on >= creation_date,
1195 "error invalid-export-snapshot recurrence slot={} precedes the series lifecycle",
1196 slot_on
1197 );
1198 if let Some(stopped_at) = series.stopped_at {
1199 ensure!(
1200 boundary <= stopped_at,
1201 "error invalid-export-snapshot recurrence slot={} exceeds the stop boundary",
1202 slot_on
1203 );
1204 }
1205
1206 let projection_state = RecurrenceProjectionState::parse(&row.projection_state)?;
1207 let outcome = optional_import_text(&row.outcome)
1208 .map(RecurrenceOutcome::parse)
1209 .transpose()?;
1210 let task_id = optional_import_text(&row.task_id)
1211 .map(str::parse::<TaskId>)
1212 .transpose()?;
1213 let resolved_at = optional_import_text(&row.resolved_at);
1214 let outcome_change_id = optional_import_text(&row.outcome_change_id);
1215 let archived_at = optional_import_text(&row.archived_at);
1216 let valid_shape = match projection_state {
1217 RecurrenceProjectionState::Projected => {
1218 task_id.is_some()
1219 && outcome.is_none()
1220 && resolved_at.is_none()
1221 && outcome_change_id.is_none()
1222 && archived_at.is_none()
1223 }
1224 RecurrenceProjectionState::Resolved => {
1225 task_id.is_some()
1226 && outcome.is_some()
1227 && resolved_at.is_some()
1228 && outcome_change_id.is_some()
1229 && archived_at.is_none()
1230 }
1231 RecurrenceProjectionState::Archived => {
1232 task_id.is_some()
1233 && outcome.is_none()
1234 && resolved_at.is_none()
1235 && outcome_change_id.is_none()
1236 && archived_at.is_some()
1237 }
1238 };
1239 ensure!(
1240 valid_shape,
1241 "error invalid-export-snapshot recurrence occurrence state and fields disagree"
1242 );
1243 if matches!(projection_state, RecurrenceProjectionState::Projected) {
1244 ensure!(
1245 projected_series.insert(series_key.clone()),
1246 "error invalid-export-snapshot recurrence projection is not unique"
1247 );
1248 }
1249 for value in [resolved_at, archived_at].into_iter().flatten() {
1250 DateTime::parse_from_rfc3339(value)
1251 .context("invalid recurrence occurrence timestamp")?;
1252 }
1253
1254 if let Some(task_id) = task_id {
1255 ensure!(
1256 task_ids
1257 .get(&row.workspace_id)
1258 .is_some_and(|tasks| tasks.contains(&task_id)),
1259 "error invalid-export-snapshot recurrence task={} is missing",
1260 task_id
1261 );
1262 ensure!(
1263 occurrence_tasks.insert((row.workspace_id.clone(), task_id.clone())),
1264 "error invalid-export-snapshot recurrence task link is duplicated"
1265 );
1266 let identity = derive_occurrence_identity(
1267 &row.workspace_id,
1268 &row.series_id,
1269 &series.schedule,
1270 slot_on,
1271 )?;
1272 ensure!(
1273 identity.task_id == task_id,
1274 "error invalid-export-snapshot recurrence deterministic task identity mismatch slot={slot_on}"
1275 );
1276 let task = task_rows
1277 .get(&(row.workspace_id.clone(), task_id.clone()))
1278 .context("error invalid-export-snapshot recurrence task row is missing")?;
1279 ensure!(
1280 task.created_at == identity.created_at,
1281 "error invalid-export-snapshot recurrence deterministic task timestamp mismatch slot={slot_on}"
1282 );
1283 match outcome {
1284 Some(RecurrenceOutcome::Completed) => ensure!(
1285 task.status == "done",
1286 "error invalid-export-snapshot recurrence completed outcome requires done task"
1287 ),
1288 Some(RecurrenceOutcome::Skipped) => ensure!(
1289 task.status == "canceled",
1290 "error invalid-export-snapshot recurrence skipped outcome requires canceled task"
1291 ),
1292 None if matches!(projection_state, RecurrenceProjectionState::Projected) => {
1293 ensure!(
1294 TaskStatus::parse(&task.status)?.is_open(),
1295 "error invalid-export-snapshot recurrence projected task must be open"
1296 );
1297 }
1298 None => {}
1299 }
1300 validate_deterministic_change(
1301 change_rows.get(identity.task_change_id.as_str()).copied(),
1302 &identity,
1303 slot_on,
1304 false,
1305 )?;
1306 validate_deterministic_change(
1307 change_rows
1308 .get(identity.occurrence_change_id.as_str())
1309 .copied(),
1310 &identity,
1311 slot_on,
1312 true,
1313 )?;
1314 for field in TaskField::VERSIONED {
1315 let version = field_versions
1316 .get(&(
1317 row.workspace_id.clone(),
1318 "task",
1319 task_id.as_str(),
1320 field.as_str(),
1321 ))
1322 .context(
1323 "error invalid-export-snapshot recurrence task field version is missing",
1324 )?;
1325 if *version == identity.field_version_seeds.task {
1326 continue;
1327 }
1328 ensure!(
1329 change_rows.contains_key(version),
1330 "error invalid-export-snapshot recurrence task field version has no change"
1331 );
1332 }
1333 }
1334
1335 if let Some(change_id) = outcome_change_id {
1336 let change = change_rows
1337 .get(change_id)
1338 .context("error invalid-export-snapshot recurrence outcome change is missing")?;
1339 ensure!(
1340 change.entity_type == "recurrence_series"
1341 && change.entity_id == row.series_id.as_str()
1342 && change.field.as_deref() == Some("outcome")
1343 && change.op_type == "resolve_recurrence_occurrence",
1344 "error invalid-export-snapshot recurrence outcome change identity mismatch"
1345 );
1346 let payload: Value = serde_json::from_str(&change.payload)
1347 .context("invalid recurrence outcome change payload")?;
1348 ensure!(
1349 payload.get("slot_on").and_then(Value::as_str) == Some(row.slot_on.as_str())
1350 && payload.get("outcome").and_then(Value::as_str)
1351 == outcome.map(RecurrenceOutcome::as_str)
1352 && payload.get("resolved_at").and_then(Value::as_str) == resolved_at,
1353 "error invalid-export-snapshot recurrence outcome change payload mismatch"
1354 );
1355 }
1356 }
1357
1358 let mut pause_ids = HashSet::new();
1359 let mut pauses_by_series: HashMap<_, Vec<_>> = HashMap::new();
1360 for row in &export.tables.recurrence_pause_intervals {
1361 let series_key = (row.workspace_id.clone(), row.series_id.clone());
1362 let series = series_by_id
1363 .get(&series_key)
1364 .context("error invalid-export-snapshot recurrence pause has no series")?;
1365 ensure!(
1366 pause_ids.insert((row.workspace_id.clone(), row.id.as_str())),
1367 "error invalid-export-snapshot recurrence pause identity is duplicated"
1368 );
1369 let paused_at = DateTime::parse_from_rfc3339(&row.paused_at)
1370 .context("invalid recurrence pause time")?;
1371 let resumed_at = optional_import_text(&row.resumed_at)
1372 .map(DateTime::parse_from_rfc3339)
1373 .transpose()
1374 .context("invalid recurrence resume time")?;
1375 ensure!(
1376 resumed_at.is_none_or(|resumed| resumed > paused_at),
1377 "error invalid-export-snapshot recurrence pause interval is inverted"
1378 );
1379 ensure!(
1380 paused_at >= series.created_at,
1381 "error invalid-export-snapshot recurrence pause precedes the series lifecycle"
1382 );
1383 if let Some(stopped_at) = series.stopped_at {
1384 ensure!(
1385 paused_at <= stopped_at && resumed_at.is_none_or(|resumed| resumed <= stopped_at),
1386 "error invalid-export-snapshot recurrence pause exceeds the stop boundary"
1387 );
1388 }
1389 ensure!(
1390 resumed_at.is_some() != row.resolved_by_change_id.is_empty(),
1391 "error invalid-export-snapshot recurrence pause resolution fields disagree"
1392 );
1393 ensure!(
1394 row.suspended_slot_on.is_empty() == row.suspended_task_id.is_empty(),
1395 "error invalid-export-snapshot recurrence suspended task fields disagree"
1396 );
1397 if !row.suspended_slot_on.is_empty() {
1398 let slot_on = row.suspended_slot_on.parse::<NaiveDate>()?;
1399 let task_id = row.suspended_task_id.parse::<TaskId>()?;
1400 ensure!(
1401 occurrence_keys.contains(&(
1402 row.workspace_id.clone(),
1403 row.series_id.clone(),
1404 slot_on
1405 )) && occurrence_tasks.contains(&(row.workspace_id.clone(), task_id)),
1406 "error invalid-export-snapshot recurrence suspended task link is missing"
1407 );
1408 }
1409 for change_id in [
1410 Some(row.created_by_change_id.as_str()),
1411 optional_import_text(&row.resolved_by_change_id),
1412 ]
1413 .into_iter()
1414 .flatten()
1415 {
1416 let change = change_rows
1417 .get(change_id)
1418 .context("error invalid-export-snapshot recurrence pause change is missing")?;
1419 ensure!(
1420 change.entity_type == "recurrence_series"
1421 && change.entity_id == row.series_id.as_str(),
1422 "error invalid-export-snapshot recurrence pause change identity mismatch"
1423 );
1424 }
1425 pauses_by_series
1426 .entry(series_key)
1427 .or_default()
1428 .push((paused_at, resumed_at));
1429 }
1430 for pauses in pauses_by_series.values_mut() {
1431 pauses.sort_by_key(|(paused_at, _)| *paused_at);
1432 for pair in pauses.windows(2) {
1433 ensure!(
1434 pair[0].1.is_some_and(|resumed_at| resumed_at <= pair[1].0),
1435 "error invalid-export-snapshot recurrence pause intervals overlap"
1436 );
1437 }
1438 }
1439
1440 for series in series_by_id.values() {
1441 let has_lifecycle_conflict = export.tables.conflicts.iter().any(|conflict| {
1442 conflict.workspace_id == series.row.workspace_id
1443 && conflict.entity_type == "recurrence_series"
1444 && conflict.entity_id == series.row.id.as_str()
1445 && conflict.field == "state"
1446 && conflict.resolved == 0
1447 });
1448 let has_open_pause = pauses_by_series
1449 .get(&(series.row.workspace_id.clone(), series.row.id.clone()))
1450 .is_some_and(|pauses| pauses.iter().any(|(_, resumed_at)| resumed_at.is_none()));
1451 if !has_lifecycle_conflict {
1452 ensure!(
1453 matches!(series.state, RecurrenceSeriesState::Paused) == has_open_pause,
1454 "error invalid-export-snapshot recurrence state and open pause disagree"
1455 );
1456 }
1457 }
1458
1459 Ok(())
1460}
1461
1462fn validate_deterministic_change(
1463 change: Option<&ChangeRow>,
1464 identity: &crate::recurrence::RecurrenceOccurrenceIdentity,
1465 slot_on: NaiveDate,
1466 projection: bool,
1467) -> Result<()> {
1468 let change = change
1469 .context("error invalid-export-snapshot recurrence deterministic change is missing")?;
1470 let (entity_type, entity_id, field, op_type, created_at) = if projection {
1471 (
1472 "recurrence_series",
1473 identity.occurrence_link.series_id.as_str(),
1474 Some("projection"),
1475 "project_recurrence_occurrence",
1476 identity.occurrence_link.projected_at.as_str(),
1477 )
1478 } else {
1479 (
1480 "task",
1481 identity.task_id.as_str(),
1482 None,
1483 "create_task",
1484 identity.created_at.as_str(),
1485 )
1486 };
1487 ensure!(
1488 change.entity_type == entity_type
1489 && change.entity_id == entity_id
1490 && change.field.as_deref() == field
1491 && change.op_type == op_type
1492 && change.created_at == created_at,
1493 "error invalid-export-snapshot recurrence deterministic change identity mismatch"
1494 );
1495 let payload: Value = serde_json::from_str(&change.payload)
1496 .context("invalid recurrence deterministic change payload")?;
1497 for (key, expected) in [
1498 ("task_id", identity.task_id.as_str()),
1499 ("series_id", identity.occurrence_link.series_id.as_str()),
1500 ("task_change_id", identity.task_change_id.as_str()),
1501 (
1502 "occurrence_change_id",
1503 identity.occurrence_change_id.as_str(),
1504 ),
1505 (
1506 "task_field_version_seed",
1507 identity.field_version_seeds.task.as_str(),
1508 ),
1509 (
1510 "occurrence_field_version_seed",
1511 identity.field_version_seeds.occurrence.as_str(),
1512 ),
1513 ] {
1514 ensure!(
1515 payload.get(key).and_then(Value::as_str) == Some(expected),
1516 "error invalid-export-snapshot recurrence deterministic payload field={key} mismatch"
1517 );
1518 }
1519 ensure!(
1520 payload.get("slot_on").and_then(Value::as_str) == Some(&slot_on.to_string()),
1521 "error invalid-export-snapshot recurrence deterministic payload slot mismatch"
1522 );
1523 if projection {
1524 ensure!(
1525 payload.get("projected_at").and_then(Value::as_str)
1526 == Some(identity.occurrence_link.projected_at.as_str()),
1527 "error invalid-export-snapshot recurrence deterministic projection link mismatch"
1528 );
1529 }
1530 Ok(())
1531}
1532
1533fn optional_import_text(value: &str) -> Option<&str> {
1534 (!value.is_empty()).then_some(value)
1535}
1536
1537async fn replace_from_export(
1538 tx: &mut SqliteConnection,
1539 export: &AvenExport,
1540 target_client_id: &str,
1541) -> Result<()> {
1542 let recurrence_data = has_recurrence_data(export);
1543 let delete_order = [
1544 "DELETE FROM recurrence_pause_intervals",
1545 "DELETE FROM recurrence_occurrences",
1546 "DELETE FROM recurrence_series_metadata",
1547 "DELETE FROM recurrence_series_labels",
1548 "DELETE FROM recurrence_series",
1549 "DELETE FROM task_metadata",
1550 "DELETE FROM task_attachments",
1551 "DELETE FROM blob_inventory",
1552 "DELETE FROM task_epic_links",
1553 "DELETE FROM task_dependencies",
1554 "DELETE FROM task_labels",
1555 "DELETE FROM notes",
1556 "DELETE FROM conflicts",
1557 "DELETE FROM field_versions",
1558 "DELETE FROM changes",
1559 "DELETE FROM project_paths",
1560 "DELETE FROM project_id_aliases",
1561 "DELETE FROM metadata_field_id_aliases",
1562 "DELETE FROM tasks",
1563 "DELETE FROM metadata_fields",
1564 "DELETE FROM labels",
1565 "DELETE FROM projects",
1566 "DELETE FROM workspaces",
1567 "DELETE FROM meta",
1568 ];
1569 for sql in delete_order {
1570 sqlx::query(sql).execute(&mut *tx).await?;
1571 }
1572
1573 db::set_meta(tx, "client_id", target_client_id).await?;
1574 db::set_meta(tx, "sync_cursor", "0").await?;
1575 let local_seq = export
1576 .tables
1577 .changes
1578 .iter()
1579 .map(|row| row.local_seq)
1580 .max()
1581 .unwrap_or(0);
1582 db::set_meta(tx, "local_seq", &local_seq.to_string()).await?;
1583
1584 for meta in &export.tables.meta {
1585 if matches!(
1586 meta.key.as_str(),
1587 "client_id" | "sync_server_url" | "sync_cursor" | "local_seq"
1588 ) {
1589 continue;
1590 }
1591 db::set_meta(tx, &meta.key, &meta.value).await?;
1592 }
1593
1594 let suppressed_attachment_changes = export
1595 .tables
1596 .changes
1597 .iter()
1598 .filter(|change| {
1599 change.server_seq.is_none()
1600 && change.field.as_deref() == Some("attachments")
1601 && matches!(
1602 change.op_type.as_str(),
1603 "attachment_add" | "attachment_delete"
1604 )
1605 })
1606 .map(|change| change.change_id.as_str())
1607 .collect::<HashSet<_>>();
1608 let mut attachments = export.tables.task_attachments.clone();
1609 for attachment in &mut attachments {
1610 if attachment
1611 .created_by_change_id
1612 .as_deref()
1613 .is_some_and(|id| suppressed_attachment_changes.contains(id))
1614 {
1615 attachment.created_by_change_id = None;
1616 }
1617 if attachment
1618 .deleted_by_change_id
1619 .as_deref()
1620 .is_some_and(|id| suppressed_attachment_changes.contains(id))
1621 {
1622 attachment.deleted_by_change_id = None;
1623 }
1624 }
1625 let changes = export
1626 .tables
1627 .changes
1628 .iter()
1629 .filter(|change| !suppressed_attachment_changes.contains(change.change_id.as_str()))
1630 .filter(|change| {
1631 recurrence_data
1632 || (change.entity_type != "recurrence_series"
1633 && !change.op_type.contains("recurrence"))
1634 })
1635 .cloned()
1636 .collect::<Vec<_>>();
1637 let field_versions = export
1638 .tables
1639 .field_versions
1640 .iter()
1641 .filter(|row| recurrence_data || row.entity_type != "recurrence_series")
1642 .cloned()
1643 .collect::<Vec<_>>();
1644 let conflicts = export
1645 .tables
1646 .conflicts
1647 .iter()
1648 .filter(|row| recurrence_data || row.entity_type != "recurrence_series")
1649 .cloned()
1650 .collect::<Vec<_>>();
1651
1652 tables::import_workspaces(tx, &export.tables.workspaces).await?;
1653 tables::import_projects(tx, &export.tables.projects).await?;
1654 tables::import_project_id_aliases(tx, &export.tables.project_id_aliases).await?;
1655 tables::import_project_paths(tx, &export.tables.project_paths).await?;
1656 tables::import_labels(tx, &export.tables.labels).await?;
1657 tables::import_metadata_fields(tx, &export.tables.metadata_fields).await?;
1658 tables::import_metadata_field_id_aliases(tx, &export.tables.metadata_field_id_aliases).await?;
1659 tables::import_tasks(tx, &export.tables.tasks).await?;
1660 tables::import_task_metadata(tx, &export.tables.task_metadata).await?;
1661 tables::import_task_labels(tx, &export.tables.task_labels).await?;
1662 tables::import_notes(tx, &export.tables.notes).await?;
1663 tables::import_task_dependencies(tx, &export.tables.task_dependencies).await?;
1664 tables::import_task_epic_links(tx, &export.tables.task_epic_links).await?;
1665 tables::import_blob_inventory(tx, &export.tables.blob_inventory).await?;
1666 tables::import_task_attachments(tx, &attachments).await?;
1667 if recurrence_data {
1668 tables::import_recurrence_series(tx, &export.tables.recurrence_series).await?;
1669 tables::import_recurrence_series_labels(tx, &export.tables.recurrence_series_labels)
1670 .await?;
1671 tables::import_recurrence_series_metadata(tx, &export.tables.recurrence_series_metadata)
1672 .await?;
1673 tables::import_recurrence_occurrences(tx, &export.tables.recurrence_occurrences).await?;
1674 tables::import_recurrence_pause_intervals(tx, &export.tables.recurrence_pause_intervals)
1675 .await?;
1676 }
1677 tables::import_changes(tx, &changes).await?;
1678 tables::import_field_versions(tx, &field_versions).await?;
1679 tables::import_conflicts(tx, &conflicts).await?;
1680
1681 Ok(())
1682}
1683
1684async fn database_integrity_report_with_connection(
1685 conn: &mut SqliteConnection,
1686) -> Result<IntegrityReport> {
1687 let quick_check_value: String = query_scalar("PRAGMA quick_check")
1688 .fetch_one(&mut *conn)
1689 .await?;
1690 let mut checks = Vec::new();
1691 checks.push(count_check(
1692 conn,
1693 "task projects",
1694 "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",
1695 )
1696 .await?);
1697 checks.push(count_check(
1698 conn,
1699 "project paths",
1700 "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",
1701 )
1702 .await?);
1703 checks.push(count_check(
1704 conn,
1705 "project aliases",
1706 "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",
1707 )
1708 .await?);
1709 checks.push(count_check(
1710 conn,
1711 "metadata field workspaces",
1712 "SELECT count(*) FROM metadata_fields f LEFT JOIN workspaces w ON w.id = f.workspace_id WHERE w.id IS NULL",
1713 )
1714 .await?);
1715 checks.push(count_check(
1716 conn,
1717 "metadata aliases",
1718 "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",
1719 )
1720 .await?);
1721 checks.push(count_check(
1722 conn,
1723 "task metadata tasks",
1724 "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",
1725 )
1726 .await?);
1727 checks.push(count_check(
1728 conn,
1729 "task metadata fields",
1730 "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",
1731 )
1732 .await?);
1733 checks.push(
1734 count_check(
1735 conn,
1736 "task metadata value size",
1737 "SELECT count(*) FROM task_metadata WHERE length(CAST(value AS BLOB)) > 4096",
1738 )
1739 .await?,
1740 );
1741 checks.push(count_check(
1742 conn,
1743 "task metadata aggregate limits",
1744 "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)",
1745 )
1746 .await?);
1747 checks.push(count_check(
1748 conn,
1749 "recurrence metadata series",
1750 "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",
1751 )
1752 .await?);
1753 checks.push(count_check(
1754 conn,
1755 "recurrence metadata fields",
1756 "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",
1757 )
1758 .await?);
1759 checks.push(count_check(
1760 conn,
1761 "recurrence metadata value size",
1762 "SELECT count(*) FROM recurrence_series_metadata WHERE length(CAST(value AS BLOB)) > 4096",
1763 )
1764 .await?);
1765 checks.push(count_check(
1766 conn,
1767 "recurrence metadata aggregate limits",
1768 "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)",
1769 )
1770 .await?);
1771 let invalid_metadata_keys = sqlx::query_scalar::<_, String>("SELECT key FROM metadata_fields")
1772 .fetch_all(&mut *conn)
1773 .await?
1774 .into_iter()
1775 .filter(|key| match crate::metadata::normalize_metadata_key(key) {
1776 Ok(normalized) => normalized != key.as_str(),
1777 Err(_) => true,
1778 })
1779 .count();
1780 checks.push(IntegrityCheck {
1781 label: "metadata field keys",
1782 ok: invalid_metadata_keys == 0,
1783 value: invalid_metadata_keys.to_string(),
1784 });
1785 checks.push(count_check(
1786 conn,
1787 "task label tasks",
1788 "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",
1789 )
1790 .await?);
1791 checks.push(count_check(
1792 conn,
1793 "task label labels",
1794 "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",
1795 )
1796 .await?);
1797 checks.push(count_check(
1798 conn,
1799 "notes",
1800 "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",
1801 )
1802 .await?);
1803 checks.push(count_check(
1804 conn,
1805 "note changes",
1806 "SELECT count(*) FROM notes n LEFT JOIN changes c ON c.change_id = n.change_id WHERE c.change_id IS NULL",
1807 )
1808 .await?);
1809 checks.push(count_check(
1810 conn,
1811 "dependency tasks",
1812 "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",
1813 )
1814 .await?);
1815 checks.push(count_check(
1816 conn,
1817 "dependency targets",
1818 "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",
1819 )
1820 .await?);
1821 checks.push(count_check(
1822 conn,
1823 "epic link children",
1824 "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",
1825 )
1826 .await?);
1827 checks.push(count_check(
1828 conn,
1829 "epic link parents",
1830 "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",
1831 )
1832 .await?);
1833 checks.push(count_check(
1834 conn,
1835 "epic link parent flags",
1836 "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",
1837 )
1838 .await?);
1839 checks.push(count_check(
1840 conn,
1841 "conflict tasks",
1842 "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",
1843 )
1844 .await?);
1845 checks.push(count_check(
1846 conn,
1847 "task due dates",
1848 "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)",
1849 )
1850 .await?);
1851 checks.push(count_check(
1852 conn,
1853 "field version tasks",
1854 "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",
1855 )
1856 .await?);
1857 checks.push(count_check(
1858 conn,
1859 "field version changes",
1860 "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))",
1861 )
1862 .await?);
1863 checks.extend(integrity::recurrence_integrity_checks(conn).await?);
1864 push_meta_checks(conn, &mut checks).await?;
1865
1866 Ok(IntegrityReport {
1867 quick_check_ok: quick_check_value == "ok",
1868 quick_check_value,
1869 checks,
1870 })
1871}
1872
1873pub(crate) fn ensure_integrity_ok(report: &IntegrityReport) -> Result<()> {
1874 let mut bad = vec![];
1875 if !report.quick_check_ok {
1876 bad.push("quick check");
1877 }
1878 for check in &report.checks {
1879 if !check.ok && check.label != "recurrence projection gaps" {
1880 bad.push(check.label);
1881 }
1882 }
1883 if bad.is_empty() {
1884 return Ok(());
1885 }
1886 bail!("error data-integrity-failed checks={}", bad.join(", "))
1887}
1888
1889async fn count_check(
1890 conn: &mut SqliteConnection,
1891 label: &'static str,
1892 query: &'static str,
1893) -> Result<IntegrityCheck> {
1894 let count: i64 = query_scalar(query).fetch_one(&mut *conn).await?;
1895 Ok(IntegrityCheck {
1896 label,
1897 ok: count == 0,
1898 value: format!("{count} orphaned"),
1899 })
1900}
1901
1902async fn push_meta_checks(
1903 conn: &mut SqliteConnection,
1904 checks: &mut Vec<IntegrityCheck>,
1905) -> Result<()> {
1906 let local_seq = db::get_meta(conn, "local_seq").await?;
1907 let local_seq_check = match local_seq {
1908 Some(raw) => match raw.parse::<i64>() {
1909 Ok(value) => {
1910 let max_seq: i64 = query_scalar("SELECT COALESCE(MAX(local_seq), 0) FROM changes")
1911 .fetch_one(&mut *conn)
1912 .await?;
1913 let ok = value >= max_seq;
1914 IntegrityCheck {
1915 label: "meta local_seq",
1916 ok,
1917 value: value.to_string(),
1918 }
1919 }
1920 Err(error) => IntegrityCheck {
1921 label: "meta local_seq",
1922 ok: false,
1923 value: error.to_string(),
1924 },
1925 },
1926 None => IntegrityCheck {
1927 label: "meta local_seq",
1928 ok: false,
1929 value: "missing".to_string(),
1930 },
1931 };
1932 checks.push(local_seq_check);
1933
1934 let sync_cursor = db::get_meta(conn, "sync_cursor").await?;
1935 let sync_cursor_ok = match sync_cursor {
1936 Some(raw) => match raw.parse::<i64>() {
1937 Ok(_) => IntegrityCheck {
1938 label: "sync cursor",
1939 ok: true,
1940 value: raw,
1941 },
1942 Err(error) => IntegrityCheck {
1943 label: "sync cursor",
1944 ok: false,
1945 value: error.to_string(),
1946 },
1947 },
1948 None => IntegrityCheck {
1949 label: "sync cursor",
1950 ok: false,
1951 value: "missing".to_string(),
1952 },
1953 };
1954 checks.push(sync_cursor_ok);
1955
1956 Ok(())
1957}