Skip to main content

max / makenotwork

7.6 KB · 238 lines History Blame Raw
1 //! SyncKit per-key state: upsert/read a sync key, batch app usage stats, and
2 //! prune the sync change-log past a retention horizon.
3
4 use sqlx::PgPool;
5
6 use crate::db::{SyncAppId, UserId};
7 use crate::error::Result;
8
9 // ── Sync Keys ──
10
11 /// Upsert an encrypted master key for a user within an app with optimistic
12 /// concurrency control.
13 ///
14 /// `expected_version` is checked on UPDATE: if the current `key_version`
15 /// doesn't match, the update is skipped and `false` is returned (the caller
16 /// should return 409 Conflict). On INSERT (first key), `expected_version`
17 /// must be 0.
18 ///
19 /// Returns `true` if the key was inserted or updated, `false` on version mismatch.
20 #[tracing::instrument(skip_all)]
21 pub async fn upsert_sync_key(
22 pool: &PgPool,
23 app_id: SyncAppId,
24 user_id: UserId,
25 encrypted_key: &str,
26 expected_version: i32,
27 ) -> Result<bool> {
28 let result = sqlx::query(
29 r"
30 INSERT INTO sync_keys (app_id, user_id, encrypted_key)
31 VALUES ($1, $2, $3)
32 ON CONFLICT (app_id, user_id)
33 DO UPDATE SET encrypted_key = EXCLUDED.encrypted_key,
34 key_version = sync_keys.key_version + 1,
35 updated_at = NOW()
36 WHERE sync_keys.key_version = $4
37 ",
38 )
39 .bind(app_id)
40 .bind(user_id)
41 .bind(encrypted_key)
42 .bind(expected_version)
43 .execute(pool)
44 .await?;
45
46 Ok(result.rows_affected() > 0)
47 }
48
49 /// Encryption key info returned by `get_sync_key`.
50 pub struct SyncKeyInfo {
51 pub encrypted_key: String,
52 pub key_version: i32,
53 pub key_id: i32,
54 /// If a key rotation is in progress, the new key envelope and its key_id.
55 pub pending_key: Option<(String, i32)>,
56 }
57
58 /// Get the encrypted master key, version, key_id, and any pending rotation key.
59 #[tracing::instrument(skip_all)]
60 pub async fn get_sync_key(
61 pool: &PgPool,
62 app_id: SyncAppId,
63 user_id: UserId,
64 ) -> Result<Option<SyncKeyInfo>> {
65 let row: Option<(String, i32, i32)> = sqlx::query_as(
66 "SELECT encrypted_key, key_version, key_id FROM sync_keys WHERE app_id = $1 AND user_id = $2",
67 )
68 .bind(app_id)
69 .bind(user_id)
70 .fetch_optional(pool)
71 .await?;
72
73 let Some((encrypted_key, key_version, key_id)) = row else {
74 return Ok(None);
75 };
76
77 // Check for an active rotation
78 let pending: Option<(String, i32)> = sqlx::query_as(
79 "SELECT new_encrypted_key, new_key_id FROM sync_key_rotations WHERE app_id = $1 AND user_id = $2",
80 )
81 .bind(app_id)
82 .bind(user_id)
83 .fetch_optional(pool)
84 .await?;
85
86 Ok(Some(SyncKeyInfo {
87 encrypted_key,
88 key_version,
89 key_id,
90 pending_key: pending,
91 }))
92 }
93
94 /// Get device count and sync log entry count for all apps owned by a creator.
95 /// Returns Vec of (app_id, device_count, log_entry_count). Single query replaces N+1 loop.
96 #[tracing::instrument(skip_all)]
97 pub async fn get_sync_app_stats_batch(
98 pool: &PgPool,
99 creator_id: UserId,
100 ) -> Result<Vec<(SyncAppId, i64, i64)>> {
101 let rows: Vec<(SyncAppId, i64, i64)> = sqlx::query_as(
102 r"
103 SELECT
104 a.id,
105 (SELECT COUNT(*) FROM sync_devices d WHERE d.app_id = a.id),
106 (SELECT COUNT(*) FROM sync_log l WHERE l.app_id = a.id)
107 FROM sync_apps a
108 WHERE a.creator_id = $1
109 ",
110 )
111 .bind(creator_id)
112 .fetch_all(pool)
113 .await?;
114
115 Ok(rows)
116 }
117
118 /// Delete sync log entries older than the given number of days.
119 ///
120 /// `retain_days` must be positive. Returns 0 immediately for non-positive values
121 /// to prevent accidental deletion of all entries.
122 #[tracing::instrument(skip_all)]
123 pub async fn prune_sync_log(pool: &PgPool, retain_days: i64) -> Result<u64> {
124 if retain_days <= 0 {
125 tracing::warn!(
126 "prune_sync_log called with non-positive retain_days={retain_days}, skipping"
127 );
128 return Ok(0);
129 }
130 let result = sqlx::query(
131 "DELETE FROM sync_log WHERE created_at < NOW() - make_interval(days => $1::int)",
132 )
133 .bind(retain_days)
134 .execute(pool)
135 .await?;
136
137 Ok(result.rows_affected())
138 }
139
140 /// Compact the sync log by removing entries that all devices for a given
141 /// (app_id, user_id) have already pulled. Keeps a safety margin of entries
142 /// newer than `min_age_days` regardless of cursor positions.
143 ///
144 /// Returns the number of entries deleted.
145 #[allow(dead_code)] // Public API for targeted per-user compaction
146 #[tracing::instrument(skip_all)]
147 pub async fn compact_sync_log(
148 pool: &PgPool,
149 app_id: SyncAppId,
150 user_id: UserId,
151 min_age_days: i64,
152 ) -> Result<u64> {
153 if min_age_days <= 0 {
154 return Ok(0);
155 }
156
157 // Find the lowest cursor across all devices for this user+app.
158 // Entries below this seq have been pulled by every device.
159 let min_cursor: Option<i64> = sqlx::query_scalar(
160 "SELECT MIN(last_pulled_seq) FROM sync_devices WHERE app_id = $1 AND user_id = $2",
161 )
162 .bind(app_id)
163 .bind(user_id)
164 .fetch_one(pool)
165 .await?;
166
167 let safe_seq = match min_cursor {
168 Some(seq) if seq > 0 => seq,
169 _ => return Ok(0), // No devices or no pulls yet
170 };
171
172 // Delete entries below the safe cursor AND older than the safety margin.
173 let result = sqlx::query(
174 r"
175 DELETE FROM sync_log
176 WHERE app_id = $1 AND user_id = $2
177 AND seq <= $3
178 AND created_at < NOW() - make_interval(days => $4::int)
179 ",
180 )
181 .bind(app_id)
182 .bind(user_id)
183 .bind(safe_seq)
184 .bind(min_age_days)
185 .execute(pool)
186 .await?;
187
188 Ok(result.rows_affected())
189 }
190
191 /// Compact sync logs for all user+app pairs that have compactable entries.
192 /// Finds pairs where MIN(last_pulled_seq) across devices > 0, then deletes
193 /// entries below that threshold (with age safety margin).
194 ///
195 /// Returns total entries deleted across all users.
196 #[tracing::instrument(skip_all)]
197 pub async fn compact_all_sync_logs(pool: &PgPool, min_age_days: i64) -> Result<u64> {
198 if min_age_days <= 0 {
199 return Ok(0);
200 }
201
202 // Find (app_id, user_id) pairs where compaction is safe: EVERY registered
203 // device has pulled past the compaction floor. We deliberately do NOT filter
204 // `last_pulled_seq > 0` in the WHERE, a freshly-registered device sits at
205 // seq 0 until its first pull, and excluding it would let us delete entries
206 // it has never seen, so its first pull-from-0 would silently miss changes.
207 // Including every device means a single not-yet-pulled device pins MIN to 0
208 // and the `HAVING MIN > 0` skips the pair entirely until it catches up. A
209 // truly-abandoned never-pulled device is bounded by `prune_sync_log`'s
210 // retention backstop, not this cursor path.
211 // One set-based DELETE over every compactable pair instead of a SELECT + a
212 // serial DELETE per (app, user) pair on the shared pool (ultra-fuzz Run 6,
213 // "SyncKit compaction at scale"). The `floors` CTE reproduces the per-pair
214 // safe sequence; the never-pulled-device safety still rides on `HAVING MIN > 0`
215 // over ALL devices, so a freshly-registered device at seq 0 pins its pair out
216 // of compaction until it catches up.
217 let result = sqlx::query(
218 r"
219 DELETE FROM sync_log sl
220 USING (
221 SELECT app_id, user_id, MIN(last_pulled_seq) AS safe_seq
222 FROM sync_devices
223 GROUP BY app_id, user_id
224 HAVING MIN(last_pulled_seq) > 0
225 ) floors
226 WHERE sl.app_id = floors.app_id
227 AND sl.user_id = floors.user_id
228 AND sl.seq <= floors.safe_seq
229 AND sl.created_at < NOW() - make_interval(days => $1::int)
230 ",
231 )
232 .bind(min_age_days)
233 .execute(pool)
234 .await?;
235
236 Ok(result.rows_affected())
237 }
238