Skip to main content

max / makenotwork

31.9 KB · 895 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: None,
269 current_period_end: s.current_period_end.map(|t| t.to_rfc3339()),
270 },
271 None => SyncSubscriptionStatusResponse {
272 active: false,
273 tier: None,
274 status: None,
275 storage_limit_bytes: None,
276 pending_storage_limit_bytes: None,
277 storage_used_bytes: None,
278 current_period_end: None,
279 },
280 };
281
282 Ok(Json(response))
283 }
284
285 /// Return the pricing-formula constants for an app. The client uses these to
286 /// quote a price locally as the user adjusts the cap slider; the same formula
287 /// is enforced server-side at checkout so the client number is only advisory.
288 #[utoipa::path(post, path = "/api/v1/sync/app/pricing", tag = "SyncKit",
289 request_body = AppPricingRequest,
290 responses((status = 200, description = "Pricing formula", body = AppPricingResponse)),
291 )]
292 #[tracing::instrument(skip_all, name = "synckit::get_app_pricing")]
293 pub(super) async fn get_app_pricing(
294 State(db): State<PgPool>,
295 Json(req): Json<AppPricingRequest>,
296 ) -> Result<impl IntoResponse> {
297 let app = db::synckit::get_sync_app_by_api_key(&db, &req.api_key)
298 .await?
299 .ok_or(AppError::NotFound)?;
300
301 Ok(Json(AppPricingResponse {
302 app_name: app.name,
303 min_charge_cents: payments::MIN_CHARGE_CENTS,
304 per_gb_tenths_of_cent_per_month:
305 payments::synckit_app_pricing::PER_GB_TENTHS_OF_CENT_PER_MONTH,
306 annual_multiplier: payments::ANNUAL_MULTIPLIER,
307 min_cap_bytes: payments::MIN_CAP_BYTES,
308 max_cap_bytes: payments::MAX_CAP_BYTES,
309 }))
310 }
311
312 /// Quote the price for a (cap, interval) pair. Authenticated so clients
313 /// cannot scrape pricing without an account, but otherwise pure: the result
314 /// only depends on the formula constants returned by `app/pricing`.
315 #[utoipa::path(post, path = "/api/v1/sync/subscription/quote", tag = "SyncKit",
316 request_body = SyncQuoteRequest,
317 responses((status = 200, description = "Quoted price", body = SyncQuoteResponse)),
318 security(("bearer" = [])),
319 )]
320 #[tracing::instrument(skip_all, name = "synckit::quote_subscription_price")]
321 pub(super) async fn quote_subscription_price(
322 _sync_user: SyncUser,
323 Json(req): Json<SyncQuoteRequest>,
324 ) -> Result<impl IntoResponse> {
325 let interval = SyncBillingInterval::parse(&req.interval)?;
326 let price_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
327 Ok(Json(SyncQuoteResponse {
328 cap_bytes: req.cap_bytes,
329 interval: interval.as_str().to_string(),
330 price_cents,
331 }))
332 }
333
334 /// Create a Stripe Checkout Session for subscribing this user to the app's
335 /// cloud sync at their chosen storage cap. The `app_sync_subscriptions` row
336 /// is written by the Stripe webhook on `checkout.session.completed`.
337 #[utoipa::path(post, path = "/api/v1/sync/subscription/checkout", tag = "SyncKit",
338 request_body = SyncSubscribeRequest,
339 responses((status = 200, description = "Checkout URL", body = SyncCheckoutResponse)),
340 security(("bearer" = [])),
341 )]
342 #[tracing::instrument(skip_all, name = "synckit::create_subscription_checkout")]
343 pub(super) async fn create_subscription_checkout(
344 State(db): State<PgPool>,
345 State(payments): State<crate::Billing>,
346 State(config): State<Config>,
347 sync_user: SyncUser,
348 Json(req): Json<SyncSubscribeRequest>,
349 ) -> Result<impl IntoResponse> {
350 if db::synckit::get_user_app_subscription(&db, sync_user.user_id, sync_user.app_id)
351 .await?
352 .is_some()
353 {
354 return Err(AppError::BadRequest(
355 "Already subscribed; use the storage-cap endpoint to adjust your cap".to_string(),
356 ));
357 }
358
359 let interval = SyncBillingInterval::parse(&req.interval)?;
360 let amount_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
361
362 let app = db::synckit::get_sync_app_by_id(&db, sync_user.app_id)
363 .await?
364 .ok_or(AppError::NotFound)?;
365 let cap_gib = req.cap_bytes / (1024 * 1024 * 1024);
366 let product_name = format!("{} cloud sync ({} GiB)", app.name, cap_gib);
367
368 let stripe = payments
369 .stripe
370 .as_ref()
371 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
372
373 let success_url = format!("{}/sync/subscribed", config.host_url);
374 let cancel_url = format!("{}/sync/canceled", config.host_url);
375
376 let result = stripe
377 .create_synckit_app_sub_checkout_session(&crate::payments::SynckitAppSubCheckoutParams {
378 product_name: &product_name,
379 amount_cents,
380 interval: interval.as_str(),
381 user_id: sync_user.user_id,
382 app_id: sync_user.app_id,
383 storage_limit_bytes: Some(req.cap_bytes),
384 success_url: &success_url,
385 cancel_url: &cancel_url,
386 })
387 .await?;
388
389 let checkout_url = result
390 .url
391 .ok_or_else(|| AppError::BadRequest("No checkout URL returned".to_string()))?;
392
393 Ok(Json(SyncCheckoutResponse { checkout_url }))
394 }
395
396 /// Queue a storage-cap change to take effect at the next billing cycle.
397 /// Holding the change for the cycle boundary avoids mid-cycle proration
398 /// surprises and keeps the user in control of when their bill changes.
399 ///
400 /// Updates Stripe first (re-price the subscription item with
401 /// `proration_behavior=None`), then records the pending cap in the DB. The
402 /// renewal webhook promotes the pending cap to active when Stripe rolls the
403 /// period, and because the Stripe price is already updated, the new invoice
404 /// is at the new price.
405 #[utoipa::path(post, path = "/api/v1/sync/subscription/storage-cap", tag = "SyncKit",
406 request_body = SyncCapChangeRequest,
407 responses((status = 200, description = "Pending cap recorded", body = SyncSubscriptionStatusResponse)),
408 security(("bearer" = [])),
409 )]
410 #[tracing::instrument(skip_all, name = "synckit::queue_storage_cap_change")]
411 pub(super) async fn queue_storage_cap_change(
412 State(db): State<PgPool>,
413 State(payments): State<crate::Billing>,
414 sync_user: SyncUser,
415 Json(req): Json<SyncCapChangeRequest>,
416 ) -> Result<impl IntoResponse> {
417 let sub = db::synckit::get_user_app_subscription(&db, sync_user.user_id, sync_user.app_id)
418 .await?
419 .ok_or_else(|| AppError::BadRequest("No active subscription to adjust".to_string()))?;
420
421 if req.cap_bytes < payments::MIN_CAP_BYTES || req.cap_bytes > payments::MAX_CAP_BYTES {
422 return Err(AppError::BadRequest(format!(
423 "Storage cap must be between {} and {} GiB",
424 payments::MIN_CAP_BYTES / (1024 * 1024 * 1024),
425 payments::MAX_CAP_BYTES / (1024 * 1024 * 1024)
426 )));
427 }
428
429 let interval = payments::SyncBillingInterval::parse(&sub.interval)?;
430 let new_price_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
431
432 let stripe = payments
433 .stripe
434 .as_ref()
435 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
436
437 let app = db::synckit::get_sync_app_by_id(&db, sync_user.app_id)
438 .await?
439 .ok_or(AppError::NotFound)?;
440 let cap_gib = req.cap_bytes / (1024 * 1024 * 1024);
441 let product_name = format!("{} cloud sync ({} GiB)", app.name, cap_gib);
442
443 // Stripe first: if this fails, we want the DB pending cap to stay
444 // unchanged so the user isn't sitting on an upgrade they never paid for.
445 stripe
446 .update_synckit_app_sub_price(
447 &sub.stripe_subscription_id,
448 new_price_cents,
449 interval,
450 &product_name,
451 )
452 .await?;
453
454 db::synckit::set_pending_storage_cap(&db, sync_user.user_id, sync_user.app_id, req.cap_bytes)
455 .await?;
456
457 Ok(Json(SyncSubscriptionStatusResponse {
458 active: sub.status == "active",
459 tier: Some(sub.interval),
460 status: Some(sub.status),
461 storage_limit_bytes: sub.storage_limit_bytes,
462 pending_storage_limit_bytes: Some(req.cap_bytes),
463 storage_used_bytes: None,
464 current_period_end: sub.current_period_end.map(|t| t.to_rfc3339()),
465 }))
466 }
467
468 // ── Device endpoints (JWT auth) ──
469
470 /// Register a new sync device (or update an existing one by name).
471 #[utoipa::path(post, path = "/api/v1/sync/devices", tag = "SyncKit",
472 request_body = RegisterDeviceRequest,
473 responses((status = 200, description = "Registered device", body = SyncDeviceResponse)),
474 security(("bearer" = [])),
475 )]
476 #[tracing::instrument(skip_all, name = "synckit::register_device")]
477 pub(super) async fn register_device(
478 State(db): State<PgPool>,
479 sync_user: SyncUser,
480 headers: axum::http::HeaderMap,
481 Json(req): Json<RegisterDeviceRequest>,
482 ) -> Result<impl IntoResponse> {
483 validation::validate_sync_device_name(&req.device_name)?;
484
485 // Serialize concurrent registrations for this (app, user) so the
486 // count-then-upsert cap check below is race-free. The lock is transaction
487 // scoped; holding `lock_tx` to the end of the handler keeps a second
488 // concurrent registration blocked until this one's upsert commits, so it
489 // sees the updated count and can't push past the cap (Run 21 concurrency).
490 let mut lock_tx = db.begin().await?;
491 sqlx::query(
492 "SELECT pg_advisory_xact_lock(hashtextextended('synckit_device:' || $1::text || ':' || $2::text, 0))",
493 )
494 .bind(sync_user.app_id)
495 .bind(sync_user.user_id)
496 .execute(&mut *lock_tx)
497 .await?;
498
499 // Enforce device limit (upsert on existing name is fine, only new names count)
500 let count = db::synckit::count_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
501 if count >= constants::SYNCKIT_MAX_DEVICES_PER_APP {
502 // Check if this is an existing device (upsert), allow updates
503 let existing =
504 db::synckit::get_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
505 if !existing.iter().any(|d| d.device_name == req.device_name) {
506 return Err(AppError::BadRequest(format!(
507 "Maximum {} devices per app",
508 constants::SYNCKIT_MAX_DEVICES_PER_APP
509 )));
510 }
511 }
512
513 let device = db::synckit::upsert_sync_device(
514 &db,
515 sync_user.app_id,
516 sync_user.user_id,
517 &req.device_name,
518 req.platform,
519 super::client_version(&headers).as_deref(),
520 )
521 .await?;
522
523 Ok(Json(SyncDeviceResponse {
524 id: device.id,
525 app_id: device.app_id,
526 user_id: device.user_id,
527 device_name: device.device_name,
528 platform: device.platform.to_string(),
529 last_seen_at: device.last_seen_at,
530 created_at: device.created_at,
531 }))
532 }
533
534 /// List all devices registered for the authenticated user and app.
535 #[utoipa::path(get, path = "/api/v1/sync/devices", tag = "SyncKit",
536 responses((status = 200, description = "List of devices", body = Vec<SyncDeviceResponse>)),
537 security(("bearer" = [])),
538 )]
539 #[tracing::instrument(skip_all, name = "synckit::list_devices")]
540 pub(super) async fn list_devices(
541 State(db): State<PgPool>,
542 sync_user: SyncUser,
543 ) -> Result<impl IntoResponse> {
544 let devices = db::synckit::get_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
545
546 let response: Vec<SyncDeviceResponse> = devices
547 .into_iter()
548 .map(|d| SyncDeviceResponse {
549 id: d.id,
550 app_id: d.app_id,
551 user_id: d.user_id,
552 device_name: d.device_name,
553 platform: d.platform.to_string(),
554 last_seen_at: d.last_seen_at,
555 created_at: d.created_at,
556 })
557 .collect();
558
559 Ok(Json(response))
560 }
561
562 /// Remove a registered device.
563 #[utoipa::path(delete, path = "/api/v1/sync/devices/{id}", tag = "SyncKit",
564 params(("id" = String, Path, description = "Device ID")),
565 responses((status = 204, description = "Device deleted"), (status = 404, description = "Device not found")),
566 security(("bearer" = [])),
567 )]
568 #[tracing::instrument(skip_all, name = "synckit::delete_device")]
569 pub(super) async fn delete_device(
570 State(db): State<PgPool>,
571 sync_user: SyncUser,
572 headers: axum::http::HeaderMap,
573 Path(device_id): Path<SyncDeviceId>,
574 ) -> Result<impl IntoResponse> {
575 let deleted =
576 db::synckit::delete_sync_device(&db, device_id, sync_user.app_id, sync_user.user_id)
577 .await?;
578
579 if !deleted {
580 return Err(AppError::NotFound);
581 }
582
583 // Security response to a device removal: invalidate the user's sync tokens
584 // so the removed device's JWT dies immediately (before this it survived to
585 // expiry), and record the event for the audit log. Best-effort, a logging
586 // or invalidation hiccup must not fail the delete the user asked for.
587 let ip = crate::helpers::extract_client_ip(&headers);
588 if let Err(e) = db::synckit::invalidate_user_sync_tokens(&db, sync_user.user_id).await {
589 tracing::error!(error = ?e, user_id = %sync_user.user_id,
590 "failed to invalidate sync tokens after device removal");
591 }
592 if let Err(e) = db::synckit::record_security_event(
593 &db,
594 sync_user.app_id,
595 Some(sync_user.user_id),
596 db::synckit::sync_security_event::DEVICE_REMOVED,
597 Some(serde_json::json!({ "device_id": device_id.to_string() })),
598 ip.as_deref(),
599 )
600 .await
601 {
602 tracing::error!(error = ?e, "failed to record device_removed security event");
603 }
604
605 Ok(axum::http::StatusCode::NO_CONTENT)
606 }
607
608 // ── Key management endpoints (JWT auth) ──
609
610 /// Store or update the user's encrypted master key envelope.
611 #[utoipa::path(put, path = "/api/v1/sync/keys", tag = "SyncKit",
612 request_body = PutKeyRequest,
613 responses((status = 204, description = "Key stored"), (status = 409, description = "Version mismatch")),
614 security(("bearer" = [])),
615 )]
616 #[tracing::instrument(skip_all, name = "synckit::put_sync_key")]
617 pub(super) async fn put_sync_key(
618 State(db): State<PgPool>,
619 sync_user: SyncUser,
620 Json(req): Json<PutKeyRequest>,
621 ) -> Result<impl IntoResponse> {
622 // Max 4 KB for the encrypted key envelope
623 if req.encrypted_key.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES {
624 return Err(AppError::BadRequest(
625 "Encrypted key exceeds 4KB limit".to_string(),
626 ));
627 }
628
629 let updated = db::synckit::upsert_sync_key(
630 &db,
631 sync_user.app_id,
632 sync_user.user_id,
633 &req.encrypted_key,
634 req.expected_version,
635 )
636 .await?;
637
638 if !updated {
639 return Err(AppError::Conflict(
640 "Key version mismatch, another device changed the password. Fetch the latest key and retry.".to_string(),
641 ));
642 }
643
644 Ok(axum::http::StatusCode::NO_CONTENT)
645 }
646
647 /// Retrieve the user's encrypted master key envelope.
648 #[utoipa::path(get, path = "/api/v1/sync/keys", tag = "SyncKit",
649 responses((status = 200, description = "Encrypted key envelope", body = GetKeyResponse), (status = 404, description = "No key stored")),
650 security(("bearer" = [])),
651 )]
652 #[tracing::instrument(skip_all, name = "synckit::get_sync_key")]
653 pub(super) async fn get_sync_key(
654 State(db): State<PgPool>,
655 sync_user: SyncUser,
656 ) -> Result<impl IntoResponse> {
657 let info = db::synckit::get_sync_key(&db, sync_user.app_id, sync_user.user_id)
658 .await?
659 .ok_or(AppError::NotFound)?;
660
661 let pending_key = info
662 .pending_key
663 .map(|(encrypted_key, key_id)| PendingKeyInfo {
664 encrypted_key,
665 key_id,
666 });
667
668 Ok(Json(GetKeyResponse {
669 encrypted_key: info.encrypted_key,
670 key_version: info.key_version,
671 key_id: info.key_id,
672 pending_key,
673 }))
674 }
675
676 // ── Key rotation endpoints ──
677
678 /// Begin a key rotation.
679 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate", tag = "SyncKit",
680 request_body = BeginRotationRequest,
681 responses(
682 (status = 200, description = "Rotation started or resumed", body = BeginRotationResponse),
683 (status = 409, description = "Version mismatch or rotation in progress"),
684 ),
685 security(("bearer" = [])),
686 )]
687 #[tracing::instrument(skip_all, name = "synckit::begin_rotation")]
688 pub(super) async fn begin_rotation(
689 State(db): State<PgPool>,
690 sync_user: SyncUser,
691 Json(req): Json<BeginRotationRequest>,
692 ) -> Result<impl IntoResponse> {
693 if req.new_encrypted_key.len() > constants::SYNCKIT_MAX_KEY_ENVELOPE_BYTES {
694 return Err(AppError::BadRequest(
695 "Encrypted key exceeds 4KB limit".to_string(),
696 ));
697 }
698
699 // Verify device belongs to this user + app via an indexed point lookup, not a
700 // fetch-all-then-scan (ultra-fuzz Run 4 Perf).
701 if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id)
702 .await?
703 {
704 return Err(AppError::BadRequest("Unknown device".to_string()));
705 }
706
707 let result = db::synckit::begin_key_rotation(
708 &db,
709 sync_user.app_id,
710 sync_user.user_id,
711 req.device_id,
712 &req.new_encrypted_key,
713 req.expected_key_version,
714 )
715 .await?;
716
717 match result {
718 Ok(rotation) => Ok(Json(BeginRotationResponse {
719 rotation_id: rotation.id,
720 target_seq: rotation.target_seq,
721 new_key_id: rotation.new_key_id,
722 })
723 .into_response()),
724 Err(msg) => Err(AppError::Conflict(msg.to_string())),
725 }
726 }
727
728 /// Pull entries that need re-encryption during a rotation.
729 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate/entries", tag = "SyncKit",
730 request_body = RotationEntriesRequest,
731 responses((status = 200, description = "Entries needing re-encryption", body = RotationEntriesResponse)),
732 security(("bearer" = [])),
733 )]
734 #[tracing::instrument(skip_all, name = "synckit::rotation_entries")]
735 pub(super) async fn rotation_entries(
736 State(db): State<PgPool>,
737 sync_user: SyncUser,
738 Json(req): Json<RotationEntriesRequest>,
739 ) -> Result<impl IntoResponse> {
740 let rotation = db::synckit::get_key_rotation(&db, sync_user.app_id, sync_user.user_id)
741 .await?
742 .ok_or_else(|| AppError::BadRequest("No active rotation".to_string()))?;
743
744 if rotation.id != req.rotation_id {
745 return Err(AppError::BadRequest("Rotation ID mismatch".to_string()));
746 }
747
748 let page_size = constants::SYNCKIT_ROTATION_BATCH_MAX as i64;
749 let raw_entries = db::synckit::get_rotation_entries(
750 &db,
751 sync_user.app_id,
752 sync_user.user_id,
753 rotation.new_key_id,
754 req.after_seq,
755 page_size,
756 )
757 .await?;
758
759 let has_more = raw_entries.len() as i64 == page_size;
760 let entries: Vec<RotationEntry> = raw_entries
761 .into_iter()
762 .map(|(seq, table, row_id, data)| RotationEntry {
763 seq,
764 table,
765 row_id,
766 data,
767 })
768 .collect();
769
770 Ok(Json(RotationEntriesResponse { entries, has_more }))
771 }
772
773 /// Submit a batch of re-encrypted entries during rotation.
774 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate/batch", tag = "SyncKit",
775 request_body = RotationBatchRequest,
776 responses((status = 200, description = "Batch processed", body = RotationBatchResponse)),
777 security(("bearer" = [])),
778 )]
779 #[tracing::instrument(skip_all, name = "synckit::rotation_batch")]
780 pub(super) async fn rotation_batch(
781 State(db): State<PgPool>,
782 sync_user: SyncUser,
783 Json(req): Json<RotationBatchRequest>,
784 ) -> Result<impl IntoResponse> {
785 if req.entries.is_empty() {
786 return Err(AppError::BadRequest("No entries provided".to_string()));
787 }
788 if req.entries.len() > constants::SYNCKIT_ROTATION_BATCH_MAX {
789 return Err(AppError::BadRequest(format!(
790 "Maximum {} entries per batch",
791 constants::SYNCKIT_ROTATION_BATCH_MAX
792 )));
793 }
794
795 let rotation = db::synckit::get_key_rotation(&db, sync_user.app_id, sync_user.user_id)
796 .await?
797 .ok_or_else(|| AppError::BadRequest("No active rotation".to_string()))?;
798
799 if rotation.id != req.rotation_id {
800 return Err(AppError::BadRequest("Rotation ID mismatch".to_string()));
801 }
802
803 let entries: Vec<(i64, Option<serde_json::Value>)> =
804 req.entries.into_iter().map(|e| (e.seq, e.data)).collect();
805
806 let updated_count = db::synckit::submit_rotation_batch(
807 &db,
808 sync_user.app_id,
809 sync_user.user_id,
810 rotation.id,
811 rotation.new_key_id,
812 &entries,
813 )
814 .await?;
815
816 Ok(Json(RotationBatchResponse { updated_count }))
817 }
818
819 /// Complete a key rotation.
820 #[utoipa::path(post, path = "/api/v1/sync/keys/rotate/complete", tag = "SyncKit",
821 request_body = CompleteRotationRequest,
822 responses(
823 (status = 204, description = "Rotation completed"),
824 (status = 409, description = "Entries still need re-encryption"),
825 ),
826 security(("bearer" = [])),
827 )]
828 #[tracing::instrument(skip_all, name = "synckit::complete_rotation")]
829 pub(super) async fn complete_rotation(
830 State(db): State<PgPool>,
831 sync_user: SyncUser,
832 Json(req): Json<CompleteRotationRequest>,
833 ) -> Result<impl IntoResponse> {
834 let result = db::synckit::complete_key_rotation(
835 &db,
836 sync_user.app_id,
837 sync_user.user_id,
838 req.rotation_id,
839 )
840 .await?;
841
842 match result {
843 Ok(_new_key_id) => {
844 // Audit the completed rotation (best-effort).
845 if let Err(e) = db::synckit::record_security_event(
846 &db,
847 sync_user.app_id,
848 Some(sync_user.user_id),
849 db::synckit::sync_security_event::KEY_ROTATION_COMPLETED,
850 Some(serde_json::json!({ "rotation_id": req.rotation_id.to_string() })),
851 None,
852 )
853 .await
854 {
855 tracing::error!(error = ?e, "failed to record key_rotation_completed security event");
856 }
857 Ok(axum::http::StatusCode::NO_CONTENT.into_response())
858 }
859 Err(0) => Err(AppError::BadRequest("No active rotation".to_string())),
860 Err(remaining) => Ok((
861 axum::http::StatusCode::CONFLICT,
862 Json(CompleteRotationErrorResponse { remaining }),
863 )
864 .into_response()),
865 }
866 }
867
868 /// Cancel a stale rotation (>24h without activity).
869 #[utoipa::path(delete, path = "/api/v1/sync/keys/rotate", tag = "SyncKit",
870 responses(
871 (status = 204, description = "Stale rotation cancelled"),
872 (status = 404, description = "No stale rotation found"),
873 ),
874 security(("bearer" = [])),
875 )]
876 #[tracing::instrument(skip_all, name = "synckit::cancel_rotation")]
877 pub(super) async fn cancel_rotation(
878 State(db): State<PgPool>,
879 sync_user: SyncUser,
880 ) -> Result<impl IntoResponse> {
881 let cancelled = db::synckit::cancel_stale_rotation(
882 &db,
883 sync_user.app_id,
884 sync_user.user_id,
885 constants::SYNCKIT_ROTATION_STALE_HOURS,
886 )
887 .await?;
888
889 if !cancelled {
890 return Err(AppError::NotFound);
891 }
892
893 Ok(axum::http::StatusCode::NO_CONTENT)
894 }
895