Skip to main content

max / makenotwork

15.1 KB · 458 lines History Blame Raw
1 //! Paid-only first-party (internal) SyncKit: a logged-in user of an internal
2 //! app (GO/BB/AF) may sync only with an `active` end-user subscription, and the
3 //! per-user `storage_limit_bytes` is enforced atomically from the authoritative
4 //! S3 object size, never the client's declared size.
5
6 use std::fmt::Write as _;
7 use std::sync::Arc;
8
9 use makenotwork::db::{SyncAppId, UserId};
10 use serde_json::json;
11 use sqlx::PgPool;
12
13 use crate::harness::{
14 BuildOptions, TestHarness, client::TestResponse, storage::InMemoryStorage, stripe,
15 };
16
17 const GIB: i64 = 1024 * 1024 * 1024;
18
19 pub(crate) async fn harness_with_blobs() -> (TestHarness, Arc<InMemoryStorage>) {
20 let synckit_mem = Arc::new(InMemoryStorage::new());
21 let mock_stripe = Arc::new(stripe::MockPaymentProvider::new());
22 let mut h = TestHarness::build(BuildOptions {
23 synckit_storage: Some(synckit_mem.clone()),
24 stripe_client: Some(mock_stripe.clone()),
25 payment_caps: makenotwork::payments::PaymentCapabilities::all(mock_stripe.clone()),
26 ..Default::default()
27 })
28 .await;
29 h.mock_stripe = Some(mock_stripe);
30 (h, synckit_mem)
31 }
32
33 /// Create an active first-party (internal) sync app.
34 pub(crate) async fn create_internal_app(pool: &PgPool, user_id: UserId) -> (SyncAppId, String) {
35 let api_key = "test-api-key-paid-sync";
36 let key_hash = crate::harness::hash_api_key(api_key);
37 let key_prefix = &api_key[..8];
38 let app_id: SyncAppId = sqlx::query_scalar(
39 "INSERT INTO sync_apps (creator_id, name, api_key_hash, api_key_prefix, is_internal, billing_status)
40 VALUES ($1, 'GoingsOn', $2, $3, TRUE, 'active')
41 RETURNING id",
42 )
43 .bind(user_id)
44 .bind(&key_hash)
45 .bind(key_prefix)
46 .fetch_one(pool)
47 .await
48 .expect("insert internal sync_app");
49 (app_id, api_key.to_string())
50 }
51
52 /// Seed an end-user subscription row with the given status and cap.
53 pub(crate) async fn seed_subscription(
54 pool: &PgPool,
55 user_id: UserId,
56 app_id: SyncAppId,
57 status: &str,
58 limit_bytes: i64,
59 ) {
60 sqlx::query(
61 "INSERT INTO app_sync_subscriptions
62 (user_id, app_id, stripe_subscription_id, stripe_customer_id, tier, status, storage_limit_bytes)
63 VALUES ($1, $2, $3, 'cus_test', 'monthly', $4, $5)",
64 )
65 .bind(user_id)
66 .bind(app_id)
67 .bind(format!("sub_{user_id}_{app_id}"))
68 .bind(status)
69 .bind(limit_bytes)
70 .execute(pool)
71 .await
72 .expect("seed subscription");
73 }
74
75 pub(crate) fn auth_as(h: &mut TestHarness, user_id: UserId, app_id: SyncAppId, key: &str) {
76 let token = makenotwork::synckit_auth::create_sync_token(
77 "test-synckit-jwt-secret",
78 user_id,
79 app_id,
80 key,
81 )
82 .expect("mint test JWT");
83 h.client.set_bearer_token(&token);
84 }
85
86 pub(crate) fn fake_hash(seed: u8) -> String {
87 let mut s = String::with_capacity(64);
88 for _ in 0..32 {
89 write!(s, "{seed:02x}").unwrap();
90 }
91 s
92 }
93
94 /// Put a blob of `actual_bytes` directly into storage, then call confirm with
95 /// the (possibly different) `declared_bytes`. Returns the confirm response.
96 async fn put_and_confirm(
97 h: &mut TestHarness,
98 blobs: &Arc<InMemoryStorage>,
99 app_id: SyncAppId,
100 user_id: UserId,
101 hash: &str,
102 actual_bytes: usize,
103 declared_bytes: i64,
104 ) -> TestResponse {
105 let s3_key = format!("{app_id}/{user_id}/{hash}");
106 blobs.put(&s3_key, vec![0u8; actual_bytes]);
107 h.client
108 .post_json(
109 "/api/sync/blobs/confirm",
110 &json!({ "hash": hash, "size_bytes": declared_bytes }).to_string(),
111 )
112 .await
113 }
114
115 // ── Tests ──
116
117 #[tokio::test]
118 async fn upload_url_refused_without_subscription() {
119 let (mut h, _blobs) = harness_with_blobs().await;
120 let user_id = h
121 .signup("paid_nosub", "paid_nosub@example.com", "Password1!")
122 .await;
123 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
124 auth_as(&mut h, user_id, app_id, "user-key");
125
126 let resp = h
127 .client
128 .post_json(
129 "/api/sync/blobs/upload",
130 &json!({ "hash": fake_hash(0x01), "size_bytes": 100 }).to_string(),
131 )
132 .await;
133 assert_eq!(
134 resp.status, 402,
135 "upload-url must be refused without a sub: {}",
136 resp.text
137 );
138 let body: serde_json::Value = resp.json();
139 assert_eq!(body["reason"], "no_subscription");
140 }
141
142 #[tokio::test]
143 async fn confirm_refused_without_subscription() {
144 let (mut h, blobs) = harness_with_blobs().await;
145 let user_id = h
146 .signup("paid_nosub2", "paid_nosub2@example.com", "Password1!")
147 .await;
148 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
149 auth_as(&mut h, user_id, app_id, "user-key");
150
151 // Bypass the upload-url gate and put the object directly to exercise the
152 // confirm-side subscription check.
153 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x02), 100, 100).await;
154 assert_eq!(
155 resp.status, 402,
156 "confirm must be refused without a sub: {}",
157 resp.text
158 );
159 let body: serde_json::Value = resp.json();
160 assert_eq!(body["reason"], "no_subscription");
161 }
162
163 #[tokio::test]
164 async fn push_refused_without_subscription() {
165 let (mut h, _blobs) = harness_with_blobs().await;
166 let user_id = h
167 .signup("paid_push", "paid_push@example.com", "Password1!")
168 .await;
169 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
170 auth_as(&mut h, user_id, app_id, "user-key");
171
172 let resp = h
173 .client
174 .post_json(
175 "/api/sync/push",
176 &json!({
177 "device_id": uuid::Uuid::new_v4().to_string(),
178 "batch_id": uuid::Uuid::new_v4().to_string(),
179 "changes": [
180 { "table": "tasks", "op": "INSERT", "row_id": "a", "timestamp": "2025-01-01T00:00:00Z", "data": {"t": 1} }
181 ]
182 })
183 .to_string(),
184 )
185 .await;
186 assert_eq!(
187 resp.status, 402,
188 "push must be refused without a sub: {}",
189 resp.text
190 );
191 let body: serde_json::Value = resp.json();
192 assert_eq!(body["reason"], "no_subscription");
193 }
194
195 #[tokio::test]
196 async fn confirm_succeeds_within_cap_with_active_subscription() {
197 let (mut h, blobs) = harness_with_blobs().await;
198 let user_id = h
199 .signup("paid_ok", "paid_ok@example.com", "Password1!")
200 .await;
201 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
202 seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
203 auth_as(&mut h, user_id, app_id, "user-key");
204
205 let resp = put_and_confirm(
206 &mut h,
207 &blobs,
208 app_id,
209 user_id,
210 &fake_hash(0x03),
211 1000,
212 1000,
213 )
214 .await;
215 assert_eq!(
216 resp.status, 204,
217 "confirm within cap should succeed: {}",
218 resp.text
219 );
220
221 // The blob is recorded with the authoritative size.
222 let stored: i64 = sqlx::query_scalar(
223 "SELECT size_bytes FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
224 )
225 .bind(app_id)
226 .bind(user_id)
227 .bind(fake_hash(0x03))
228 .fetch_one(&h.db)
229 .await
230 .unwrap();
231 assert_eq!(stored, 1000);
232 }
233
234 #[tokio::test]
235 async fn confirm_rejected_over_cap() {
236 let (mut h, blobs) = harness_with_blobs().await;
237 let user_id = h
238 .signup("paid_over", "paid_over@example.com", "Password1!")
239 .await;
240 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
241 seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
242 auth_as(&mut h, user_id, app_id, "user-key");
243
244 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x04), 100, 100).await;
245 assert_eq!(
246 resp.status, 402,
247 "confirm over cap should be refused: {}",
248 resp.text
249 );
250 let body: serde_json::Value = resp.json();
251 assert_eq!(body["reason"], "storage_limit_reached");
252 assert_eq!(body["dimension"], "storage");
253 assert_eq!(body["limit"], 50);
254 }
255
256 #[tokio::test]
257 async fn confirm_uses_authoritative_object_size_not_client_claim() {
258 let (mut h, blobs) = harness_with_blobs().await;
259 let user_id = h
260 .signup("paid_spoof", "paid_spoof@example.com", "Password1!")
261 .await;
262 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
263 seed_subscription(&h.db, user_id, app_id, "active", 50).await; // 50-byte cap
264 auth_as(&mut h, user_id, app_id, "user-key");
265
266 // Client lies: declares 1 byte but the actual S3 object is 100 bytes. The
267 // cap (50) must be enforced against the real 100, not the claimed 1.
268 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &fake_hash(0x05), 100, 1).await;
269 assert_eq!(
270 resp.status, 402,
271 "size spoof (declare 1, upload 100) must be caught by object_size: {}",
272 resp.text
273 );
274 let body: serde_json::Value = resp.json();
275 assert_eq!(body["reason"], "storage_limit_reached");
276 }
277
278 #[tokio::test]
279 async fn delete_blob_removes_row_dead_letters_object_and_frees_space() {
280 let (mut h, blobs) = harness_with_blobs().await;
281 let user_id = h
282 .signup("paid_del", "paid_del@example.com", "Password1!")
283 .await;
284 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
285 seed_subscription(&h.db, user_id, app_id, "active", 1500).await; // 1500-byte cap
286 auth_as(&mut h, user_id, app_id, "user-key");
287
288 // Fill most of the cap with one 1000-byte blob.
289 let hash = fake_hash(0x06);
290 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash, 1000, 1000).await;
291 assert_eq!(
292 resp.status, 204,
293 "first confirm should succeed: {}",
294 resp.text
295 );
296
297 // A second 1000-byte blob would breach the 1500 cap (1000 + 1000 > 1500).
298 let hash2 = fake_hash(0x07);
299 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
300 assert_eq!(
301 resp.status, 402,
302 "second confirm should breach the cap: {}",
303 resp.text
304 );
305
306 // Delete the first blob.
307 let s3_key = format!("{app_id}/{user_id}/{hash}");
308 let resp = h.client.delete(&format!("/api/sync/blobs/{hash}")).await;
309 assert_eq!(resp.status, 204, "delete should succeed: {}", resp.text);
310
311 // Row is gone.
312 let remaining: i64 = sqlx::query_scalar(
313 "SELECT COUNT(*) FROM sync_blobs WHERE app_id = $1 AND user_id = $2 AND hash = $3",
314 )
315 .bind(app_id)
316 .bind(user_id)
317 .bind(&hash)
318 .fetch_one(&h.db)
319 .await
320 .unwrap();
321 assert_eq!(remaining, 0, "deleted blob row must be gone");
322
323 // The S3 object was dead-lettered for the synckit bucket, not dropped.
324 let dead_lettered: i64 = sqlx::query_scalar(
325 "SELECT COUNT(*) FROM pending_s3_deletions
326 WHERE s3_key = $1 AND bucket = 'synckit' AND source = 'synckit_blob_delete'",
327 )
328 .bind(&s3_key)
329 .fetch_one(&h.db)
330 .await
331 .unwrap();
332 assert_eq!(
333 dead_lettered, 1,
334 "deleted object must be enqueued to the dead-letter ladder"
335 );
336
337 // Space is freed: the previously-rejected second blob now fits (internal
338 // usage is summed from sync_blobs, so removing the row shrinks usage).
339 let resp = put_and_confirm(&mut h, &blobs, app_id, user_id, &hash2, 1000, 1000).await;
340 assert_eq!(
341 resp.status, 204,
342 "after delete, the second blob should fit: {}",
343 resp.text
344 );
345 }
346
347 #[tokio::test]
348 async fn compaction_waits_for_a_freshly_registered_device() {
349 // Regression: a new device sits at last_pulled_seq = 0 until its first pull.
350 // Compaction must NOT delete log entries below the other devices' cursors
351 // while a never-pulled device exists, or that device's first pull-from-0
352 // would silently miss the compacted-away changes.
353 let (mut h, _blobs) = harness_with_blobs().await;
354 let user_id = h
355 .signup("compact_dev", "compact_dev@example.com", "Password1!")
356 .await;
357 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
358
359 // Device A has pulled; device B is freshly registered (seq 0).
360 let dev_a: uuid::Uuid = sqlx::query_scalar(
361 "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
362 VALUES ($1, $2, 'A', 'test') RETURNING id",
363 )
364 .bind(app_id)
365 .bind(user_id)
366 .fetch_one(&h.db)
367 .await
368 .unwrap();
369 let _dev_b: uuid::Uuid = sqlx::query_scalar(
370 "INSERT INTO sync_devices (app_id, user_id, device_name, platform)
371 VALUES ($1, $2, 'B', 'test') RETURNING id",
372 )
373 .bind(app_id)
374 .bind(user_id)
375 .fetch_one(&h.db)
376 .await
377 .unwrap();
378
379 // One old log entry (backdated well past the 7-day compaction margin).
380 let seq: i64 = sqlx::query_scalar(
381 "INSERT INTO sync_log
382 (app_id, user_id, device_id, table_name, operation, row_id, client_timestamp, created_at)
383 VALUES ($1, $2, $3, 'tasks', 'INSERT', 'r1', NOW(), NOW() - INTERVAL '30 days')
384 RETURNING seq",
385 )
386 .bind(app_id).bind(user_id).bind(dev_a).fetch_one(&h.db).await.unwrap();
387
388 // Device A has pulled past the entry; device B has not pulled at all.
389 sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE id = $2")
390 .bind(seq)
391 .bind(dev_a)
392 .execute(&h.db)
393 .await
394 .unwrap();
395
396 // With device B pinned at 0, compaction must delete nothing.
397 let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7)
398 .await
399 .unwrap();
400 assert_eq!(
401 deleted, 0,
402 "compaction must wait for the never-pulled device"
403 );
404 let present: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM sync_log WHERE seq = $1")
405 .bind(seq)
406 .fetch_one(&h.db)
407 .await
408 .unwrap();
409 assert_eq!(
410 present, 1,
411 "log entry must survive while a new device sits at seq 0"
412 );
413
414 // Once device B catches up, the same entry is safely compactable.
415 sqlx::query("UPDATE sync_devices SET last_pulled_seq = $1 WHERE device_name = 'B'")
416 .bind(seq)
417 .execute(&h.db)
418 .await
419 .unwrap();
420 let deleted = makenotwork::db::synckit::compact_all_sync_logs(&h.db, 7)
421 .await
422 .unwrap();
423 assert_eq!(
424 deleted, 1,
425 "after every device caught up, the old entry compacts"
426 );
427 }
428
429 #[tokio::test]
430 async fn delete_absent_blob_is_idempotent_no_op() {
431 let (mut h, _blobs) = harness_with_blobs().await;
432 let user_id = h
433 .signup("paid_del2", "paid_del2@example.com", "Password1!")
434 .await;
435 let (app_id, _api_key) = create_internal_app(&h.db, user_id).await;
436 seed_subscription(&h.db, user_id, app_id, "active", GIB).await;
437 auth_as(&mut h, user_id, app_id, "user-key");
438
439 // Deleting a hash that was never stored is a 204 no-op, and enqueues nothing.
440 let resp = h
441 .client
442 .delete(&format!("/api/sync/blobs/{}", fake_hash(0x08)))
443 .await;
444 assert_eq!(
445 resp.status, 204,
446 "deleting an absent blob should be a 204 no-op: {}",
447 resp.text
448 );
449
450 let enqueued: i64 = sqlx::query_scalar(
451 "SELECT COUNT(*) FROM pending_s3_deletions WHERE source = 'synckit_blob_delete'",
452 )
453 .fetch_one(&h.db)
454 .await
455 .unwrap();
456 assert_eq!(enqueued, 0, "no-op delete must not dead-letter anything");
457 }
458