Skip to main content

max / makenotwork

Decompose AppState Phase 2: migrate synckit to slices Migrate the whole synckit route module (auth, keys, apps, billing, blobs, sync, subscribe) off State<AppState>. Per-handler slices: - auth: PgPool + Config - keys, apps: PgPool (apps helpers parse_and_verify_* take &PgPool) - billing: PgPool + Billing (stripe) + Config as needed; synckit_return_url helper takes &Config - blobs: PgPool + AppStorage (synckit_s3); download adds BackgroundTx (egress bump) - sync: mostly PgPool; checkout/cap-change add Billing (+Config); push adds Sync - subscribe: Sync only (SSE fan-out maps) quote_subscription_price dropped its unused State extractor. mod.rs keeps State<AppState> (router return type). No behavior change; 74 synckit + 77 sync integration tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 20:56 UTC
Commit: 31590b15b0fa24a25f76ee959b3def90d87ec246
Parent: 886154b
7 files changed, +185 insertions, -173 deletions
M server/src/routes/synckit/apps.rs +27 -26
@@ -6,12 +6,13 @@ use axum::{
6 6 Json,
7 7 };
8 8
9 + use sqlx::PgPool;
10 +
9 11 use crate::{
10 12 auth::AuthUser,
11 13 db::{self, SyncAppId},
12 14 error::{AppError, Result},
13 15 validation,
14 - AppState,
15 16 };
16 17
17 18 use super::UpdateAppSlugRequest;
@@ -24,19 +25,19 @@ use super::{CreateAppRequest, UpdateAppLinkRequest};
24 25 /// Returns the app data plus the plaintext API key (shown only once).
25 26 #[tracing::instrument(skip_all, name = "synckit::create_app")]
26 27 pub(super) async fn create_app(
27 - State(state): State<AppState>,
28 + State(db): State<PgPool>,
28 29 AuthUser(user): AuthUser,
29 30 Json(req): Json<CreateAppRequest>,
30 31 ) -> Result<impl IntoResponse> {
31 32 user.check_not_sandbox()?;
32 33 validation::validate_sync_app_name(&req.name)?;
33 34
34 - let project_id = parse_and_verify_project(&state, user.id, req.project_id.as_deref()).await?;
35 - let item_id = parse_and_verify_item(&state, user.id, req.item_id.as_deref()).await?;
35 + let project_id = parse_and_verify_project(&db, user.id, req.project_id.as_deref()).await?;
36 + let item_id = parse_and_verify_item(&db, user.id, req.item_id.as_deref()).await?;
36 37
37 38 let api_key = super::generate_api_key();
38 39 let app = db::synckit::create_sync_app(
39 - &state.db, user.id, &req.name, &api_key, project_id, item_id,
40 + &db, user.id, &req.name, &api_key, project_id, item_id,
40 41 ).await?;
41 42
42 43 Ok((axum::http::StatusCode::CREATED, Json(super::AppWithKey { app, api_key })))
@@ -47,10 +48,10 @@ pub(super) async fn create_app(
47 48 /// `GET /api/sync/apps`: Session auth required.
48 49 #[tracing::instrument(skip_all, name = "synckit::list_apps")]
49 50 pub(super) async fn list_apps(
50 - State(state): State<AppState>,
51 + State(db): State<PgPool>,
51 52 AuthUser(user): AuthUser,
52 53 ) -> Result<impl IntoResponse> {
53 - let apps = db::synckit::get_sync_apps_by_creator(&state.db, user.id).await?;
54 + let apps = db::synckit::get_sync_apps_by_creator(&db, user.id).await?;
54 55
55 56 Ok(Json(apps))
56 57 }
@@ -60,11 +61,11 @@ pub(super) async fn list_apps(
60 61 /// `POST /api/sync/apps/{id}/regenerate-key`: Session auth required.
61 62 #[tracing::instrument(skip_all, name = "synckit::regenerate_app_key")]
62 63 pub(super) async fn regenerate_app_key(
63 - State(state): State<AppState>,
64 + State(db): State<PgPool>,
64 65 AuthUser(user): AuthUser,
65 66 Path(app_id): Path<SyncAppId>,
66 67 ) -> Result<impl IntoResponse> {
67 - let app = db::synckit::get_sync_app_by_id(&state.db, app_id)
68 + let app = db::synckit::get_sync_app_by_id(&db, app_id)
68 69 .await?
69 70 .ok_or(AppError::NotFound)?;
70 71
@@ -73,7 +74,7 @@ pub(super) async fn regenerate_app_key(
73 74 }
74 75
75 76 let new_key = super::generate_api_key();
76 - let updated = db::synckit::regenerate_sync_app_key(&state.db, app_id, &new_key).await?;
77 + let updated = db::synckit::regenerate_sync_app_key(&db, app_id, &new_key).await?;
77 78
78 79 Ok(Json(super::AppWithKey { app: updated, api_key: new_key }))
79 80 }
@@ -83,11 +84,11 @@ pub(super) async fn regenerate_app_key(
83 84 /// `DELETE /api/sync/apps/{id}`: Session auth required.
84 85 #[tracing::instrument(skip_all, name = "synckit::delete_app")]
85 86 pub(super) async fn delete_app(
86 - State(state): State<AppState>,
87 + State(db): State<PgPool>,
87 88 AuthUser(user): AuthUser,
88 89 Path(app_id): Path<SyncAppId>,
89 90 ) -> Result<impl IntoResponse> {
90 - let app = db::synckit::get_sync_app_by_id(&state.db, app_id)
91 + let app = db::synckit::get_sync_app_by_id(&db, app_id)
91 92 .await?
92 93 .ok_or(AppError::NotFound)?;
93 94
@@ -95,7 +96,7 @@ pub(super) async fn delete_app(
95 96 return Err(AppError::Forbidden);
96 97 }
97 98
98 - db::synckit::delete_sync_app(&state.db, app_id).await?;
99 + db::synckit::delete_sync_app(&db, app_id).await?;
99 100
100 101 Ok(axum::http::StatusCode::NO_CONTENT)
101 102 }
@@ -105,12 +106,12 @@ pub(super) async fn delete_app(
105 106 /// `PUT /api/sync/apps/{id}/link`: Session auth required.
106 107 #[tracing::instrument(skip_all, name = "synckit::update_app_link")]
107 108 pub(super) async fn update_app_link(
108 - State(state): State<AppState>,
109 + State(db): State<PgPool>,
109 110 AuthUser(user): AuthUser,
110 111 Path(app_id): Path<SyncAppId>,
111 112 Json(req): Json<UpdateAppLinkRequest>,
112 113 ) -> Result<impl IntoResponse> {
113 - let app = db::synckit::get_sync_app_by_id(&state.db, app_id)
114 + let app = db::synckit::get_sync_app_by_id(&db, app_id)
114 115 .await?
115 116 .ok_or(AppError::NotFound)?;
116 117
@@ -118,11 +119,11 @@ pub(super) async fn update_app_link(
118 119 return Err(AppError::Forbidden);
119 120 }
120 121
121 - let project_id = parse_and_verify_project(&state, user.id, req.project_id.as_deref()).await?;
122 - let item_id = parse_and_verify_item(&state, user.id, req.item_id.as_deref()).await?;
122 + let project_id = parse_and_verify_project(&db, user.id, req.project_id.as_deref()).await?;
123 + let item_id = parse_and_verify_item(&db, user.id, req.item_id.as_deref()).await?;
123 124
124 125 let updated =
125 - db::synckit::update_sync_app_link(&state.db, app_id, project_id, item_id).await?;
126 + db::synckit::update_sync_app_link(&db, app_id, project_id, item_id).await?;
126 127
127 128 Ok(Json(updated))
128 129 }
@@ -132,12 +133,12 @@ pub(super) async fn update_app_link(
132 133 /// `PUT /api/sync/apps/{id}/slug`: Session auth required.
133 134 #[tracing::instrument(skip_all, name = "synckit::update_app_slug")]
134 135 pub(super) async fn update_app_slug(
135 - State(state): State<AppState>,
136 + State(db): State<PgPool>,
136 137 AuthUser(user): AuthUser,
137 138 Path(app_id): Path<SyncAppId>,
138 139 Json(req): Json<UpdateAppSlugRequest>,
139 140 ) -> Result<impl IntoResponse> {
140 - let app = db::synckit::get_sync_app_by_id(&state.db, app_id)
141 + let app = db::synckit::get_sync_app_by_id(&db, app_id)
141 142 .await?
142 143 .ok_or(AppError::NotFound)?;
143 144
@@ -148,7 +149,7 @@ pub(super) async fn update_app_slug(
148 149 // Reuse the OTA slug validation
149 150 crate::routes::ota::validate_slug_public(&req.slug)?;
150 151
151 - db::ota::set_app_slug(&state.db, app_id, &req.slug).await?;
152 + db::ota::set_app_slug(&db, app_id, &req.slug).await?;
152 153
153 154 Ok(axum::http::StatusCode::NO_CONTENT)
154 155 }
@@ -157,7 +158,7 @@ pub(super) async fn update_app_slug(
157 158
158 159 /// Parse an optional UUID string and verify the project belongs to the user.
159 160 async fn parse_and_verify_project(
160 - state: &AppState,
161 + db: &PgPool,
161 162 user_id: db::UserId,
162 163 raw: Option<&str>,
163 164 ) -> Result<Option<db::ProjectId>> {
@@ -167,7 +168,7 @@ async fn parse_and_verify_project(
167 168 let pid: db::ProjectId = s
168 169 .parse()
169 170 .map_err(|_| AppError::BadRequest("Invalid project_id".to_string()))?;
170 - let project = db::projects::get_project_by_id(&state.db, pid)
171 + let project = db::projects::get_project_by_id(db, pid)
171 172 .await?
172 173 .ok_or(AppError::BadRequest("Project not found".to_string()))?;
173 174 if project.user_id != user_id {
@@ -179,7 +180,7 @@ async fn parse_and_verify_project(
179 180 /// Parse an optional UUID string and verify the item belongs to the user
180 181 /// (via its parent project).
181 182 async fn parse_and_verify_item(
182 - state: &AppState,
183 + db: &PgPool,
183 184 user_id: db::UserId,
184 185 raw: Option<&str>,
185 186 ) -> Result<Option<db::ItemId>> {
@@ -189,10 +190,10 @@ async fn parse_and_verify_item(
189 190 let iid: db::ItemId = s
190 191 .parse()
191 192 .map_err(|_| AppError::BadRequest("Invalid item_id".to_string()))?;
192 - let item = db::items::get_item_by_id(&state.db, iid)
193 + let item = db::items::get_item_by_id(db, iid)
193 194 .await?
194 195 .ok_or(AppError::BadRequest("Item not found".to_string()))?;
195 - let project = db::projects::get_project_by_id(&state.db, item.project_id)
196 + let project = db::projects::get_project_by_id(db, item.project_id)
196 197 .await?
197 198 .ok_or(AppError::BadRequest("Item's project not found".to_string()))?;
198 199 if project.user_id != user_id {
M server/src/routes/synckit/auth.rs +14 -12
@@ -6,13 +6,15 @@ use axum::{
6 6 Json,
7 7 };
8 8
9 + use sqlx::PgPool;
10 +
9 11 use crate::{
10 12 auth::verify_password_async,
13 + config::Config,
11 14 db,
12 15 error::{AppError, Result},
13 16 synckit_auth,
14 17 validation,
15 - AppState,
16 18 };
17 19
18 20 /// Pre-computed dummy Argon2 hash used to equalize timing when a user is not found,
@@ -40,12 +42,12 @@ use super::{SyncAuthRequest, SyncAuthResponse, ValidateAppQuery, ValidateAppResp
40 42 )]
41 43 #[tracing::instrument(skip_all, name = "synckit::sync_auth")]
42 44 pub(super) async fn sync_auth(
43 - State(state): State<AppState>,
45 + State(db): State<PgPool>,
46 + State(config): State<Config>,
44 47 headers: axum::http::HeaderMap,
45 48 Json(req): Json<SyncAuthRequest>,
46 49 ) -> Result<impl IntoResponse> {
47 - let secret = state
48 - .config
50 + let secret = config
49 51 .synckit_jwt_secret
50 52 .as_deref()
51 53 .ok_or_else(|| AppError::ServiceUnavailable("SyncKit is not configured".to_string()))?;
@@ -53,7 +55,7 @@ pub(super) async fn sync_auth(
53 55 validation::validate_synckit_key(&req.key)?;
54 56
55 57 // Verify app exists and is active
56 - let app = db::synckit::get_sync_app_by_api_key(&state.db, &req.api_key)
58 + let app = db::synckit::get_sync_app_by_api_key(&db, &req.api_key)
57 59 .await?
58 60 .ok_or(AppError::Unauthorized)?;
59 61
@@ -75,7 +77,7 @@ pub(super) async fn sync_auth(
75 77 return Err(AppError::Unauthorized);
76 78 }
77 79 };
78 - let user = match db::users::get_user_by_email(&state.db, &email).await? {
80 + let user = match db::users::get_user_by_email(&db, &email).await? {
79 81 Some(u) => u,
80 82 None => {
81 83 // Equalize timing to prevent email enumeration
@@ -94,7 +96,7 @@ pub(super) async fn sync_auth(
94 96 // also avoids leaking 2FA status. (ultra-fuzz Run 3 SECURITY #4.) The folding
95 97 // and counter ordering live in the shared relying-party gate so this flow and
96 98 // OAuth can't drift apart (Run 11 Sec M1).
97 - match crate::auth::relying_party_login_gate(&state.db, &user, &req.password).await? {
99 + match crate::auth::relying_party_login_gate(&db, &user, &req.password).await? {
98 100 crate::auth::LoginGate::Deny { .. } => {
99 101 // Audit the failed sync auth (best-effort). Recorded only for a real
100 102 // account; the email-enumeration equalization paths above (unknown
@@ -102,7 +104,7 @@ pub(super) async fn sync_auth(
102 104 // and because they carry no user_id.
103 105 let ip = crate::helpers::extract_client_ip(&headers);
104 106 if let Err(e) = db::synckit::record_security_event(
105 - &state.db,
107 + &db,
106 108 app.id,
107 109 Some(user.id),
108 110 db::synckit::sync_security_event::AUTH_FAILURE,
@@ -127,12 +129,12 @@ pub(super) async fn sync_auth(
127 129 // internal and `bulk`/`app_wide` apps are uncapped and skip the claim.
128 130 // `claim_key` is idempotent for an already-claimed key and enforces the cap
129 131 // atomically under the usage-row lock.
130 - let billing = db::synckit_billing::get_app_with_billing(&state.db, app.id)
132 + let billing = db::synckit_billing::get_app_with_billing(&db, app.id)
131 133 .await?
132 134 .ok_or(AppError::Unauthorized)?;
133 135 if !billing.is_internal && billing.enforcement_mode == "per_key" {
134 136 let key_cap = billing.key_cap.unwrap_or(0);
135 - let claim = db::synckit_billing::claim_key(&state.db, app.id, &req.key, Some(key_cap)).await?;
137 + let claim = db::synckit_billing::claim_key(&db, app.id, &req.key, Some(key_cap)).await?;
136 138 if claim.cap_reached {
137 139 return Err(AppError::PaymentRequired(format!(
138 140 "key limit reached ({} of {key_cap} keys claimed); release an unused key or raise the cap",
@@ -165,10 +167,10 @@ pub(super) async fn sync_auth(
165 167 )]
166 168 #[tracing::instrument(skip_all, name = "synckit::validate_app")]
167 169 pub(super) async fn validate_app(
168 - State(state): State<AppState>,
170 + State(db): State<PgPool>,
169 171 Json(params): Json<ValidateAppQuery>,
170 172 ) -> Result<impl IntoResponse> {
171 - let app = db::synckit::get_sync_app_by_api_key(&state.db, &params.api_key)
173 + let app = db::synckit::get_sync_app_by_api_key(&db, &params.api_key)
172 174 .await?
173 175 .ok_or(AppError::Unauthorized)?;
174 176 Ok(Json(ValidateAppResponse {
M server/src/routes/synckit/billing.rs +37 -33
@@ -16,12 +16,14 @@ use axum::{
16 16 Json,
17 17 };
18 18
19 + use sqlx::PgPool;
20 +
19 21 use crate::{
20 22 auth::AuthUser,
23 + config::Config,
21 24 db::{self, SyncAppId},
22 25 error::{AppError, Result},
23 26 synckit_billing::monthly_price_cents,
24 - AppState,
25 27 };
26 28
27 29 use super::{
@@ -35,14 +37,16 @@ use super::{
35 37 /// `POST /api/sync/apps/{id}/billing/setup`
36 38 #[tracing::instrument(skip_all, name = "synckit::billing::setup")]
37 39 pub(super) async fn setup(
38 - State(state): State<AppState>,
40 + State(db): State<PgPool>,
41 + State(payments): State<crate::Billing>,
42 + State(config): State<Config>,
39 43 AuthUser(user): AuthUser,
40 44 Path(app_id): Path<SyncAppId>,
41 45 ) -> Result<impl IntoResponse> {
42 46 user.check_not_sandbox()?;
43 47 user.check_not_suspended()?;
44 48
45 - let app = db::synckit_billing::get_app_with_billing(&state.db, app_id)
49 + let app = db::synckit_billing::get_app_with_billing(&db, app_id)
46 50 .await?
47 51 .ok_or(AppError::NotFound)?;
48 52 if app.creator_id != user.id {
@@ -55,8 +59,7 @@ pub(super) async fn setup(
55 59 )));
56 60 }
57 61
58 - let stripe = state
59 - .stripe
62 + let stripe = payments.stripe
60 63 .as_ref()
61 64 .ok_or_else(|| AppError::ServiceUnavailable("Stripe is not configured".to_string()))?;
62 65
@@ -64,18 +67,18 @@ pub(super) async fn setup(
64 67 let customer_id = match app.stripe_customer_id.as_deref() {
65 68 Some(id) => id.to_string(),
66 69 None => {
67 - let developer = db::users::get_user_by_id(&state.db, user.id)
70 + let developer = db::users::get_user_by_id(&db, user.id)
68 71 .await?
69 72 .ok_or(AppError::Unauthorized)?;
70 73 let id = stripe
71 74 .create_synckit_customer(user.id, app_id, developer.email.as_str(), &app.name)
72 75 .await?;
73 - db::synckit_billing::set_stripe_customer(&state.db, app_id, &id).await?;
76 + db::synckit_billing::set_stripe_customer(&db, app_id, &id).await?;
74 77 id
75 78 }
76 79 };
77 80
78 - let return_url = synckit_return_url(&state, &app);
81 + let return_url = synckit_return_url(&config, &app);
79 82 let portal_url = stripe
80 83 .create_synckit_billing_portal(&customer_id, &return_url)
81 84 .await?;
@@ -92,7 +95,8 @@ pub(super) async fn setup(
92 95 /// `POST /api/sync/apps/{id}/billing/activate`
93 96 #[tracing::instrument(skip_all, name = "synckit::billing::activate")]
94 97 pub(super) async fn activate(
95 - State(state): State<AppState>,
98 + State(db): State<PgPool>,
99 + State(payments): State<crate::Billing>,
96 100 AuthUser(user): AuthUser,
97 101 Path(app_id): Path<SyncAppId>,
98 102 Json(req): Json<BillingActivateRequest>,
@@ -101,7 +105,7 @@ pub(super) async fn activate(
101 105 user.check_not_suspended()?;
102 106 let mode = validate_knobs(&req.enforcement_mode, req.storage_gb_cap, req.key_cap, req.gb_per_key)?;
103 107
104 - let app = db::synckit_billing::get_app_with_billing(&state.db, app_id)
108 + let app = db::synckit_billing::get_app_with_billing(&db, app_id)
105 109 .await?
106 110 .ok_or(AppError::NotFound)?;
107 111 if app.creator_id != user.id {
@@ -126,8 +130,7 @@ pub(super) async fn activate(
126 130 req.gb_per_key,
127 131 );
128 132
129 - let stripe = state
130 - .stripe
133 + let stripe = payments.stripe
131 134 .as_ref()
132 135 .ok_or_else(|| AppError::ServiceUnavailable("Stripe is not configured".to_string()))?;
133 136 let sub = stripe
@@ -140,7 +143,7 @@ pub(super) async fn activate(
140 143 .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Invalid period_end from Stripe")))?;
141 144
142 145 db::synckit_billing::activate_billing(
143 - &state.db,
146 + &db,
144 147 app_id,
145 148 mode,
146 149 req.storage_gb_cap.map(|v| v as i32),
@@ -164,7 +167,8 @@ pub(super) async fn activate(
164 167 /// `PATCH /api/sync/apps/{id}/billing`
165 168 #[tracing::instrument(skip_all, name = "synckit::billing::patch")]
166 169 pub(super) async fn patch(
167 - State(state): State<AppState>,
170 + State(db): State<PgPool>,
171 + State(payments): State<crate::Billing>,
168 172 AuthUser(user): AuthUser,
169 173 Path(app_id): Path<SyncAppId>,
170 174 Json(req): Json<BillingPatchRequest>,
@@ -173,7 +177,7 @@ pub(super) async fn patch(
173 177 user.check_not_suspended()?;
174 178 let mode = validate_knobs(&req.enforcement_mode, req.storage_gb_cap, req.key_cap, req.gb_per_key)?;
175 179
176 - let app = db::synckit_billing::get_app_with_billing(&state.db, app_id)
180 + let app = db::synckit_billing::get_app_with_billing(&db, app_id)
177 181 .await?
178 182 .ok_or(AppError::NotFound)?;
179 183 if app.creator_id != user.id {
@@ -198,8 +202,7 @@ pub(super) async fn patch(
198 202 req.gb_per_key,
199 203 );
200 204
201 - let stripe = state
202 - .stripe
205 + let stripe = payments.stripe
203 206 .as_ref()
204 207 .ok_or_else(|| AppError::ServiceUnavailable("Stripe is not configured".to_string()))?;
205 208 stripe
@@ -207,7 +210,7 @@ pub(super) async fn patch(
207 210 .await?;
208 211
209 212 db::synckit_billing::update_knobs(
210 - &state.db,
213 + &db,
211 214 app_id,
212 215 mode,
213 216 req.storage_gb_cap.map(|v| v as i32),
@@ -228,13 +231,14 @@ pub(super) async fn patch(
228 231 /// `DELETE /api/sync/apps/{id}/billing`
229 232 #[tracing::instrument(skip_all, name = "synckit::billing::cancel")]
230 233 pub(super) async fn cancel(
231 - State(state): State<AppState>,
234 + State(db): State<PgPool>,
235 + State(payments): State<crate::Billing>,
232 236 AuthUser(user): AuthUser,
233 237 Path(app_id): Path<SyncAppId>,
234 238 ) -> Result<impl IntoResponse> {
235 239 user.check_not_sandbox()?;
236 240
237 - let app = db::synckit_billing::get_app_with_billing(&state.db, app_id)
241 + let app = db::synckit_billing::get_app_with_billing(&db, app_id)
238 242 .await?
239 243 .ok_or(AppError::NotFound)?;
240 244 if app.creator_id != user.id {
@@ -245,14 +249,13 @@ pub(super) async fn cancel(
245 249 }
246 250
247 251 if let Some(sub_id) = app.stripe_subscription_id.as_deref() {
248 - let stripe = state
249 - .stripe
252 + let stripe = payments.stripe
250 253 .as_ref()
251 254 .ok_or_else(|| AppError::ServiceUnavailable("Stripe is not configured".to_string()))?;
252 255 stripe.cancel_synckit_subscription(sub_id).await?;
253 256 }
254 257
255 - db::synckit_billing::apply_billing_update(&state.db, app_id, Some("canceled"), None).await?;
258 + db::synckit_billing::apply_billing_update(&db, app_id, Some("canceled"), None).await?;
256 259
257 260 Ok(axum::http::StatusCode::NO_CONTENT)
258 261 }
@@ -262,11 +265,11 @@ pub(super) async fn cancel(
262 265 /// `GET /api/sync/apps/{id}/billing`
263 266 #[tracing::instrument(skip_all, name = "synckit::billing::get")]
264 267 pub(super) async fn get(
265 - State(state): State<AppState>,
268 + State(db): State<PgPool>,
266 269 AuthUser(user): AuthUser,
267 270 Path(app_id): Path<SyncAppId>,
268 271 ) -> Result<impl IntoResponse> {
269 - let app = db::synckit_billing::get_app_with_billing(&state.db, app_id)
272 + let app = db::synckit_billing::get_app_with_billing(&db, app_id)
270 273 .await?
271 274 .ok_or(AppError::NotFound)?;
272 275 if app.creator_id != user.id {
@@ -311,13 +314,15 @@ pub(super) async fn get(
311 314 /// `GET /api/sync/apps/{id}/billing/portal`
312 315 #[tracing::instrument(skip_all, name = "synckit::billing::portal")]
313 316 pub(super) async fn portal(
314 - State(state): State<AppState>,
317 + State(db): State<PgPool>,
318 + State(payments): State<crate::Billing>,
319 + State(config): State<Config>,
315 320 AuthUser(user): AuthUser,
316 321 Path(app_id): Path<SyncAppId>,
317 322 ) -> Result<impl IntoResponse> {
318 323 user.check_not_sandbox()?;
319 324
320 - let app = db::synckit_billing::get_app_with_billing(&state.db, app_id)
325 + let app = db::synckit_billing::get_app_with_billing(&db, app_id)
321 326 .await?
322 327 .ok_or(AppError::NotFound)?;
323 328 if app.creator_id != user.id {
@@ -330,12 +335,11 @@ pub(super) async fn portal(
330 335 )
331 336 })?;
332 337
333 - let stripe = state
334 - .stripe
338 + let stripe = payments.stripe
335 339 .as_ref()
336 340 .ok_or_else(|| AppError::ServiceUnavailable("Stripe is not configured".to_string()))?;
337 341
338 - let return_url = synckit_return_url(&state, &app);
342 + let return_url = synckit_return_url(&config, &app);
339 343 let portal_url = stripe
340 344 .create_synckit_billing_portal(customer_id, &return_url)
341 345 .await?;
@@ -349,15 +353,15 @@ pub(super) async fn portal(
349 353 /// developer back to the SyncKit tab on the project dashboard when the app is
350 354 /// linked to a project, or the user-dashboard SyncKit tab otherwise.
351 355 fn synckit_return_url(
352 - state: &AppState,
356 + config: &Config,
353 357 app: &crate::db::DbSyncAppBilling,
354 358 ) -> String {
355 359 match app.project_slug.as_deref() {
356 360 Some(slug) => format!(
357 361 "{}/dashboard/project/{}#tab-synckit",
358 - state.config.host_url, slug
362 + config.host_url, slug
359 363 ),
360 - None => format!("{}/dashboard#tab-synckit", state.config.host_url),
364 + None => format!("{}/dashboard#tab-synckit", config.host_url),
361 365 }
362 366 }
363 367
M server/src/routes/synckit/blobs.rs +25 -23
@@ -8,13 +8,14 @@ use axum::{
8 8 };
9 9 use serde_json::json;
10 10
11 + use sqlx::PgPool;
12 +
11 13 use crate::{
12 14 constants,
13 15 db::{self, synckit_billing},
14 16 error::{AppError, Result, ResultExt},
15 17 synckit_auth::SyncUser,
16 18 validation,
17 - AppState,
18 19 };
19 20
20 21 use super::{
@@ -34,12 +35,12 @@ use super::{
34 35 )]
35 36 #[tracing::instrument(skip_all, name = "synckit::blob_upload_url")]
36 37 pub(super) async fn blob_upload_url(
37 - State(state): State<AppState>,
38 + State(db): State<PgPool>,
39 + State(storage): State<crate::AppStorage>,
38 40 sync_user: SyncUser,
39 41 Json(req): Json<BlobUploadUrlRequest>,
40 42 ) -> Result<Response> {
41 - let synckit_s3 = state
42 - .storage.synckit_s3
43 + let synckit_s3 = storage.synckit_s3
43 44 .as_ref()
44 45 .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
45 46
@@ -55,7 +56,7 @@ pub(super) async fn blob_upload_url(
55 56 // an unsubscribed user, so they can't stage orphan S3 objects. Non-internal
56 57 // (developer-billed) apps are always allowed here. Storage-cap enforcement
57 58 // still happens atomically at confirm time.
58 - if !db::synckit::internal_write_allowed(&state.db, sync_user.app_id, sync_user.user_id).await? {
59 + if !db::synckit::internal_write_allowed(&db, sync_user.app_id, sync_user.user_id).await? {
59 60 return Ok((
60 61 StatusCode::PAYMENT_REQUIRED,
61 62 Json(json!({ "reason": "no_subscription" })),
@@ -65,7 +66,7 @@ pub(super) async fn blob_upload_url(
65 66
66 67 // Check dedup — if this hash already exists, skip upload
67 68 if let Some(_existing) = db::synckit::get_sync_blob_by_hash(
68 - &state.db,
69 + &db,
69 70 sync_user.app_id,
70 71 sync_user.user_id,
71 72 &req.hash,
@@ -84,7 +85,7 @@ pub(super) async fn blob_upload_url(
84 85 );
85 86
86 87 // Track the pending upload so the reaper can clean it up if never confirmed
87 - db::pending_uploads::record_pending_upload(&state.db, sync_user.user_id, &s3_key, "synckit").await?;
88 + db::pending_uploads::record_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
88 89
89 90 let upload_url = synckit_s3
90 91 .presign_upload(
@@ -131,18 +132,18 @@ pub(super) async fn blob_upload_url(
131 132 )]
132 133 #[tracing::instrument(skip_all, name = "synckit::blob_confirm_upload")]
133 134 pub(super) async fn blob_confirm_upload(
134 - State(state): State<AppState>,
135 + State(db): State<PgPool>,
136 + State(storage): State<crate::AppStorage>,
135 137 sync_user: SyncUser,
136 138 Json(req): Json<BlobConfirmRequest>,
137 139 ) -> Result<Response> {
138 - let synckit_s3 = state
139 - .storage.synckit_s3
140 + let synckit_s3 = storage.synckit_s3
140 141 .as_ref()
141 142 .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
142 143
143 144 validation::validate_sync_blob_hash(&req.hash)?;
144 145
145 - let billing = synckit_billing::get_app_with_billing(&state.db, sync_user.app_id)
146 + let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
146 147 .await?
147 148 .ok_or(AppError::NotFound)?;
148 149
@@ -165,14 +166,14 @@ pub(super) async fn blob_confirm_upload(
165 166 }
166 167
167 168 // Clear the pending upload record now that the upload is confirmed.
168 - db::pending_uploads::remove_pending_upload(&state.db, sync_user.user_id, &s3_key, "synckit").await?;
169 + db::pending_uploads::remove_pending_upload(&db, sync_user.user_id, &s3_key, "synckit").await?;
169 170
170 171 // Record the blob and enforce the cap atomically. Internal apps gate on the
171 172 // user's paid subscription + per-user cap; developer apps on the app/per-key
172 173 // counters. Both are single-transaction (lock, check, insert, count).
173 174 let outcome = if billing.is_internal {
174 175 db::synckit::confirm_internal_blob(
175 - &state.db, sync_user.app_id, sync_user.user_id,
176 + &db, sync_user.app_id, sync_user.user_id,
176 177 &req.hash, actual_size, &s3_key, &sync_user.key,
177 178 )
178 179 .await?
@@ -185,7 +186,7 @@ pub(super) async fn blob_confirm_upload(
185 186 .into_response());
186 187 }
187 188 db::synckit::confirm_developer_blob(
188 - &state.db, sync_user.app_id, sync_user.user_id,
189 + &db, sync_user.app_id, sync_user.user_id,
189 190 &req.hash, actual_size, &s3_key, &sync_user.key,
190 191 billing.enforcement_mode, billing.storage_gb_cap, billing.key_cap, billing.gb_per_key,
191 192 )
@@ -224,19 +225,20 @@ pub(super) async fn blob_confirm_upload(
224 225 )]
225 226 #[tracing::instrument(skip_all, name = "synckit::blob_download_url")]
226 227 pub(super) async fn blob_download_url(
227 - State(state): State<AppState>,
228 + State(db): State<PgPool>,
229 + State(storage): State<crate::AppStorage>,
230 + State(bg): State<crate::background::BackgroundTx>,
228 231 sync_user: SyncUser,
229 232 Json(req): Json<BlobDownloadUrlRequest>,
230 233 ) -> Result<Response> {
231 - let synckit_s3 = state
232 - .storage.synckit_s3
234 + let synckit_s3 = storage.synckit_s3
233 235 .as_ref()
234 236 .ok_or_else(|| AppError::ServiceUnavailable("SyncKit blob storage is not configured".to_string()))?;
235 237
236 238 validation::validate_sync_blob_hash(&req.hash)?;
237 239
238 240 let blob = db::synckit::get_sync_blob_by_hash(
239 - &state.db,
241 + &db,
240 242 sync_user.app_id,
241 243 sync_user.user_id,
242 244 &req.hash,
@@ -247,7 +249,7 @@ pub(super) async fn blob_download_url(
247 249 // Billing check (internal apps bypass). Egress is NOT enforced — it's a
248 250 // free metric for the developer's dashboard, absorbed in the storage rate
249 251 // margin. We still count it at presign time so devs see the stat.
250 - let billing = synckit_billing::get_app_with_billing(&state.db, sync_user.app_id)
252 + let billing = synckit_billing::get_app_with_billing(&db, sync_user.app_id)
251 253 .await?
252 254 .ok_or(AppError::NotFound)?;
253 255 if !billing.is_internal {
@@ -264,10 +266,10 @@ pub(super) async fn blob_download_url(
264 266 // dashboard metric. Deferred onto the bounded background pool: it's a
265 267 // single hot-row UPDATE that the download path shouldn't wait on or
266 268 // contend its lock against (Perf P4).
267 - let db = state.db.clone();
269 + let db = db.clone();
268 270 let app_id = sync_user.app_id;
269 271 let egress_bytes = blob.size_bytes;
270 - state.bg.spawn("synckit-egress-bump", async move {
272 + bg.spawn("synckit-egress-bump", async move {
271 273 if let Err(e) = synckit_billing::add_bytes_egress(&db, app_id, egress_bytes).await {
272 274 tracing::error!(error = ?e, app_id = %app_id, "failed to bump bytes_egress_period");
273 275 }
@@ -300,11 +302,11 @@ pub(super) async fn blob_download_url(
300 302 )]
301 303 #[tracing::instrument(skip_all, name = "synckit::blob_delete")]
302 304 pub(super) async fn blob_delete(
303 - State(state): State<AppState>,
305 + State(db): State<PgPool>,
304 306 sync_user: SyncUser,
305 307 axum::extract::Path(hash): axum::extract::Path<String>,
306 308 ) -> Result<Response> {
307 309 validation::validate_sync_blob_hash(&hash)?;
308 - db::synckit::delete_sync_blob(&state.db, sync_user.app_id, sync_user.user_id, &hash).await?;
310 + db::synckit::delete_sync_blob(&db, sync_user.app_id, sync_user.user_id, &hash).await?;
309 311 Ok(StatusCode::NO_CONTENT.into_response())
310 312 }
M server/src/routes/synckit/keys.rs +12 -11
@@ -18,10 +18,11 @@ use axum::{
18 18 };
19 19 use serde_json::json;
20 20
21 + use sqlx::PgPool;
22 +
21 23 use crate::{
22 24 db::{self, synckit_billing},
23 25 error::{AppError, Result},
24 - AppState,
25 26 };
26 27
27 28 use super::{
@@ -40,14 +41,14 @@ use super::{
40 41 /// always idempotent OK).
41 42 #[tracing::instrument(skip_all, name = "synckit::keys::claim")]
42 43 pub(super) async fn claim(
43 - State(state): State<AppState>,
44 + State(db): State<PgPool>,
44 45 Json(req): Json<ClaimKeyRequest>,
45 46 ) -> Result<axum::response::Response> {
46 - let app = db::synckit::get_sync_app_by_api_key(&state.db, &req.api_key)
47 + let app = db::synckit::get_sync_app_by_api_key(&db, &req.api_key)
47 48 .await?
48 49 .ok_or(AppError::Unauthorized)?;
49 50
50 - let billing = synckit_billing::get_app_with_billing(&state.db, app.id)
51 + let billing = synckit_billing::get_app_with_billing(&db, app.id)
51 52 .await?
52 53 .ok_or(AppError::NotFound)?;
53 54
@@ -71,7 +72,7 @@ pub(super) async fn claim(
71 72 None
72 73 };
73 74
74 - let result = synckit_billing::claim_key(&state.db, app.id, &req.key, key_cap).await?;
75 + let result = synckit_billing::claim_key(&db, app.id, &req.key, key_cap).await?;
75 76 if result.cap_reached {
76 77 return Ok((
77 78 StatusCode::PAYMENT_REQUIRED,
@@ -96,14 +97,14 @@ pub(super) async fn claim(
96 97 /// cleanup paths can drain stale claims.
97 98 #[tracing::instrument(skip_all, name = "synckit::keys::release")]
98 99 pub(super) async fn release(
99 - State(state): State<AppState>,
100 + State(db): State<PgPool>,
100 101 Json(req): Json<ReleaseKeyRequest>,
101 102 ) -> Result<impl IntoResponse> {
102 - let app = db::synckit::get_sync_app_by_api_key(&state.db, &req.api_key)
103 + let app = db::synckit::get_sync_app_by_api_key(&db, &req.api_key)
103 104 .await?
104 105 .ok_or(AppError::Unauthorized)?;
105 106
106 - let result = synckit_billing::release_key(&state.db, app.id, &req.key).await?;
107 + let result = synckit_billing::release_key(&db, app.id, &req.key).await?;
107 108 Ok(Json(ReleaseKeyResponse {
108 109 newly_released: result.newly_released,
109 110 total_claimed: result.total_claimed,
@@ -116,10 +117,10 @@ pub(super) async fn release(
116 117 /// keeping the api_key out of access logs.
117 118 #[tracing::instrument(skip_all, name = "synckit::keys::list")]
118 119 pub(super) async fn list(
119 - State(state): State<AppState>,
120 + State(db): State<PgPool>,
120 121 Json(req): Json<ListKeysRequest>,
121 122 ) -> Result<impl IntoResponse> {
122 - let app = db::synckit::get_sync_app_by_api_key(&state.db, &req.api_key)
123 + let app = db::synckit::get_sync_app_by_api_key(&db, &req.api_key)
123 124 .await?
124 125 .ok_or(AppError::Unauthorized)?;
125 126
@@ -128,7 +129,7 @@ pub(super) async fn list(
128 129 // deep-scan, the same DoS-shaped cost the limit clamp guards against.
129 130 let offset = (req.offset.unwrap_or(0) as i64).clamp(0, 1_000_000_000);
130 131
131 - let rows = synckit_billing::list_active_keys(&state.db, app.id, limit, offset).await?;
132 + let rows = synckit_billing::list_active_keys(&db, app.id, limit, offset).await?;
132 133
133 134 let keys = rows
134 135 .into_iter()
M server/src/routes/synckit/subscribe.rs +5 -6
@@ -22,7 +22,6 @@ use crate::{
22 22 db::{SyncAppId, UserId},
23 23 error::{AppError, Result},
24 24 synckit_auth::SyncUser,
25 - AppState,
26 25 };
27 26
28 27 /// Drop guard that decrements the per-user SSE connection counter.
@@ -92,7 +91,7 @@ pub struct SubscribeQuery {
92 91 /// A keepalive comment is sent every 30 seconds to prevent proxy timeouts.
93 92 #[tracing::instrument(skip_all, name = "synckit::subscribe")]
94 93 pub(super) async fn sync_subscribe(
95 - State(state): State<AppState>,
94 + State(sync): State<crate::Sync>,
96 95 sync_user: SyncUser,
97 96 Query(query): Query<SubscribeQuery>,
98 97 ) -> Result<Sse<impl tokio_stream::Stream<Item = std::result::Result<Event, Infallible>>>> {
@@ -104,7 +103,7 @@ pub(super) async fn sync_subscribe(
104 103 }
105 104
106 105 // Enforce per-user SSE connection limit
107 - let counter = state.caches.sse_connections
106 + let counter = sync.sse_connections
108 107 .entry(sync_user.user_id)
109 108 .or_insert_with(|| std::sync::atomic::AtomicUsize::new(0));
110 109 let current = counter.value().fetch_add(1, Ordering::AcqRel);
@@ -118,15 +117,15 @@ pub(super) async fn sync_subscribe(
118 117 let key = (sync_user.app_id, sync_user.user_id);
119 118
120 119 let guard = SseConnectionGuard {
121 - sse_connections: state.caches.sse_connections.clone(),
122 - sync_notify: state.caches.sync_notify.clone(),
120 + sse_connections: sync.sse_connections.clone(),
121 + sync_notify: sync.sync_notify.clone(),
123 122 user_id: sync_user.user_id,
124 123 app_id: sync_user.app_id,
125 124 };
126 125
127 126 // Get or create the broadcast channel for this app+user
128 127 let rx = {
129 - let entry = state.caches.sync_notify.entry(key).or_insert_with(|| {
128 + let entry = sync.sync_notify.entry(key).or_insert_with(|| {
130 129 let (tx, _) = tokio::sync::broadcast::channel(16);
131 130 tx
132 131 });
M server/src/routes/synckit/sync.rs +65 -62
@@ -8,14 +8,16 @@ use axum::{
8 8 };
9 9 use serde_json::json;
10 10
11 + use sqlx::PgPool;
12 +
11 13 use crate::{
14 + config::Config,
12 15 constants,
13 16 db::{self, SyncDeviceId},
14 17 error::{AppError, Result},
15 18 payments::{self, SyncBillingInterval},
16 19 synckit_auth::SyncUser,
17 20 validation,
18 - AppState,
19 21 };
20 22
21 23 use super::{
@@ -38,7 +40,8 @@ use super::{
38 40 )]
39 41 #[tracing::instrument(skip_all, name = "synckit::sync_push", fields(app_id, user_id))]
40 42 pub(super) async fn sync_push(
41 - State(state): State<AppState>,
43 + State(db): State<PgPool>,
44 + State(sync): State<crate::Sync>,
42 45 sync_user: SyncUser,
43 46 Json(req): Json<PushRequest>,
44 47 ) -> Result<Response> {
@@ -50,7 +53,7 @@ pub(super) async fn sync_push(
50 53 // Paid-only sync: first-party apps require an active end-user subscription
51 54 // to write. Reads (pull) stay open so a lapsed user can still export their
52 55 // data. Non-internal (developer-billed) apps are always allowed here.
53 - if !db::synckit::internal_write_allowed(&state.db, app_id, user_id).await? {
56 + if !db::synckit::internal_write_allowed(&db, app_id, user_id).await? {
54 57 return Ok((
55 58 StatusCode::PAYMENT_REQUIRED,
56 59 Json(json!({ "reason": "no_subscription" })),
@@ -81,12 +84,12 @@ pub(super) async fn sync_push(
81 84
82 85 // Verify device belongs to this user + app (indexed point lookup, not a
83 86 // full device fetch + linear scan).
84 - if !db::synckit::sync_device_belongs(&state.db, req.device_id, app_id, user_id).await? {
87 + if !db::synckit::sync_device_belongs(&db, req.device_id, app_id, user_id).await? {
85 88 return Err(AppError::BadRequest("Unknown device".to_string()));
86 89 }
87 90
88 91 // Touch device
89 - db::synckit::touch_sync_device(&state.db, req.device_id).await?;
92 + db::synckit::touch_sync_device(&db, req.device_id).await?;
90 93
91 94 // Build change tuples (op converted to string for TEXT column)
92 95 let changes: Vec<_> = req
@@ -104,7 +107,7 @@ pub(super) async fn sync_push(
104 107 .collect();
105 108
106 109 let cursor = db::synckit::push_sync_changes(
107 - &state.db,
110 + &db,
108 111 app_id,
109 112 user_id,
110 113 req.device_id,
@@ -116,7 +119,7 @@ pub(super) async fn sync_push(
116 119 // Notify SSE subscribers, carrying the new max seq so a device already at
117 120 // (or past) this cursor can skip a redundant pull — avoids the thundering
118 121 // herd where every subscriber pulls on every push.
119 - if let Some(sender) = state.caches.sync_notify.get(&(app_id, user_id)) {
122 + if let Some(sender) = sync.sync_notify.get(&(app_id, user_id)) {
120 123 let _ = sender.send(cursor); // Ignore errors (no subscribers = ok)
121 124 }
122 125
@@ -131,7 +134,7 @@ pub(super) async fn sync_push(
131 134 )]
132 135 #[tracing::instrument(skip_all, name = "synckit::sync_pull", fields(app_id, user_id))]
133 136 pub(super) async fn sync_pull(
134 - State(state): State<AppState>,
137 + State(db): State<PgPool>,
135 138 sync_user: SyncUser,
136 139 Json(req): Json<PullRequest>,
137 140 ) -> Result<impl IntoResponse> {
@@ -142,7 +145,7 @@ pub(super) async fn sync_pull(
142 145
143 146 // Verify device belongs to this user + app (indexed point lookup, not a
144 147 // full device fetch + linear scan).
145 - if !db::synckit::sync_device_belongs(&state.db, req.device_id, app_id, user_id).await? {
148 + if !db::synckit::sync_device_belongs(&db, req.device_id, app_id, user_id).await? {
146 149 return Err(AppError::BadRequest("Unknown device".to_string()));
147 150 }
148 151
@@ -160,7 +163,7 @@ pub(super) async fn sync_pull(
160 163
161 164 let page_size = constants::SYNCKIT_PULL_PAGE_SIZE;
162 165 let entries = db::synckit::pull_sync_changes_filtered(
163 - &state.db,
166 + &db,
164 167 app_id,
165 168 user_id,
166 169 req.cursor,
@@ -175,7 +178,7 @@ pub(super) async fn sync_pull(
175 178
176 179 // Mark the device seen and advance its compaction cursor in one statement.
177 180 // GREATEST keeps the cursor monotonic even if `new_cursor == req.cursor`.
178 - db::synckit::touch_and_advance_cursor(&state.db, req.device_id, new_cursor).await?;
181 + db::synckit::touch_and_advance_cursor(&db, req.device_id, new_cursor).await?;
179 182
180 183 let changes: Vec<PullChangeEntry> = entries
181 184 .into_iter()
@@ -205,14 +208,14 @@ pub(super) async fn sync_pull(
205 208 )]
206 209 #[tracing::instrument(skip_all, name = "synckit::sync_status")]
207 210 pub(super) async fn sync_status(
208 - State(state): State<AppState>,
211 + State(db): State<PgPool>,
209 212 sync_user: SyncUser,
210 213 ) -> Result<impl IntoResponse> {
211 214 let app_id = sync_user.app_id;
212 215 let user_id = sync_user.user_id;
213 216
214 217 let (total_changes, latest_cursor) =
215 - db::synckit::get_sync_status(&state.db, app_id, user_id).await?;
218 + db::synckit::get_sync_status(&db, app_id, user_id).await?;
216 219
217 220 Ok(Json(SyncStatusResponse {
218 221 total_changes,
@@ -228,10 +231,10 @@ pub(super) async fn sync_status(
228 231 )]
229 232 #[tracing::instrument(skip_all, name = "synckit::sync_account")]
230 233 pub(super) async fn sync_account(
231 - State(state): State<AppState>,
234 + State(db): State<PgPool>,
232 235 sync_user: SyncUser,
233 236 ) -> Result<impl IntoResponse> {
234 - let user = db::users::get_user_by_id(&state.db, sync_user.user_id)
237 + let user = db::users::get_user_by_id(&db, sync_user.user_id)
235 238 .await?
236 239 .ok_or(AppError::NotFound)?;
237 240
@@ -250,11 +253,11 @@ pub(super) async fn sync_account(
250 253 )]
251 254 #[tracing::instrument(skip_all, name = "synckit::sync_subscription_status")]
252 255 pub(super) async fn sync_subscription_status(
253 - State(state): State<AppState>,
256 + State(db): State<PgPool>,
254 257 sync_user: SyncUser,
255 258 ) -> Result<impl IntoResponse> {
256 259 let sub = db::synckit::get_user_app_subscription(
257 - &state.db,
260 + &db,
258 261 sync_user.user_id,
259 262 sync_user.app_id,
260 263 )
@@ -293,10 +296,10 @@ pub(super) async fn sync_subscription_status(
293 296 )]
294 297 #[tracing::instrument(skip_all, name = "synckit::get_app_pricing")]
295 298 pub(super) async fn get_app_pricing(
296 - State(state): State<AppState>,
299 + State(db): State<PgPool>,
297 300 Json(req): Json<AppPricingRequest>,
298 301 ) -> Result<impl IntoResponse> {
299 - let app = db::synckit::get_sync_app_by_api_key(&state.db, &req.api_key)
302 + let app = db::synckit::get_sync_app_by_api_key(&db, &req.api_key)
300 303 .await?
301 304 .ok_or(AppError::NotFound)?;
302 305
@@ -320,7 +323,6 @@ pub(super) async fn get_app_pricing(
320 323 )]
321 324 #[tracing::instrument(skip_all, name = "synckit::quote_subscription_price")]
322 325 pub(super) async fn quote_subscription_price(
323 - State(_state): State<AppState>,
324 326 _sync_user: SyncUser,
325 327 Json(req): Json<SyncQuoteRequest>,
326 328 ) -> Result<impl IntoResponse> {
@@ -343,11 +345,13 @@ pub(super) async fn quote_subscription_price(
343 345 )]
344 346 #[tracing::instrument(skip_all, name = "synckit::create_subscription_checkout")]
345 347 pub(super) async fn create_subscription_checkout(
346 - State(state): State<AppState>,
348 + State(db): State<PgPool>,
349 + State(payments): State<crate::Billing>,
350 + State(config): State<Config>,
347 351 sync_user: SyncUser,
348 352 Json(req): Json<SyncSubscribeRequest>,
349 353 ) -> Result<impl IntoResponse> {
350 - if db::synckit::get_user_app_subscription(&state.db, sync_user.user_id, sync_user.app_id)
354 + if db::synckit::get_user_app_subscription(&db, sync_user.user_id, sync_user.app_id)
351 355 .await?
352 356 .is_some()
353 357 {
@@ -359,19 +363,18 @@ pub(super) async fn create_subscription_checkout(
359 363 let interval = SyncBillingInterval::parse(&req.interval)?;
360 364 let amount_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
361 365
362 - let app = db::synckit::get_sync_app_by_id(&state.db, sync_user.app_id)
366 + let app = db::synckit::get_sync_app_by_id(&db, sync_user.app_id)
363 367 .await?
364 368 .ok_or(AppError::NotFound)?;
365 369 let cap_gib = req.cap_bytes / (1024 * 1024 * 1024);
366 370 let product_name = format!("{} cloud sync — {} GiB", app.name, cap_gib);
367 371
368 - let stripe = state
369 - .stripe
372 + let stripe = payments.stripe
370 373 .as_ref()
371 374 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
372 375
373 - let success_url = format!("{}/sync/subscribed", state.config.host_url);
374 - let cancel_url = format!("{}/sync/canceled", state.config.host_url);
376 + let success_url = format!("{}/sync/subscribed", config.host_url);
377 + let cancel_url = format!("{}/sync/canceled", config.host_url);
375 378
376 379 let result = stripe
377 380 .create_synckit_app_sub_checkout_session(&crate::payments::SynckitAppSubCheckoutParams {
@@ -409,11 +412,12 @@ pub(super) async fn create_subscription_checkout(
409 412 )]
410 413 #[tracing::instrument(skip_all, name = "synckit::queue_storage_cap_change")]
411 414 pub(super) async fn queue_storage_cap_change(
412 - State(state): State<AppState>,
415 + State(db): State<PgPool>,
416 + State(payments): State<crate::Billing>,
413 417 sync_user: SyncUser,
414 418 Json(req): Json<SyncCapChangeRequest>,
415 419 ) -> Result<impl IntoResponse> {
416 - let sub = db::synckit::get_user_app_subscription(&state.db, sync_user.user_id, sync_user.app_id)
420 + let sub = db::synckit::get_user_app_subscription(&db, sync_user.user_id, sync_user.app_id)
417 421 .await?
418 422 .ok_or_else(|| AppError::BadRequest("No active subscription to adjust".to_string()))?;
419 423
@@ -428,12 +432,11 @@ pub(super) async fn queue_storage_cap_change(
428 432 let interval = payments::SyncBillingInterval::parse(&sub.interval)?;
429 433 let new_price_cents = payments::quote_price_cents(req.cap_bytes, interval)?;
430 434
431 - let stripe = state
432 - .stripe
435 + let stripe = payments.stripe
433 436 .as_ref()
434 437 .ok_or_else(|| AppError::BadRequest("Stripe is not configured".to_string()))?;
435 438
436 - let app = db::synckit::get_sync_app_by_id(&state.db, sync_user.app_id)
439 + let app = db::synckit::get_sync_app_by_id(&db, sync_user.app_id)
437 440 .await?
438 441 .ok_or(AppError::NotFound)?;
439 442 let cap_gib = req.cap_bytes / (1024 * 1024 * 1024);
@@ -451,7 +454,7 @@ pub(super) async fn queue_storage_cap_change(
451 454 .await?;
452 455
453 456 db::synckit::set_pending_storage_cap(
454 - &state.db,
457 + &db,
455 458 sync_user.user_id,
456 459 sync_user.app_id,
457 460 req.cap_bytes,
@@ -479,7 +482,7 @@ pub(super) async fn queue_storage_cap_change(
479 482 )]
480 483 #[tracing::instrument(skip_all, name = "synckit::register_device")]
481 484 pub(super) async fn register_device(
482 - State(state): State<AppState>,
485 + State(db): State<PgPool>,
483 486 sync_user: SyncUser,
484 487 Json(req): Json<RegisterDeviceRequest>,
485 488 ) -> Result<impl IntoResponse> {
@@ -490,7 +493,7 @@ pub(super) async fn register_device(
490 493 // scoped; holding `_lock_tx` to the end of the handler keeps a second
491 494 // concurrent registration blocked until this one's upsert commits, so it
492 495 // sees the updated count and can't push past the cap (Run 21 concurrency).
493 - let mut _lock_tx = state.db.begin().await?;
496 + let mut _lock_tx = db.begin().await?;
494 497 sqlx::query(
495 498 "SELECT pg_advisory_xact_lock(hashtextextended('synckit_device:' || $1::text || ':' || $2::text, 0))",
496 499 )
@@ -500,10 +503,10 @@ pub(super) async fn register_device(
500 503 .await?;
501 504
502 505 // Enforce device limit (upsert on existing name is fine, only new names count)
503 - let count = db::synckit::count_sync_devices(&state.db, sync_user.app_id, sync_user.user_id).await?;
506 + let count = db::synckit::count_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
504 507 if count >= constants::SYNCKIT_MAX_DEVICES_PER_APP {
505 508 // Check if this is an existing device (upsert) — allow updates
506 - let existing = db::synckit::get_sync_devices(&state.db, sync_user.app_id, sync_user.user_id).await?;
509 + let existing = db::synckit::get_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
507 510 if !existing.iter().any(|d| d.device_name == req.device_name) {
508 511 return Err(AppError::BadRequest(format!(
509 512 "Maximum {} devices per app",
@@ -513,7 +516,7 @@ pub(super) async fn register_device(
513 516 }
514 517
515 518 let device = db::synckit::upsert_sync_device(
516 - &state.db,
519 + &db,
517 520 sync_user.app_id,
518 521 sync_user.user_id,
519 522 &req.device_name,
@@ -539,11 +542,11 @@ pub(super) async fn register_device(
539 542 )]
540 543 #[tracing::instrument(skip_all, name = "synckit::list_devices")]
541 544 pub(super) async fn list_devices(
542 - State(state): State<AppState>,
545 + State(db): State<PgPool>,
543 546 sync_user: SyncUser,
544 547 ) -> Result<impl IntoResponse> {
545 548 let devices =
546 - db::synckit::get_sync_devices(&state.db, sync_user.app_id, sync_user.user_id).await?;
549 + db::synckit::get_sync_devices(&db, sync_user.app_id, sync_user.user_id).await?;
547 550
548 551 let response: Vec<SyncDeviceResponse> = devices.into_iter().map(|d| SyncDeviceResponse {
549 552 id: d.id,
@@ -566,13 +569,13 @@ pub(super) async fn list_devices(
566 569 )]
567 570 #[tracing::instrument(skip_all, name = "synckit::delete_device")]
568 571 pub(super) async fn delete_device(
569 - State(state): State<AppState>,
572 + State(db): State<PgPool>,
570 573 sync_user: SyncUser,
571 574 headers: axum::http::HeaderMap,
572 575 Path(device_id): Path<SyncDeviceId>,
573 576 ) -> Result<impl IntoResponse> {
574 577 let deleted =
575 - db::synckit::delete_sync_device(&state.db, device_id, sync_user.app_id, sync_user.user_id)
578 + db::synckit::delete_sync_device(&db, device_id, sync_user.app_id, sync_user.user_id)
576 579 .await?;
577 580
578 581 if !deleted {
@@ -584,12 +587,12 @@ pub(super) async fn delete_device(
584 587 // expiry), and record the event for the audit log. Best-effort — a logging
585 588 // or invalidation hiccup must not fail the delete the user asked for.
586 589 let ip = crate::helpers::extract_client_ip(&headers);
587 - if let Err(e) = db::synckit::invalidate_user_sync_tokens(&state.db, sync_user.user_id).await {
590 + if let Err(e) = db::synckit::invalidate_user_sync_tokens(&db, sync_user.user_id).await {
588 591 tracing::error!(error = ?e, user_id = %sync_user.user_id,
589 592 "failed to invalidate sync tokens after device removal");
590 593 }
591 594 if let Err(e) = db::synckit::record_security_event(
592 - &state.db,
595 + &db,
593 596 sync_user.app_id,
594 597 Some(sync_user.user_id),
595 598 db::synckit::sync_security_event::DEVICE_REMOVED,
@@ -614,7 +617,7 @@ pub(super) async fn delete_device(
614 617 )]
615 618 #[tracing::instrument(skip_all, name = "synckit::put_sync_key")]
616 619 pub(super) async fn put_sync_key(
617 - State(state): State<AppState>,
620 + State(db): State<PgPool>,
618 621 sync_user: SyncUser,
619 622 Json(req): Json<PutKeyRequest>,
620 623 ) -> Result<impl IntoResponse> {
@@ -626,7 +629,7 @@ pub(super) async fn put_sync_key(
626 629 }
627 630
628 631 let updated = db::synckit::upsert_sync_key(
629 - &state.db,
632 + &db,
630 633 sync_user.app_id,
631 634 sync_user.user_id,
632 635 &req.encrypted_key,
@@ -650,11 +653,11 @@ pub(super) async fn put_sync_key(
650 653 )]
651 654 #[tracing::instrument(skip_all, name = "synckit::get_sync_key")]
652 655 pub(super) async fn get_sync_key(
653 - State(state): State<AppState>,
656 + State(db): State<PgPool>,
654 657 sync_user: SyncUser,
655 658 ) -> Result<impl IntoResponse> {
656 659 let info =
657 - db::synckit::get_sync_key(&state.db, sync_user.app_id, sync_user.user_id)
660 + db::synckit::get_sync_key(&db, sync_user.app_id, sync_user.user_id)
658 661 .await?
659 662 .ok_or(AppError::NotFound)?;
660 663
@@ -683,7 +686,7 @@ pub(super) async fn get_sync_key(
683 686 )]
684 687 #[tracing::instrument(skip_all, name = "synckit::begin_rotation")]
685 688 pub(super) async fn begin_rotation(
686 - State(state): State<AppState>,
689 + State(db): State<PgPool>,
687 690 sync_user: SyncUser,
688 691 Json(req): Json<BeginRotationRequest>,
689 692 ) -> Result<impl IntoResponse> {
@@ -695,12 +698,12 @@ pub(super) async fn begin_rotation(
695 698
696 699 // Verify device belongs to this user + app via an indexed point lookup, not a
697 700 // fetch-all-then-scan (ultra-fuzz Run 4 Perf).
698 - if !db::synckit::sync_device_belongs(&state.db, req.device_id, sync_user.app_id, sync_user.user_id).await? {
701 + if !db::synckit::sync_device_belongs(&db, req.device_id, sync_user.app_id, sync_user.user_id).await? {
699 702 return Err(AppError::BadRequest("Unknown device".to_string()));
700 703 }
701 704
702 705 let result = db::synckit::begin_key_rotation(
703 - &state.db,
706 + &db,
704 707 sync_user.app_id,
705 708 sync_user.user_id,
706 709 req.device_id,
@@ -727,12 +730,12 @@ pub(super) async fn begin_rotation(
727 730 )]
728 731 #[tracing::instrument(skip_all, name = "synckit::rotation_entries")]
729 732 pub(super) async fn rotation_entries(
730 - State(state): State<AppState>,
733 + State(db): State<PgPool>,
731 734 sync_user: SyncUser,
732 735 Json(req): Json<RotationEntriesRequest>,
733 736 ) -> Result<impl IntoResponse> {
734 737 let rotation = db::synckit::get_key_rotation(
735 - &state.db,
738 + &db,
736 739 sync_user.app_id,
737 740 sync_user.user_id,
738 741 )
@@ -745,7 +748,7 @@ pub(super) async fn rotation_entries(
745 748
746 749 let page_size = constants::SYNCKIT_ROTATION_BATCH_MAX as i64;
747 750 let raw_entries = db::synckit::get_rotation_entries(
748 - &state.db,
751 + &db,
749 752 sync_user.app_id,
750 753 sync_user.user_id,
751 754 rotation.new_key_id,
@@ -771,7 +774,7 @@ pub(super) async fn rotation_entries(
771 774 )]
772 775 #[tracing::instrument(skip_all, name = "synckit::rotation_batch")]
773 776 pub(super) async fn rotation_batch(
774 - State(state): State<AppState>,
777 + State(db): State<PgPool>,
775 778 sync_user: SyncUser,
776 779 Json(req): Json<RotationBatchRequest>,
777 780 ) -> Result<impl IntoResponse> {
@@ -786,7 +789,7 @@ pub(super) async fn rotation_batch(
786 789 }
787 790
788 791 let rotation = db::synckit::get_key_rotation(
789 - &state.db,
792 + &db,
790 793 sync_user.app_id,
791 794 sync_user.user_id,
792 795 )
@@ -804,7 +807,7 @@ pub(super) async fn rotation_batch(
804 807 .collect();
805 808
806 809 let updated_count = db::synckit::submit_rotation_batch(
807 - &state.db,
810 + &db,
808 811 sync_user.app_id,
809 812 sync_user.user_id,
810 813 rotation.id,
@@ -827,12 +830,12 @@ pub(super) async fn rotation_batch(
827 830 )]
828 831 #[tracing::instrument(skip_all, name = "synckit::complete_rotation")]
829 832 pub(super) async fn complete_rotation(
830 - State(state): State<AppState>,
833 + State(db): State<PgPool>,
831 834 sync_user: SyncUser,
832 835 Json(req): Json<CompleteRotationRequest>,
833 836 ) -> Result<impl IntoResponse> {
834 837 let result = db::synckit::complete_key_rotation(
835 - &state.db,
838 + &db,
836 839 sync_user.app_id,
837 840 sync_user.user_id,
838 841 req.rotation_id,
@@ -843,7 +846,7 @@ pub(super) async fn complete_rotation(
843 846 Ok(_new_key_id) => {
844 847 // Audit the completed rotation (best-effort).
845 848 if let Err(e) = db::synckit::record_security_event(
846 - &state.db,
849 + &db,
847 850 sync_user.app_id,
848 851 Some(sync_user.user_id),
849 852 db::synckit::sync_security_event::KEY_ROTATION_COMPLETED,
@@ -874,11 +877,11 @@ pub(super) async fn complete_rotation(
874 877 )]
875 878 #[tracing::instrument(skip_all, name = "synckit::cancel_rotation")]
876 879 pub(super) async fn cancel_rotation(
877 - State(state): State<AppState>,
880 + State(db): State<PgPool>,
878 881 sync_user: SyncUser,
879 882 ) -> Result<impl IntoResponse> {
880 883 let cancelled = db::synckit::cancel_stale_rotation(
881 - &state.db,
884 + &db,
882 885 sync_user.app_id,
883 886 sync_user.user_id,
884 887 constants::SYNCKIT_ROTATION_STALE_HOURS,