Skip to main content

max / makenotwork

32.7 KB · 918 lines History Blame Raw
1 //! SyncKit push/pull, status, device management, and key management endpoints.
2
3 use axum::{
4 Json,
5 extract::{Path, State},
6 http::StatusCode,
7 response::{IntoResponse, Response},
8 };
9 use serde_json::json;
10
11 use sqlx::PgPool;
12
13 use crate::{
14 config::Config,
15 constants,
16 db::{self, SyncDeviceId},
17 error::{AppError, Result},
18 payments::{self, SyncBillingInterval},
19 synckit_auth::SyncUser,
20 validation,
21 };
22
23 use super::{
24 AppPricingRequest, AppPricingResponse, BeginRotationRequest, BeginRotationResponse,
25 CompleteRotationErrorResponse, CompleteRotationRequest, GetKeyResponse, PendingKeyInfo,
26 PullChangeEntry, PullRequest, PullResponse, PushRequest, PushResponse, PutKeyRequest,
27 RegisterDeviceRequest, RotationBatchRequest, RotationBatchResponse, RotationEntriesRequest,
28 RotationEntriesResponse, RotationEntry, SyncAccountResponse, SyncCapChangeRequest,
29 SyncCheckoutResponse, SyncDeviceResponse, SyncQuoteRequest, SyncQuoteResponse,
30 SyncStatusResponse, SyncSubscribeRequest, SyncSubscriptionStatusResponse,
31 };
32
33 // --- Sync endpoints (JWT auth) ---
34
35 /// Push encrypted changelog entries from a device.
36 #[utoipa::path(post, path = "/api/v1/sync/push", tag = "SyncKit",
37 request_body = PushRequest,
38 responses((status = 200, description = "New cursor position", body = PushResponse)),
39 security(("bearer" = [])),
40 )]
41 #[tracing::instrument(skip_all, name = "synckit::sync_push", fields(app_id, user_id))]
42 pub(super) async fn sync_push(
43 State(db): State<PgPool>,
44 State(sync): State<crate::Sync>,
45 sync_user: SyncUser,
46 Json(req): Json<PushRequest>,
47 ) -> Result<Response> {
48 let app_id = sync_user.app_id;
49 let user_id = sync_user.user_id;
50 tracing::Span::current().record("app_id", tracing::field::display(&app_id));
51 tracing::Span::current().record("user_id", tracing::field::display(&user_id));
52
53 // Paid-only sync: first-party apps require an active end-user subscription
54 // to write. Reads (pull) stay open so a lapsed user can still export their
55 // data. Non-internal (developer-billed) apps are always allowed here.
56 if !db::synckit::internal_write_allowed(&db, app_id, user_id).await? {
57 return Ok((
58 StatusCode::PAYMENT_REQUIRED,
59 Json(json!({ "reason": "no_subscription" })),
60 )
61 .into_response());
62 }
63
64 if req.changes.is_empty() {
65 return Err(AppError::BadRequest("No changes provided".to_string()));
66 }
67 if req.changes.len() > constants::SYNCKIT_PUSH_MAX_CHANGES {
68 return Err(AppError::BadRequest(format!(
69 "Maximum {} changes per push",
70 constants::SYNCKIT_PUSH_MAX_CHANGES
71 )));
72 }
73
74 // Validate all changes
75 for change in &req.changes {
76 validation::validate_sync_table_name(&change.table)?;
77 validation::validate_sync_row_id(&change.row_id)?;
78 if change.op == db::SyncOperation::Delete && change.data.is_some() {
79 return Err(AppError::BadRequest(
80 "DELETE operations should not include data".to_string(),
81 ));
82 }
83 }
84
85 // Verify device belongs to this user + app (indexed point lookup, not a
86 // full device fetch + linear scan).
87 if !db::synckit::sync_device_belongs(&db, req.device_id, app_id, user_id).await? {
88 return Err(AppError::BadRequest("Unknown device".to_string()));
89 }
90
91 db::synckit::touch_sync_device(&db, req.device_id).await?;
92
93 // Build change tuples (op converted to string for TEXT column)
94 let changes: Vec<_> = req
95 .changes
96 .iter()
97 .map(|c| {
98 (
99 c.table.clone(),
100 c.op.to_string(),
101 c.row_id.clone(),
102 c.timestamp,
103 c.data.clone(),
104 )
105 })
106 .collect();
107
108 let cursor =
109 db::synckit::push_sync_changes(&db, app_id, user_id, req.device_id, req.batch_id, &changes)
110 .await?;
111
112 // Notify SSE subscribers, carrying the new max seq so a device already at
113 // (or past) this cursor can skip a redundant pull, avoids the thundering
114 // herd where every subscriber pulls on every push.
115 sync.notify_push(app_id, user_id, cursor);
116
117 Ok(Json(PushResponse { cursor }).into_response())
118 }
119
120 /// Pull changelog entries after a given cursor.
121 #[utoipa::path(post, path = "/api/v1/sync/pull", tag = "SyncKit",
122 request_body = PullRequest,
123 responses((status = 200, description = "Changes since cursor", body = PullResponse)),
124 security(("bearer" = [])),
125 )]
126 #[tracing::instrument(skip_all, name = "synckit::sync_pull", fields(app_id, user_id))]
127 pub(super) async fn sync_pull(
128 State(db): State<PgPool>,
129 sync_user: SyncUser,
130 headers: axum::http::HeaderMap,
131 Json(req): Json<PullRequest>,
132 ) -> Result<impl IntoResponse> {
133 let app_id = sync_user.app_id;
134 let user_id = sync_user.user_id;
135 tracing::Span::current().record("app_id", tracing::field::display(&app_id));
136 tracing::Span::current().record("user_id", tracing::field::display(&user_id));
137
138 // Verify device belongs to this user + app (indexed point lookup, not a
139 // full device fetch + linear scan).
140 if !db::synckit::sync_device_belongs(&db, req.device_id, app_id, user_id).await? {
141 return Err(AppError::BadRequest("Unknown device".to_string()));
142 }
143
144 // Validate table name filters if provided
145 if let Some(ref tables) = req.tables {
146 if tables.len() > 50 {
147 return Err(AppError::BadRequest(
148 "Maximum 50 table names per filter".to_string(),
149 ));
150 }
151 for table in tables {
152 validation::validate_sync_table_name(table)?;
153 }
154 }
155
156 let page_size = constants::SYNCKIT_PULL_PAGE_SIZE;
157 let entries = db::synckit::pull_sync_changes_filtered(
158 &db,
159 app_id,
160 user_id,
161 req.cursor,
162 page_size,
163 req.tables.as_deref(),
164 req.since,
165 )
166 .await?;
167
168 let has_more = entries.len() as i64 == page_size;
169 let new_cursor = entries.last().map_or(req.cursor, |e| e.seq);
170
171 // Mark the device seen and advance its compaction cursor in one statement.
172 // GREATEST keeps the cursor monotonic even if `new_cursor == req.cursor`.
173 db::synckit::touch_and_advance_cursor(
174 &db,
175 req.device_id,
176 new_cursor,
177 super::client_version(&headers).as_deref(),
178 )
179 .await?;
180
181 let changes: Vec<PullChangeEntry> = entries
182 .into_iter()
183 .map(|e| PullChangeEntry {
184 seq: e.seq,
185 device_id: e.device_id,
186 table: e.table_name,
187 op: e.operation.to_string(),
188 row_id: e.row_id,
189 timestamp: e.client_timestamp,
190 data: e.data,
191 key_id: e.key_id,
192 // Personal entries key off `key_id`; GCK generations are group-only.
193 gck_version: None,
194 })
195 .collect();
196
197 Ok(Json(PullResponse {
198 changes,
199 cursor: new_cursor,
200 has_more,
201 }))
202 }
203
204 /// Return sync metadata for the authenticated user and app.
205 #[utoipa::path(get, path = "/api/v1/sync/status", tag = "SyncKit",
206 responses((status = 200, description = "Sync status", body = SyncStatusResponse)),
207 security(("bearer" = [])),
208 )]
209 #[tracing::instrument(skip_all, name = "synckit::sync_status")]
210 pub(super) async fn sync_status(
211 State(db): State<PgPool>,
212 sync_user: SyncUser,
213 ) -> Result<impl IntoResponse> {
214 let app_id = sync_user.app_id;
215 let user_id = sync_user.user_id;
216
217 let (total_changes, latest_cursor) = db::synckit::get_sync_status(&db, app_id, user_id).await?;
218
219 Ok(Json(SyncStatusResponse {
220 total_changes,
221 latest_cursor,
222 }))
223 }
224
225 /// Return the authenticated user's email and username, for the app to display
226 /// "logged in as ..." in its sync UI.
227 #[utoipa::path(get, path = "/api/v1/sync/account", tag = "SyncKit",
228 responses((status = 200, description = "Account info", body = SyncAccountResponse)),
229 security(("bearer" = [])),
230 )]
231 #[tracing::instrument(skip_all, name = "synckit::sync_account")]
232 pub(super) async fn sync_account(
233 State(db): State<PgPool>,
234 sync_user: SyncUser,
235 ) -> Result<impl IntoResponse> {
236 let user = db::users::get_user_by_id(&db, sync_user.user_id)
237 .await?
238 .ok_or(AppError::NotFound)?;
239
240 Ok(Json(SyncAccountResponse {
241 email: user.email.as_str().to_string(),
242 username: user.username.as_str().to_string(),
243 }))
244 }
245
246 /// Return the authenticated user's subscription status for this app.
247 /// Returns `active: false` and `None` fields when the user has no subscription
248 /// (rather than 404) so clients can render a "subscribe" CTA uniformly.
249 #[utoipa::path(get, path = "/api/v1/sync/subscription", tag = "SyncKit",
250 responses((status = 200, description = "Subscription status", body = SyncSubscriptionStatusResponse)),
251 security(("bearer" = [])),
252 )]
253 #[tracing::instrument(skip_all, name = "synckit::sync_subscription_status")]
254 pub(super) async fn sync_subscription_status(
255 State(db): State<PgPool>,
256 sync_user: SyncUser,
257 ) -> Result<impl IntoResponse> {
258 let sub =
259 db::synckit::get_user_app_subscription(&db, sync_user.user_id, sync_user.app_id).await?;
260
261 let response = match sub {
262 Some(s) => SyncSubscriptionStatusResponse {
263 active: s.status == "active",
264 tier: Some(s.interval),
265 status: Some(s.status),
266 storage_limit_bytes: s.storage_limit_bytes,
267 pending_storage_limit_bytes: s.pending_storage_limit_bytes,
268 storage_used_bytes: Some(
269 db::synckit::storage_used_bytes(&db, sync_user.app_id, sync_user.user_id).await?,
270 ),
271 current_period_end: s.current_period_end.map(|t| t.to_rfc3339()),
272 },
273 None => SyncSubscriptionStatusResponse {
274 active: false,
275 tier: None,
276 status: None,
277 storage_limit_bytes: None,
278 pending_storage_limit_bytes: None,
279 storage_used_bytes: None,
280 current_period_end: None,
281 },
282 };
283
284 Ok(Json(response))
285 }
286
287 /// Return the pricing-formula constants for an app. The client uses these to
288 /// quote a price locally as the user adjusts the cap slider; the same formula
289 /// is enforced server-side at checkout so the client number is only advisory.
290 #[utoipa::path(post, path = "/api/v1/sync/app/pricing", tag = "SyncKit",
291 request_body = AppPricingRequest,
292 responses((status = 200, description = "Pricing formula", body = AppPricingResponse)),
293 )]
294 #[tracing::instrument(skip_all, name = "synckit::get_app_pricing")]
295 pub(super) async fn get_app_pricing(
296 State(db): State<PgPool>,
297 Json(req): Json<AppPricingRequest>,
298 ) -> Result<impl IntoResponse> {
299 let app = db::synckit::get_sync_app_by_api_key(&db, &req.api_key)
300 .await?
301 .ok_or(AppError::NotFound)?;
302
303 Ok(Json(AppPricingResponse {
304 app_name: app.name,
305 min_charge_cents: payments::MIN_CHARGE_CENTS,
306 per_gb_tenths_of_cent_per_month:
307 payments::synckit_app_pricing::PER_GB_TENTHS_OF_CENT_PER_MONTH,
308 annual_multiplier: payments::ANNUAL_MULTIPLIER,
309 min_cap_bytes: payments::MIN_CAP_BYTES,
310 max_cap_bytes: payments::MAX_CAP_BYTES,
311 }))
312 }
313
314 /// Quote the price for a (cap, interval) pair. Authenticated so clients
315 /// cannot scrape pricing without an account, but otherwise pure: the result
316 /// only depends on the formula constants returned by `app/pricing`.
317 #[utoipa::path(post, path = "/api/v1/sync/subscription/quote", tag = "SyncKit",
318 request_body = SyncQuoteRequest,
319 responses((status = 200, description = "Quoted price", body = SyncQuoteResponse)),
320 security(("bearer" = [])),
321 )]
322 #[tracing::instrument(skip_all, name = "synckit::quote_subscription_price")]
323 pub(super) async fn quote_subscription_price(
324 _sync_user: SyncUser,
325 Json(req): Json<SyncQuoteRequest>,
326 ) -> Result<impl IntoResponse> {
327 let interval = SyncBillingInterval::parse(&req.interval)?;
328 let price_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
329 Ok(Json(SyncQuoteResponse {
330 cap_bytes: req.cap_bytes,
331 interval: interval.as_str().to_string(),
332 price_cents,
333 }))
334 }
335
336 /// Create a Stripe Checkout Session for subscribing this user to the app's
337 /// cloud sync at their chosen storage cap. The `app_sync_subscriptions` row
338 /// is written by the Stripe webhook on `checkout.session.completed`.
339 #[utoipa::path(post, path = "/api/v1/sync/subscription/checkout", tag = "SyncKit",
340 request_body = SyncSubscribeRequest,
341 responses((status = 200, description = "Checkout URL", body = SyncCheckoutResponse)),
342 security(("bearer" = [])),
343 )]
344 #[tracing::instrument(skip_all, name = "synckit::create_subscription_checkout")]
345 pub(super) async fn create_subscription_checkout(
346 State(db): State<PgPool>,
347 State(payments): State<crate::Billing>,
348 State(config): State<Config>,
349 sync_user: SyncUser,
350 Json(req): Json<SyncSubscribeRequest>,
351 ) -> Result<impl IntoResponse> {
352 if db::synckit::get_user_app_subscription(&db, sync_user.user_id, sync_user.app_id)
353 .await?
354 .is_some()
355 {
356 return Err(AppError::BadRequest(
357 "Already subscribed; use the storage-cap endpoint to adjust your cap".to_string(),
358 ));
359 }
360
361 let interval = SyncBillingInterval::parse(&req.interval)?;
362 let amount_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
363
364 let app = db::synckit::get_sync_app_by_id(&db, sync_user.app_id)
365 .await?
366 .ok_or(AppError::NotFound)?;
367 let cap_gib = req.cap_bytes / (1024 * 1024 * 1024);
368 let product_name = format!("{} cloud sync ({} GiB)", app.name, cap_gib);
369
370 let stripe = payments
371 .stripe
372 .as_ref()
373 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
374
375 let success_url = format!("{}/sync/subscribed", config.host_url);
376 let cancel_url = format!("{}/sync/canceled", config.host_url);
377
378 let result = stripe
379 .create_synckit_app_sub_checkout_session(&crate::payments::SynckitAppSubCheckoutParams {
380 product_name: &product_name,
381 amount_cents,
382 interval: interval.as_str(),
383 user_id: sync_user.user_id,
384 app_id: sync_user.app_id,
385 storage_limit_bytes: Some(req.cap_bytes),
386 success_url: &success_url,
387 cancel_url: &cancel_url,
388 })
389 .await?;
390
391 let checkout_url = result
392 .url
393 .ok_or_else(|| AppError::BadRequest("No checkout URL returned".to_string()))?;
394
395 Ok(Json(SyncCheckoutResponse { checkout_url }))
396 }
397
398 /// Queue a storage-cap change to take effect at the next billing cycle.
399 /// Holding the change for the cycle boundary avoids mid-cycle proration
400 /// surprises and keeps the user in control of when their bill changes.
401 ///
402 /// Updates Stripe first (re-price the subscription item with
403 /// `proration_behavior=None`), then records the pending cap in the DB. The
404 /// renewal webhook promotes the pending cap to active when Stripe rolls the
405 /// period, and because the Stripe price is already updated, the new invoice
406 /// is at the new price.
407 #[utoipa::path(post, path = "/api/v1/sync/subscription/storage-cap", tag = "SyncKit",
408 request_body = SyncCapChangeRequest,
409 responses((status = 200, description = "Pending cap recorded", body = SyncSubscriptionStatusResponse)),
410 security(("bearer" = [])),
411 )]
412 #[tracing::instrument(skip_all, name = "synckit::queue_storage_cap_change")]
413 pub(super) async fn queue_storage_cap_change(
414 State(db): State<PgPool>,
415 State(payments): State<crate::Billing>,
416 sync_user: SyncUser,
417 Json(req): Json<SyncCapChangeRequest>,
418 ) -> Result<impl IntoResponse> {
419 let sub = db::synckit::get_user_app_subscription(&db, sync_user.user_id, sync_user.app_id)
420 .await?
421 .ok_or_else(|| AppError::BadRequest("No active subscription to adjust".to_string()))?;
422
423 if req.cap_bytes < payments::MIN_CAP_BYTES || req.cap_bytes > payments::MAX_CAP_BYTES {
424 return Err(AppError::BadRequest(format!(
425 "Storage cap must be between {} and {} GiB",
426 payments::MIN_CAP_BYTES / (1024 * 1024 * 1024),
427 payments::MAX_CAP_BYTES / (1024 * 1024 * 1024)
428 )));
429 }
430
431 let interval = payments::SyncBillingInterval::parse(&sub.interval)?;
432 let new_price_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
433
434 let stripe = payments
435 .stripe
436 .as_ref()
437 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
438
439 let app = db::synckit::get_sync_app_by_id(&db, sync_user.app_id)
440 .await?
441 .ok_or(AppError::NotFound)?;
442 let cap_gib = req.cap_bytes / (1024 * 1024 * 1024);
443 let product_name = format!("{} cloud sync ({} GiB)", app.name, cap_gib);
444
445 // Stripe first: if this fails, we want the DB pending cap to stay
446 // unchanged so the user isn't sitting on an upgrade they never paid for.
447 stripe
448 .update_synckit_app_sub_price(
449 &sub.stripe_subscription_id,
450 new_price_cents,
451 interval,
452 &product_name,
453 )
454 .await?;
455
456 // An increase takes effect now, a decrease at the next renewal. Stripe was
457 // just re-priced with prorations either way, so queueing an increase would
458 // charge for headroom the user cannot use until the period rolls - and a
459 // user raising their cap is usually one already blocked by a full one.
460 let raising = req.cap_bytes > sub.storage_limit_bytes.unwrap_or(0);
461 if raising {
462 db::synckit::set_storage_cap_now(&db, sync_user.user_id, sync_user.app_id, req.cap_bytes)
463 .await?;
464 } else {
465 db::synckit::set_pending_storage_cap(
466 &db,
467 sync_user.user_id,
468 sync_user.app_id,
469 req.cap_bytes,
470 )
471 .await?;
472 }
473
474 Ok(Json(SyncSubscriptionStatusResponse {
475 active: sub.status == "active",
476 tier: Some(sub.interval),
477 status: Some(sub.status),
478 storage_limit_bytes: if raising {
479 Some(req.cap_bytes)
480 } else {
481 sub.storage_limit_bytes
482 },
483 pending_storage_limit_bytes: (!raising).then_some(req.cap_bytes),
484 storage_used_bytes: Some(
485 db::synckit::storage_used_bytes(&db, sync_user.app_id, sync_user.user_id).await?,
486 ),
487 current_period_end: sub.current_period_end.map(|t| t.to_rfc3339()),
488 }))
489 }
490
491 // --- Device endpoints (JWT auth) ---
492
493 /// Register a new sync device (or update an existing one by name).
494 #[utoipa::path(post, path = "/api/v1/sync/devices", tag = "SyncKit",
495 request_body = RegisterDeviceRequest,
496 responses((status = 200, description = "Registered device", body = SyncDeviceResponse)),
497 security(("bearer" = [])),
498 )]
499 #[tracing::instrument(skip_all, name = "synckit::register_device")]
500 pub(super) async fn register_device(
501 State(db): State<PgPool>,
502 sync_user: SyncUser,
503 headers: axum::http::HeaderMap,
504 Json(req): Json<RegisterDeviceRequest>,
505 ) -> Result<impl IntoResponse> {
506 validation::validate_sync_device_name(&req.device_name)?;
507
508 // Serialize concurrent registrations for this (app, user) so the
509 // count-then-upsert cap check below is race-free. The lock is transaction
510 // scoped; holding `lock_tx` to the end of the handler keeps a second
511 // concurrent registration blocked until this one's upsert commits, so it
512 // sees the updated count and can't push past the cap (Run 21 concurrency).
513 let mut lock_tx = db.begin().await?;
514 sqlx::query(
515 "SELECT pg_advisory_xact_lock(hashtextextended('synckit_device:' || $1::text || ':' || $2::text, 0))",
516 )
517 .bind(sync_user.app_id)
518 .bind(sync_user.user_id)
519 .execute(&mut *lock_tx)
520 .await?;
521
522 // Enforce device limit (upsert on existing name is fine, only new names count)
523 let count = db::synckit::count_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
524 if count >= constants::SYNCKIT_MAX_DEVICES_PER_APP {
525 // Check if this is an existing device (upsert), allow updates
526 let existing =
527 db::synckit::get_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
528 if !existing.iter().any(|d| d.device_name == req.device_name) {
529 return Err(AppError::BadRequest(format!(
530 "Maximum {} devices per app",
531 constants::SYNCKIT_MAX_DEVICES_PER_APP
532 )));
533 }
534 }
535
536 let device = db::synckit::upsert_sync_device(
537 &db,
538 sync_user.app_id,
539 sync_user.user_id,
540 &req.device_name,
541 req.platform,
542 super::client_version(&headers).as_deref(),
543 )
544 .await?;
545
546 Ok(Json(SyncDeviceResponse {
547 id: device.id,
548 app_id: device.app_id,
549 user_id: device.user_id,
550 device_name: device.device_name,
551 platform: device.platform.to_string(),
552 last_seen_at: device.last_seen_at,
553 created_at: device.created_at,
554 }))
555 }
556
557 /// List all devices registered for the authenticated user and app.
558 #[utoipa::path(get, path = "/api/v1/sync/devices", tag = "SyncKit",
559 responses((status = 200, description = "List of devices", body = Vec<SyncDeviceResponse>)),
560 security(("bearer" = [])),
561 )]
562 #[tracing::instrument(skip_all, name = "synckit::list_devices")]
563 pub(super) async fn list_devices(
564 State(db): State<PgPool>,
565 sync_user: SyncUser,
566 ) -> Result<impl IntoResponse> {
567 let devices = db::synckit::get_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
568
569 let response: Vec<SyncDeviceResponse> = devices
570 .into_iter()
571 .map(|d| SyncDeviceResponse {
572 id: d.id,
573 app_id: d.app_id,
574 user_id: d.user_id,
575 device_name: d.device_name,
576 platform: d.platform.to_string(),
577 last_seen_at: d.last_seen_at,
578 created_at: d.created_at,
579 })
580 .collect();
581
582 Ok(Json(response))
583 }
584
585 /// Remove a registered device.
586 #[utoipa::path(delete, path = "/api/v1/sync/devices/{id}", tag = "SyncKit",
587 params(("id" = String, Path, description = "Device ID")),
588 responses((status = 204, description = "Device deleted"), (status = 404, description = "Device not found")),
589 security(("bearer" = [])),
590 )]
591 #[tracing::instrument(skip_all, name = "synckit::delete_device")]
592 pub(super) async fn delete_device(
593 State(db): State<PgPool>,
594 sync_user: SyncUser,
595 headers: axum::http::HeaderMap,
596 Path(device_id): Path<SyncDeviceId>,
597 ) -> Result<impl IntoResponse> {
598 let deleted =
599 db::synckit::delete_sync_device(&db, device_id, sync_user.app_id, sync_user.user_id)
600 .await?;
601
602 if !deleted {
603 return Err(AppError::NotFound);
604 }
605
606 // Security response to a device removal: invalidate the user's sync tokens
607 // so the removed device's JWT dies immediately (before this it survived to
608 // expiry), and record the event for the audit log. Best-effort, a logging
609 // or invalidation hiccup must not fail the delete the user asked for.
610 let ip = crate::helpers::extract_client_ip(&headers);
611 if let Err(e) = db::synckit::invalidate_user_sync_tokens(&db, sync_user.user_id).await {
612 tracing::error!(error = ?e, user_id = %sync_user.user_id,
613 "failed to invalidate sync tokens after device removal");
614 }
615 if let Err(e) = db::synckit::record_security_event(
616 &db,
617 sync_user.app_id,
618 Some(sync_user.user_id),
619 db::synckit::sync_security_event::DEVICE_REMOVED,
620 Some(serde_json::json!({ "device_id": device_id.to_string() })),
621 ip.as_deref(),
622 )
623 .await
624 {
625 tracing::error!(error = ?e, "failed to record device_removed security event");
626 }
627
628 Ok(axum::http::StatusCode::NO_CONTENT)
629 }
630
631 // --- Key management endpoints (JWT auth) ---
632
633 /// Store or update the user's encrypted master key envelope.
634 #[utoipa::path(put, path = "/api/v1/sync/keys", tag = "SyncKit",
635 request_body = PutKeyRequest,
636 responses((status = 204, description = "Key stored"), (status = 409, description = "Version mismatch")),
637 security(("bearer" = [])),
638 )]
639 #[tracing::instrument(skip_all, name = "synckit::put_sync_key")]
640 pub(super) async fn put_sync_key(
641 State(db): State<PgPool>,
642 sync_user: SyncUser,
643 Json(req): Json<PutKeyRequest>,
644 ) -> Result<impl IntoResponse> {
645 // Max 4 KB for the encrypted key envelope
646 if req.encrypted_key.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES {
647 return Err(AppError::BadRequest(
648 "Encrypted key exceeds 4KB limit".to_string(),
649 ));
650 }
651
652 let updated = db::synckit::upsert_sync_key(
653 &db,
654 sync_user.app_id,
655 sync_user.user_id,
656 &req.encrypted_key,
657 req.expected_version,
658 )
659 .await?;
660
661 if !updated {
662 return Err(AppError::Conflict(
663 "Key version mismatch, another device changed the password. Fetch the latest key and retry.".to_string(),
664 ));
665 }
666
667 Ok(axum::http::StatusCode::NO_CONTENT)
668 }
669
670 /// Retrieve the user's encrypted master key envelope.
671 #[utoipa::path(get, path = "/api/v1/sync/keys", tag = "SyncKit",
672 responses((status = 200, description = "Encrypted key envelope", body = GetKeyResponse), (status = 404, description = "No key stored")),
673 security(("bearer" = [])),
674 )]
675 #[tracing::instrument(skip_all, name = "synckit::get_sync_key")]
676 pub(super) async fn get_sync_key(
677 State(db): State<PgPool>,
678 sync_user: SyncUser,
679 ) -> Result<impl IntoResponse> {
680 let info = db::synckit::get_sync_key(&db, sync_user.app_id, sync_user.user_id)
681 .await?
682 .ok_or(AppError::NotFound)?;
683
684 let pending_key = info
685 .pending_key
686 .map(|(encrypted_key, key_id)| PendingKeyInfo {
687 encrypted_key,
688 key_id,
689 });
690
691 Ok(Json(GetKeyResponse {
692 encrypted_key: info.encrypted_key,
693 key_version: info.key_version,
694 key_id: info.key_id,
695 pending_key,
696 }))
697 }
698
699 // --- Key rotation endpoints ---
700
701 /// Begin a key rotation.
702 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate", tag = "SyncKit",
703 request_body = BeginRotationRequest,
704 responses(
705 (status = 200, description = "Rotation started or resumed", body = BeginRotationResponse),
706 (status = 409, description = "Version mismatch or rotation in progress"),
707 ),
708 security(("bearer" = [])),
709 )]
710 #[tracing::instrument(skip_all, name = "synckit::begin_rotation")]
711 pub(super) async fn begin_rotation(
712 State(db): State<PgPool>,
713 sync_user: SyncUser,
714 Json(req): Json<BeginRotationRequest>,
715 ) -> Result<impl IntoResponse> {
716 if req.new_encrypted_key.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES {
717 return Err(AppError::BadRequest(
718 "Encrypted key exceeds 4KB limit".to_string(),
719 ));
720 }
721
722 // Verify device belongs to this user + app via an indexed point lookup, not a
723 // fetch-all-then-scan (ultra-fuzz Run 4 Perf).
724 if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
725 .await?
726 {
727 return Err(AppError::BadRequest("Unknown device".to_string()));
728 }
729
730 let result = db::synckit::begin_key_rotation(
731 &db,
732 sync_user.app_id,
733 sync_user.user_id,
734 req.device_id,
735 &req.new_encrypted_key,
736 req.expected_key_version,
737 )
738 .await?;
739
740 match result {
741 Ok(rotation) => Ok(Json(BeginRotationResponse {
742 rotation_id: rotation.id,
743 target_seq: rotation.target_seq,
744 new_key_id: rotation.new_key_id,
745 })
746 .into_response()),
747 Err(msg) => Err(AppError::Conflict(msg.to_string())),
748 }
749 }
750
751 /// Pull entries that need re-encryption during a rotation.
752 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate/entries", tag = "SyncKit",
753 request_body = RotationEntriesRequest,
754 responses((status = 200, description = "Entries needing re-encryption", body = RotationEntriesResponse)),
755 security(("bearer" = [])),
756 )]
757 #[tracing::instrument(skip_all, name = "synckit::rotation_entries")]
758 pub(super) async fn rotation_entries(
759 State(db): State<PgPool>,
760 sync_user: SyncUser,
761 Json(req): Json<RotationEntriesRequest>,
762 ) -> Result<impl IntoResponse> {
763 let rotation = db::synckit::get_key_rotation(&db, sync_user.app_id, sync_user.user_id)
764 .await?
765 .ok_or_else(|| AppError::BadRequest("No active rotation".to_string()))?;
766
767 if rotation.id != req.rotation_id {
768 return Err(AppError::BadRequest("Rotation ID mismatch".to_string()));
769 }
770
771 let page_size = constants::SYNCKIT_ROTATION_BATCH_MAX as i64;
772 let raw_entries = db::synckit::get_rotation_entries(
773 &db,
774 sync_user.app_id,
775 sync_user.user_id,
776 rotation.new_key_id,
777 req.after_seq,
778 page_size,
779 )
780 .await?;
781
782 let has_more = raw_entries.len() as i64 == page_size;
783 let entries: Vec<RotationEntry> = raw_entries
784 .into_iter()
785 .map(|(seq, table, row_id, data)| RotationEntry {
786 seq,
787 table,
788 row_id,
789 data,
790 })
791 .collect();
792
793 Ok(Json(RotationEntriesResponse { entries, has_more }))
794 }
795
796 /// Submit a batch of re-encrypted entries during rotation.
797 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate/batch", tag = "SyncKit",
798 request_body = RotationBatchRequest,
799 responses((status = 200, description = "Batch processed", body = RotationBatchResponse)),
800 security(("bearer" = [])),
801 )]
802 #[tracing::instrument(skip_all, name = "synckit::rotation_batch")]
803 pub(super) async fn rotation_batch(
804 State(db): State<PgPool>,
805 sync_user: SyncUser,
806 Json(req): Json<RotationBatchRequest>,
807 ) -> Result<impl IntoResponse> {
808 if req.entries.is_empty() {
809 return Err(AppError::BadRequest("No entries provided".to_string()));
810 }
811 if req.entries.len() > constants::SYNCKIT_ROTATION_BATCH_MAX {
812 return Err(AppError::BadRequest(format!(
813 "Maximum {} entries per batch",
814 constants::SYNCKIT_ROTATION_BATCH_MAX
815 )));
816 }
817
818 let rotation = db::synckit::get_key_rotation(&db, sync_user.app_id, sync_user.user_id)
819 .await?
820 .ok_or_else(|| AppError::BadRequest("No active rotation".to_string()))?;
821
822 if rotation.id != req.rotation_id {
823 return Err(AppError::BadRequest("Rotation ID mismatch".to_string()));
824 }
825
826 let entries: Vec<(i64, Option<serde_json::Value>)> =
827 req.entries.into_iter().map(|e| (e.seq, e.data)).collect();
828
829 let updated_count = db::synckit::submit_rotation_batch(
830 &db,
831 sync_user.app_id,
832 sync_user.user_id,
833 rotation.id,
834 rotation.new_key_id,
835 &entries,
836 )
837 .await?;
838
839 Ok(Json(RotationBatchResponse { updated_count }))
840 }
841
842 /// Complete a key rotation.
843 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate/complete", tag = "SyncKit",
844 request_body = CompleteRotationRequest,
845 responses(
846 (status = 204, description = "Rotation completed"),
847 (status = 409, description = "Entries still need re-encryption"),
848 ),
849 security(("bearer" = [])),
850 )]
851 #[tracing::instrument(skip_all, name = "synckit::complete_rotation")]
852 pub(super) async fn complete_rotation(
853 State(db): State<PgPool>,
854 sync_user: SyncUser,
855 Json(req): Json<CompleteRotationRequest>,
856 ) -> Result<impl IntoResponse> {
857 let result = db::synckit::complete_key_rotation(
858 &db,
859 sync_user.app_id,
860 sync_user.user_id,
861 req.rotation_id,
862 )
863 .await?;
864
865 match result {
866 Ok(_new_key_id) => {
867 // Audit the completed rotation (best-effort).
868 if let Err(e) = db::synckit::record_security_event(
869 &db,
870 sync_user.app_id,
871 Some(sync_user.user_id),
872 db::synckit::sync_security_event::KEY_ROTATION_COMPLETED,
873 Some(serde_json::json!({ "rotation_id": req.rotation_id.to_string() })),
874 None,
875 )
876 .await
877 {
878 tracing::error!(error = ?e, "failed to record key_rotation_completed security event");
879 }
880 Ok(axum::http::StatusCode::NO_CONTENT.into_response())
881 }
882 Err(0) => Err(AppError::BadRequest("No active rotation".to_string())),
883 Err(remaining) => Ok((
884 axum::http::StatusCode::CONFLICT,
885 Json(CompleteRotationErrorResponse { remaining }),
886 )
887 .into_response()),
888 }
889 }
890
891 /// Cancel a stale rotation (>24h without activity).
892 #[utoipa::path(delete, path = "/api/v1/sync/keys/rotate", tag = "SyncKit",
893 responses(
894 (status = 204, description = "Stale rotation cancelled"),
895 (status = 404, description = "No stale rotation found"),
896 ),
897 security(("bearer" = [])),
898 )]
899 #[tracing::instrument(skip_all, name = "synckit::cancel_rotation")]
900 pub(super) async fn cancel_rotation(
901 State(db): State<PgPool>,
902 sync_user: SyncUser,
903 ) -> Result<impl IntoResponse> {
904 let cancelled = db::synckit::cancel_stale_rotation(
905 &db,
906 sync_user.app_id,
907 sync_user.user_id,
908 constants::SYNCKIT_ROTATION_STALE_HOURS,
909 )
910 .await?;
911
912 if !cancelled {
913 return Err(AppError::NotFound);
914 }
915
916 Ok(axum::http::StatusCode::NO_CONTENT)
917 }
918