max / goingson

4.0 KB · 119 lines History Blame Raw

1	//! Application-level data migrations that require Rust logic (not expressible in SQL).
2
3	use goingson_core::email_id::deterministic_email_id;
4	use sqlx::{Acquire, SqlitePool};
5	use tracing::{info, warn};
6
7	const MIGRATION_KEY: &str = "migration_deterministic_email_ids";
8
9	/// Rehash existing emails from random v4 IDs to deterministic v5 IDs.
10	///
11	/// Reads all emails with a message_id, computes UUID v5, then updates:
12	/// 1. The email's PK (emails.id)
13	/// 2. The FK in tasks.source_email_id
14	/// 3. FTS triggers auto-handle the PK change on UPDATE
15	///
16	/// Tracked via sync_state table — runs once, skipped on subsequent launches.
17	#[tracing::instrument(skip_all)]
18	pub async fn migrate_deterministic_email_ids(pool: &SqlitePool) -> Result<(), String> {
19	// Check if already done
20	let done: Option<(String,)> = sqlx::query_as(
21	"SELECT value FROM sync_state WHERE key = ?",
22	)
23	.bind(MIGRATION_KEY)
24	.fetch_optional(pool)
25	.await
26	.map_err(\|e\| format!("Failed to check migration state: {e}"))?;
27
28	if done.as_ref().map(\|r\| r.0.as_str()) == Some("1") {
29	return Ok(());
30	}
31
32	info!("Running deterministic email ID migration");
33
34	// Fetch all emails that have a message_id and still use v4 IDs
35	let rows: Vec<(String, String)> = sqlx::query_as(
36	"SELECT id, message_id FROM emails WHERE message_id IS NOT NULL",
37	)
38	.fetch_all(pool)
39	.await
40	.map_err(\|e\| format!("Failed to fetch emails: {e}"))?;
41
42	let mut updated = 0u64;
43
44	// Use a dedicated connection with FK enforcement off, wrapped in a transaction.
45	// `detach()` removes it from the pool: `PRAGMA foreign_keys` is per-connection,
46	// so any early-return between here and the re-enable below must NOT hand a
47	// FK-disabled connection back to the pool. A detached connection is simply
48	// closed on drop instead.
49	let mut conn = pool.acquire().await
50	.map_err(\|e\| format!("Failed to acquire connection: {e}"))?
51	.detach();
52
53	sqlx::query("PRAGMA foreign_keys = OFF")
54	.execute(&mut conn)
55	.await
56	.map_err(\|e\| format!("Failed to disable FK: {e}"))?;
57
58	let mut tx = conn.begin().await
59	.map_err(\|e\| format!("Failed to begin transaction: {e}"))?;
60
61	for (old_id, message_id) in &rows {
62	let new_id = deterministic_email_id(Some(message_id));
63	let new_id_str = new_id.to_string();
64
65	if *old_id == new_id_str {
66	continue; // Already correct
67	}
68
69	// Update the email PK
70	let result = sqlx::query("UPDATE emails SET id = ? WHERE id = ?")
71	.bind(&new_id_str)
72	.bind(old_id)
73	.execute(&mut *tx)
74	.await;
75
76	match result {
77	Ok(_) => {
78	// Update FK references in tasks
79	sqlx::query(
80	"UPDATE tasks SET source_email_id = ? WHERE source_email_id = ?",
81	)
82	.bind(&new_id_str)
83	.bind(old_id)
84	.execute(&mut *tx)
85	.await
86	.map_err(\|e\| format!("Failed to update task FK for email {old_id}: {e}"))?;
87
88	updated += 1;
89	}
90	Err(e) => {
91	// PK collision = two emails mapped to same v5 UUID (duplicate message_id).
92	// Skip — the first one wins.
93	warn!("Skipping email {old_id}: {e}");
94	}
95	}
96	}
97
98	tx.commit().await
99	.map_err(\|e\| format!("Failed to commit migration: {e}"))?;
100
101	// Re-enable FK enforcement on the same connection
102	sqlx::query("PRAGMA foreign_keys = ON")
103	.execute(&mut conn)
104	.await
105	.map_err(\|e\| format!("Failed to re-enable FK: {e}"))?;
106
107	// Mark migration complete
108	sqlx::query(
109	"INSERT OR REPLACE INTO sync_state (key, value) VALUES (?, '1')",
110	)
111	.bind(MIGRATION_KEY)
112	.execute(pool)
113	.await
114	.map_err(\|e\| format!("Failed to mark migration done: {e}"))?;
115
116	info!("Deterministic email ID migration complete: {updated} emails updated");
117	Ok(())
118	}
119