Skip to main content

max / makenotwork

29.7 KB · 879 lines History Blame Raw
1 //! SyncKit v2 developer-billing queries.
2 //!
3 //! Operates on the billing columns added to `sync_apps` and the auxiliary
4 //! `sync_app_usage_current` table by migration 117 (`117_synckit_v2_billing.sql`),
5 //! as revised by 118 and 119.
6 //!
7 //! Schema reference, for reading these queries without the migrations to hand:
8 //!
9 //! ```text
10 //! sync_apps:
11 //! is_internal BOOLEAN NOT NULL DEFAULT FALSE
12 //! stripe_customer_id TEXT
13 //! stripe_subscription_id TEXT UNIQUE
14 //! billing_status TEXT NOT NULL CHECK IN
15 //! ('draft','active','suspended_unpaid','canceled')
16 //! storage_gb_cap INT
17 //! egress_multiple NUMERIC(6,2)
18 //! enforcement_mode TEXT NOT NULL CHECK IN ('per_key','bulk')
19 //! (117 wrote 'app_wide'; 118 renamed it 'bulk')
20 //! key_cap INT
21 //! current_period_start TIMESTAMPTZ
22 //! current_period_end TIMESTAMPTZ
23 //!
24 //! sync_app_usage_current (one row per app):
25 //! app_id, bytes_stored, bytes_egress_period, keys_claimed,
26 //! last_warning_pct, period_started_at, updated_at
27 //! ```
28
29 use chrono::{DateTime, Utc};
30 use sqlx::PgPool;
31
32 use super::id_types::SyncAppId;
33 use super::models::{DbSyncAppBilling, DbSyncAppKey};
34 use crate::error::Result;
35
36 /// Outcome of a `claim_key` call.
37 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
38 pub struct ClaimResult {
39 /// `true` if this call inserted a new active claim row;
40 /// `false` if the key was already actively claimed (idempotent re-claim)
41 /// or the cap was reached (see `cap_reached`).
42 pub newly_claimed: bool,
43 /// `true` if a new slot was refused because `key_cap` is already reached.
44 /// Mutually exclusive with `newly_claimed`. An idempotent re-claim of an
45 /// already-active key never sets this (it consumes no new slot).
46 pub cap_reached: bool,
47 /// Total active claims for this app after the operation.
48 pub total_claimed: i32,
49 }
50
51 /// Outcome of a `release_key` call.
52 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
53 pub struct ReleaseResult {
54 /// `true` if this call transitioned an active row to released;
55 /// `false` if no active row existed (idempotent release).
56 pub newly_released: bool,
57 /// Total active claims for this app after the operation.
58 pub total_claimed: i32,
59 }
60
61 /// Set the Stripe customer ID on a sync app (idempotent; re-sets the same id
62 /// without error). Called by the billing setup route once the developer-side
63 /// Customer object has been created in Stripe.
64 #[tracing::instrument(skip_all)]
65 pub async fn set_stripe_customer(
66 pool: &PgPool,
67 app_id: SyncAppId,
68 stripe_customer_id: &str,
69 ) -> Result<()> {
70 sqlx::query("UPDATE sync_apps SET stripe_customer_id = $2 WHERE id = $1")
71 .bind(app_id)
72 .bind(stripe_customer_id)
73 .execute(pool)
74 .await?;
75 Ok(())
76 }
77
78 /// Activate billing on a draft app: stamps subscription id, knobs, status,
79 /// and current period. The `egress_multiple` is passed in as `f64` and cast
80 /// to the column's `NUMERIC(6,2)` type in SQL.
81 #[tracing::instrument(skip_all)]
82 #[allow(clippy::too_many_arguments)]
83 pub async fn activate_billing(
84 pool: &PgPool,
85 app_id: SyncAppId,
86 enforcement_mode: super::SyncEnforcementMode,
87 storage_gb_cap: Option<i32>,
88 key_cap: Option<i32>,
89 gb_per_key: Option<i32>,
90 stripe_sub_id: &str,
91 period_start: DateTime<Utc>,
92 period_end: DateTime<Utc>,
93 ) -> Result<()> {
94 // Guard `billing_status = 'draft'` at the DB layer, not just in the handler:
95 // activation is only ever valid from draft, so a replay or a TOCTOU race that
96 // reaches here against a canceled (terminal) or already-active app must NOT
97 // silently resurrect it / orphan a live subscription (audit Run 13). 0 rows
98 // affected → the app left draft; surface a conflict rather than succeeding.
99 let result = sqlx::query(
100 r"
101 UPDATE sync_apps SET
102 billing_status = 'active',
103 stripe_subscription_id = $2,
104 enforcement_mode = $3,
105 storage_gb_cap = $4,
106 key_cap = $5,
107 gb_per_key = $6,
108 current_period_start = $7,
109 current_period_end = $8
110 WHERE id = $1 AND billing_status = 'draft'
111 ",
112 )
113 .bind(app_id)
114 .bind(stripe_sub_id)
115 .bind(enforcement_mode)
116 .bind(storage_gb_cap)
117 .bind(key_cap)
118 .bind(gb_per_key)
119 .bind(period_start)
120 .bind(period_end)
121 .execute(pool)
122 .await?;
123 if result.rows_affected() == 0 {
124 return Err(crate::error::AppError::Conflict(
125 "Billing activation is only valid for a draft app".to_string(),
126 ));
127 }
128 Ok(())
129 }
130
131 /// Update the pricing knobs on an already-active app (no status change).
132 #[tracing::instrument(skip_all)]
133 pub async fn update_knobs(
134 pool: &PgPool,
135 app_id: SyncAppId,
136 enforcement_mode: super::SyncEnforcementMode,
137 storage_gb_cap: Option<i32>,
138 key_cap: Option<i32>,
139 gb_per_key: Option<i32>,
140 ) -> Result<()> {
141 sqlx::query(
142 r"
143 UPDATE sync_apps SET
144 enforcement_mode = $2,
145 storage_gb_cap = $3,
146 key_cap = $4,
147 gb_per_key = $5
148 WHERE id = $1
149 ",
150 )
151 .bind(app_id)
152 .bind(enforcement_mode)
153 .bind(storage_gb_cap)
154 .bind(key_cap)
155 .bind(gb_per_key)
156 .execute(pool)
157 .await?;
158 Ok(())
159 }
160
161 /// Apply a Stripe-driven billing update to a sync app: optionally change
162 /// `billing_status` and/or refresh the current period, in ONE guarded statement.
163 /// Returns `true` if a live (non-canceled) row was updated.
164 ///
165 /// `canceled` is terminal: the `AND (billing_status != 'canceled' OR $2 = 'canceled')`
166 /// guard refuses to move a canceled app back to active/suspended_unpaid, AND,
167 /// because the period columns are written here under the same guard, a stray
168 /// `invoice.paid` after a `deleted` can no longer refresh the period (nor, via
169 /// the returned `false`, reset usage) on a canceled app. This replaces the old
170 /// split `set_billing_status` (guarded) + `set_period` (UNGUARDED), the synckit
171 /// instance of the period-without-guard bug class. The legitimate
172 /// `suspended_unpaid -> active` recovery on `invoice.paid` is unaffected
173 /// (suspended_unpaid != canceled). Initial activation from `draft` goes through
174 /// `activate_billing`, not this path. See `crate::db::subscription_writer` for
175 /// the cross-family rationale.
176 #[tracing::instrument(skip_all)]
177 /// `period` is the **raw Stripe** `(current_period_start, current_period_end)`
178 /// as Unix seconds. The conversion is sealed here: a `None` or a non-positive
179 /// `end` writes no period (the `COALESCE` keeps the existing value), so a thin/
180 /// zero webhook can never stamp a 1970 period onto an active app. Handlers pass
181 /// the raw value and cannot hand in a `DateTime`.
182 pub async fn apply_billing_update<'e>(
183 executor: impl sqlx::PgExecutor<'e>,
184 app_id: SyncAppId,
185 status: Option<&str>,
186 period: Option<(i64, i64)>,
187 ) -> Result<bool> {
188 let (period_start, period_end) = match period {
189 // Require a positive, non-inverted window. A non-positive `end` is the
190 // thin/zero-webhook guard (no 1970 period); `start <= end` additionally
191 // rejects an inverted range, so a malformed Stripe period writes nothing
192 // (the `COALESCE` keeps the existing values) rather than stamping an
193 // end-before-start window onto a live app.
194 Some((start, end)) if end > 0 && start <= end => (
195 DateTime::<Utc>::from_timestamp(start, 0),
196 DateTime::<Utc>::from_timestamp(end, 0),
197 ),
198 _ => (None, None),
199 };
200 let result = sqlx::query(
201 r"
202 UPDATE sync_apps SET
203 billing_status = COALESCE($2, billing_status),
204 current_period_start = COALESCE($3, current_period_start),
205 current_period_end = COALESCE($4, current_period_end)
206 WHERE id = $1
207 AND (billing_status != 'canceled' OR $2 = 'canceled')
208 ",
209 )
210 .bind(app_id)
211 .bind(status)
212 .bind(period_start)
213 .bind(period_end)
214 .execute(executor)
215 .await?;
216 Ok(result.rows_affected() > 0)
217 }
218
219 /// Reset the per-period usage counters on `sync_app_usage_current`. Called
220 /// from the `invoice.paid` webhook handler at period rollover.
221 #[tracing::instrument(skip_all)]
222 pub async fn reset_period_usage<'e>(
223 executor: impl sqlx::PgExecutor<'e>,
224 app_id: SyncAppId,
225 ) -> Result<()> {
226 sqlx::query(
227 r"
228 UPDATE sync_app_usage_current SET
229 bytes_egress_period = 0,
230 last_warning_pct = 0,
231 period_started_at = NOW(),
232 updated_at = NOW()
233 WHERE app_id = $1
234 ",
235 )
236 .bind(app_id)
237 .execute(executor)
238 .await?;
239 Ok(())
240 }
241
242 /// Look up the sync app that owns a given Stripe subscription. Used by the
243 /// webhook router to distinguish SyncKit v2 subscriptions from
244 /// creator-tier / Fan+ subscriptions.
245 #[tracing::instrument(skip_all)]
246 pub async fn get_app_by_stripe_subscription(
247 pool: &PgPool,
248 stripe_sub_id: &str,
249 ) -> Result<Option<SyncAppId>> {
250 let row: Option<(SyncAppId,)> =
251 sqlx::query_as("SELECT id FROM sync_apps WHERE stripe_subscription_id = $1")
252 .bind(stripe_sub_id)
253 .fetch_optional(pool)
254 .await?;
255 Ok(row.map(|(id,)| id))
256 }
257
258 /// Fetch the combined app+billing+usage view for a single app. `egress_multiple`
259 /// is cast from NUMERIC to DOUBLE PRECISION so it decodes into `f64` without
260 /// the `bigdecimal` sqlx feature. `sync_app_usage_current` is LEFT-joined: the
261 /// usage row is created transactionally with the app (`create_sync_app`) and
262 /// backfilled by migration 165, but the `Option`/LEFT JOIN stays as
263 /// defense-in-depth so a missing row can't blow up the read path.
264 #[tracing::instrument(skip_all)]
265 pub async fn get_app_with_billing(
266 pool: &PgPool,
267 app_id: SyncAppId,
268 ) -> Result<Option<DbSyncAppBilling>> {
269 let app = sqlx::query_as::<_, DbSyncAppBilling>(
270 r"
271 SELECT
272 sa.id,
273 sa.creator_id,
274 sa.name,
275 sa.is_internal,
276 sa.stripe_customer_id,
277 sa.stripe_subscription_id,
278 sa.billing_status,
279 sa.storage_gb_cap,
280 sa.enforcement_mode,
281 sa.key_cap,
282 sa.gb_per_key,
283 sa.current_period_start,
284 sa.current_period_end,
285 u.bytes_stored,
286 u.bytes_egress_period,
287 u.keys_claimed,
288 u.last_warning_pct,
289 u.period_started_at,
290 p.slug AS project_slug
291 FROM sync_apps sa
292 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
293 LEFT JOIN projects p ON p.id = sa.project_id
294 WHERE sa.id = $1
295 ",
296 )
297 .bind(app_id)
298 .fetch_optional(pool)
299 .await?;
300 Ok(app)
301 }
302
303 /// Batch variant of `get_app_with_billing` that loads every app owned by a
304 /// creator with its billing+usage join in one query. Used by the user-level
305 /// SyncKit dashboard tab.
306 #[tracing::instrument(skip_all)]
307 pub async fn get_apps_with_billing_by_creator(
308 pool: &PgPool,
309 creator_id: super::id_types::UserId,
310 ) -> Result<Vec<DbSyncAppBilling>> {
311 let apps = sqlx::query_as::<_, DbSyncAppBilling>(
312 r"
313 SELECT
314 sa.id,
315 sa.creator_id,
316 sa.name,
317 sa.is_internal,
318 sa.stripe_customer_id,
319 sa.stripe_subscription_id,
320 sa.billing_status,
321 sa.storage_gb_cap,
322 sa.enforcement_mode,
323 sa.key_cap,
324 sa.gb_per_key,
325 sa.current_period_start,
326 sa.current_period_end,
327 u.bytes_stored,
328 u.bytes_egress_period,
329 u.keys_claimed,
330 u.last_warning_pct,
331 u.period_started_at,
332 p.slug AS project_slug
333 FROM sync_apps sa
334 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
335 LEFT JOIN projects p ON p.id = sa.project_id
336 WHERE sa.creator_id = $1
337 ",
338 )
339 .bind(creator_id)
340 .fetch_all(pool)
341 .await?;
342 Ok(apps)
343 }
344
345 /// Batch variant of `get_app_with_billing` for one project. Used by the
346 /// project-level SyncKit dashboard tab.
347 #[tracing::instrument(skip_all)]
348 pub async fn get_apps_with_billing_by_project(
349 pool: &PgPool,
350 project_id: super::id_types::ProjectId,
351 ) -> Result<Vec<DbSyncAppBilling>> {
352 let apps = sqlx::query_as::<_, DbSyncAppBilling>(
353 r"
354 SELECT
355 sa.id,
356 sa.creator_id,
357 sa.name,
358 sa.is_internal,
359 sa.stripe_customer_id,
360 sa.stripe_subscription_id,
361 sa.billing_status,
362 sa.storage_gb_cap,
363 sa.enforcement_mode,
364 sa.key_cap,
365 sa.gb_per_key,
366 sa.current_period_start,
367 sa.current_period_end,
368 u.bytes_stored,
369 u.bytes_egress_period,
370 u.keys_claimed,
371 u.last_warning_pct,
372 u.period_started_at,
373 p.slug AS project_slug
374 FROM sync_apps sa
375 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
376 LEFT JOIN projects p ON p.id = sa.project_id
377 WHERE sa.project_id = $1
378 ",
379 )
380 .bind(project_id)
381 .fetch_all(pool)
382 .await?;
383 Ok(apps)
384 }
385
386 /// Claim an SDK encryption key for an app. Idempotent: a re-claim of an
387 /// already-active key returns `newly_claimed = false` without inserting.
388 ///
389 /// `key_cap` is the per-app active-key ceiling (`Some` only for `per_key`
390 /// developer apps; `None` means uncapped). The cap is checked **inside** the
391 /// transaction, under the `FOR UPDATE` lock on `sync_app_usage_current`, so
392 /// concurrent claims of distinct new keys can't collectively overshoot the cap
393 ///, the prior design checked the cap pre-transaction in the handler and could
394 /// over-allocate. A re-claim of an already-active key is admitted regardless of
395 /// the cap (it consumes no new slot).
396 #[tracing::instrument(skip_all)]
397 pub async fn claim_key(
398 pool: &sqlx::PgPool,
399 app_id: SyncAppId,
400 key: &str,
401 key_cap: Option<i32>,
402 ) -> Result<ClaimResult> {
403 let mut tx = pool.begin().await?;
404
405 // Lock the usage row for this app. Returns the current keys_claimed.
406 let (mut keys_claimed,): (i32,) = sqlx::query_as(
407 "SELECT keys_claimed FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
408 )
409 .bind(app_id)
410 .fetch_one(&mut *tx)
411 .await?;
412
413 // Is there already an active claim for this key?
414 let existing: Option<(uuid::Uuid,)> = sqlx::query_as(
415 "SELECT id FROM sync_app_keys
416 WHERE app_id = $1 AND key = $2 AND released_at IS NULL",
417 )
418 .bind(app_id)
419 .bind(key)
420 .fetch_optional(&mut *tx)
421 .await?;
422
423 if existing.is_some() {
424 // Idempotent re-claim, no new slot, cap not consulted.
425 tx.commit().await?;
426 return Ok(ClaimResult {
427 newly_claimed: false,
428 cap_reached: false,
429 total_claimed: keys_claimed,
430 });
431 }
432
433 // New claim: enforce the cap under the lock.
434 if let Some(cap) = key_cap
435 && keys_claimed >= cap
436 {
437 tx.commit().await?;
438 return Ok(ClaimResult {
439 newly_claimed: false,
440 cap_reached: true,
441 total_claimed: keys_claimed,
442 });
443 }
444
445 sqlx::query("INSERT INTO sync_app_keys (app_id, key) VALUES ($1, $2)")
446 .bind(app_id)
447 .bind(key)
448 .execute(&mut *tx)
449 .await?;
450 sqlx::query(
451 "UPDATE sync_app_usage_current
452 SET keys_claimed = keys_claimed + 1, updated_at = NOW()
453 WHERE app_id = $1",
454 )
455 .bind(app_id)
456 .execute(&mut *tx)
457 .await?;
458 keys_claimed += 1;
459
460 tx.commit().await?;
461 Ok(ClaimResult {
462 newly_claimed: true,
463 cap_reached: false,
464 total_claimed: keys_claimed,
465 })
466 }
467
468 /// Release an SDK encryption key. Idempotent: releasing a key that is not
469 /// actively claimed returns `newly_released = false`.
470 #[tracing::instrument(skip_all)]
471 pub async fn release_key(
472 pool: &sqlx::PgPool,
473 app_id: SyncAppId,
474 key: &str,
475 ) -> Result<ReleaseResult> {
476 let mut tx = pool.begin().await?;
477
478 let (mut keys_claimed,): (i32,) = sqlx::query_as(
479 "SELECT keys_claimed FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
480 )
481 .bind(app_id)
482 .fetch_one(&mut *tx)
483 .await?;
484
485 let released: Option<(uuid::Uuid,)> = sqlx::query_as(
486 "UPDATE sync_app_keys SET released_at = NOW()
487 WHERE app_id = $1 AND key = $2 AND released_at IS NULL
488 RETURNING id",
489 )
490 .bind(app_id)
491 .bind(key)
492 .fetch_optional(&mut *tx)
493 .await?;
494
495 let newly_released = if released.is_some() {
496 sqlx::query(
497 "UPDATE sync_app_usage_current
498 SET keys_claimed = GREATEST(keys_claimed - 1, 0), updated_at = NOW()
499 WHERE app_id = $1",
500 )
501 .bind(app_id)
502 .execute(&mut *tx)
503 .await?;
504 keys_claimed = (keys_claimed - 1).max(0);
505 true
506 } else {
507 false
508 };
509
510 tx.commit().await?;
511 Ok(ReleaseResult {
512 newly_released,
513 total_claimed: keys_claimed,
514 })
515 }
516
517 /// List active (un-released) key claims for an app, ordered by `claimed_at DESC`.
518 /// Used by the dashboard "Active keys" view. `bytes_stored` is LEFT-joined
519 /// from `sync_key_usage_current`, `0` when no upload has landed yet.
520 #[tracing::instrument(skip_all)]
521 pub async fn list_active_keys(
522 pool: &sqlx::PgPool,
523 app_id: SyncAppId,
524 limit: i64,
525 offset: i64,
526 ) -> Result<Vec<DbSyncAppKey>> {
527 let rows = sqlx::query_as::<_, DbSyncAppKey>(
528 r"
529 SELECT k.id, k.key, k.claimed_at,
530 COALESCE(u.bytes_stored, 0) AS bytes_stored
531 FROM sync_app_keys k
532 LEFT JOIN sync_key_usage_current u
533 ON u.app_id = k.app_id AND u.key = k.key
534 WHERE k.app_id = $1 AND k.released_at IS NULL
535 ORDER BY k.claimed_at DESC
536 LIMIT $2 OFFSET $3
537 ",
538 )
539 .bind(app_id)
540 .bind(limit)
541 .bind(offset)
542 .fetch_all(pool)
543 .await?;
544 Ok(rows)
545 }
546
547 /// Per-app top-N key usage for the dashboard panel. Returns the highest-usage
548 /// active keys (sorted by `bytes_stored DESC`) for each app in `app_ids`,
549 /// batched into a single query so the integrations page doesn't N+1 across
550 /// every per_key-mode app.
551 #[tracing::instrument(skip_all)]
552 pub async fn get_top_keys_per_app(
553 pool: &PgPool,
554 app_ids: &[SyncAppId],
555 limit_per_app: i64,
556 ) -> Result<Vec<(SyncAppId, String, i64)>> {
557 if app_ids.is_empty() {
558 return Ok(Vec::new());
559 }
560 let rows: Vec<(SyncAppId, String, i64)> = sqlx::query_as(
561 r"
562 SELECT app_id, key, bytes_stored
563 FROM (
564 SELECT u.app_id, u.key, u.bytes_stored,
565 ROW_NUMBER() OVER (PARTITION BY u.app_id ORDER BY u.bytes_stored DESC, u.key) AS rn
566 FROM sync_key_usage_current u
567 JOIN sync_app_keys k
568 ON k.app_id = u.app_id AND k.key = u.key AND k.released_at IS NULL
569 WHERE u.app_id = ANY($1)
570 ) ranked
571 WHERE rn <= $2
572 ORDER BY app_id, rn
573 ",
574 )
575 .bind(app_ids)
576 .bind(limit_per_app)
577 .fetch_all(pool)
578 .await?;
579 Ok(rows)
580 }
581
582 // ── Phase 5: Usage counters and cap enforcement ──
583
584 /// A row that may need a warning email sent.
585 ///
586 /// `threshold_pct` is the highest WARNING_THRESHOLDS_PCT band currently
587 /// breached above `last_warning_pct`. The caller is responsible for filtering
588 /// out apps where no new breach has occurred (i.e. usage hasn't crossed the
589 /// next threshold since the previous warning).
590 #[derive(Debug, Clone, PartialEq, Eq)]
591 pub struct WarningCandidate {
592 pub app_id: SyncAppId,
593 pub creator_id: super::id_types::UserId,
594 pub creator_email: String,
595 pub app_name: String,
596 pub threshold_pct: i16,
597 pub dimension: &'static str,
598 pub used: i64,
599 pub limit: i64,
600 /// SDK key that breached the threshold, when `dimension ==
601 /// "storage_per_key"`. `None` for app-wide breaches.
602 pub key: Option<String>,
603 }
604
605 /// Increment (or decrement) the per-period `bytes_egress_period` counter.
606 /// Returns the new value.
607 #[tracing::instrument(skip_all)]
608 pub async fn add_bytes_egress(pool: &PgPool, app_id: SyncAppId, delta: i64) -> Result<i64> {
609 let (new_val,): (i64,) = sqlx::query_as(
610 r"
611 UPDATE sync_app_usage_current
612 SET bytes_egress_period = GREATEST(bytes_egress_period + $2, 0), updated_at = NOW()
613 WHERE app_id = $1
614 RETURNING bytes_egress_period
615 ",
616 )
617 .bind(app_id)
618 .bind(delta)
619 .fetch_one(pool)
620 .await?;
621 Ok(new_val)
622 }
623
624 /// Fetch the list of active, non-internal apps that may need a warning email,
625 /// along with the data needed to compute which threshold (if any) has been
626 /// breached since the last notice. The per-app breach computation is done in
627 /// Rust by the caller (see `scheduler::synckit_warnings`).
628 /// `limit` bounds the app scan so a single scheduler tick does a bounded amount
629 /// of work; warned apps stamp `last_warning_pct` and drop out of the candidate
630 /// set, so any overflow drains on subsequent ticks.
631 #[tracing::instrument(skip_all)]
632 pub async fn get_apps_needing_warning(pool: &PgPool, limit: i64) -> Result<Vec<WarningCandidate>> {
633 use super::id_types::UserId;
634
635 // Egress is no longer a price input or an enforced cap (see migration 118),
636 // so the only dimension that warrants a usage warning is storage. The
637 // effective storage cap depends on enforcement_mode.
638 #[derive(sqlx::FromRow)]
639 struct Row {
640 app_id: SyncAppId,
641 creator_id: UserId,
642 creator_email: String,
643 app_name: String,
644 enforcement_mode: super::SyncEnforcementMode,
645 storage_gb_cap: Option<i32>,
646 key_cap: Option<i32>,
647 gb_per_key: Option<i32>,
648 bytes_stored: i64,
649 last_warning_pct: i16,
650 }
651
652 let rows = sqlx::query_as::<_, Row>(
653 r"
654 SELECT
655 sa.id AS app_id,
656 sa.creator_id AS creator_id,
657 u_user.email AS creator_email,
658 sa.name AS app_name,
659 sa.enforcement_mode AS enforcement_mode,
660 sa.storage_gb_cap AS storage_gb_cap,
661 sa.key_cap AS key_cap,
662 sa.gb_per_key AS gb_per_key,
663 COALESCE(u.bytes_stored, 0) AS bytes_stored,
664 COALESCE(u.last_warning_pct, 0::smallint) AS last_warning_pct
665 FROM sync_apps sa
666 JOIN users u_user ON u_user.id = sa.creator_id
667 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
668 WHERE sa.billing_status = 'active'
669 AND sa.is_internal = FALSE
670 ORDER BY sa.id
671 LIMIT $1
672 ",
673 )
674 .bind(limit)
675 .fetch_all(pool)
676 .await?;
677
678 let mut out = Vec::new();
679 let mut per_key_apps: Vec<(SyncAppId, UserId, String, String, i32)> = Vec::new();
680 for r in rows {
681 match r.enforcement_mode {
682 super::SyncEnforcementMode::Bulk => {
683 let Some(gb) = r.storage_gb_cap else { continue };
684 let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
685 if let Some(pct) =
686 highest_breached_threshold(r.bytes_stored, limit, r.last_warning_pct)
687 {
688 out.push(WarningCandidate {
689 app_id: r.app_id,
690 creator_id: r.creator_id,
691 creator_email: r.creator_email,
692 app_name: r.app_name,
693 threshold_pct: pct,
694 dimension: "storage",
695 used: r.bytes_stored,
696 limit,
697 key: None,
698 });
699 }
700 }
701 super::SyncEnforcementMode::PerKey => {
702 // Defer to a second query that fans out per (app, key); the
703 // per-key counter, not the app aggregate, is what we warn on.
704 let (Some(_), Some(g)) = (r.key_cap, r.gb_per_key) else {
705 continue;
706 };
707 per_key_apps.push((r.app_id, r.creator_id, r.creator_email, r.app_name, g));
708 }
709 }
710 }
711
712 if !per_key_apps.is_empty() {
713 #[derive(sqlx::FromRow)]
714 struct KeyRow {
715 app_id: SyncAppId,
716 key: String,
717 bytes_stored: i64,
718 last_warning_pct: i16,
719 }
720 let app_ids: Vec<SyncAppId> = per_key_apps.iter().map(|t| t.0).collect();
721 let key_rows = sqlx::query_as::<_, KeyRow>(
722 r"
723 SELECT u.app_id, u.key, u.bytes_stored, u.last_warning_pct
724 FROM sync_key_usage_current u
725 JOIN sync_app_keys k
726 ON k.app_id = u.app_id AND k.key = u.key AND k.released_at IS NULL
727 WHERE u.app_id = ANY($1)
728 ",
729 )
730 .bind(&app_ids)
731 .fetch_all(pool)
732 .await?;
733
734 // Index app metadata by id for cheap join.
735 let meta: std::collections::HashMap<SyncAppId, (UserId, String, String, i32)> =
736 per_key_apps
737 .into_iter()
738 .map(|(a, c, e, n, g)| (a, (c, e, n, g)))
739 .collect();
740
741 for r in key_rows {
742 let Some((creator_id, creator_email, app_name, gb_per_key)) =
743 meta.get(&r.app_id).cloned()
744 else {
745 continue;
746 };
747 let limit = crate::synckit_billing::storage_cap_bytes(gb_per_key as u32);
748 if let Some(pct) = highest_breached_threshold(r.bytes_stored, limit, r.last_warning_pct)
749 {
750 out.push(WarningCandidate {
751 app_id: r.app_id,
752 creator_id,
753 creator_email,
754 app_name,
755 threshold_pct: pct,
756 dimension: "storage_per_key",
757 used: r.bytes_stored,
758 limit,
759 key: Some(r.key),
760 });
761 }
762 }
763 }
764
765 // Bound the per-tick candidate count (the per-key fan-out can expand beyond
766 // the app LIMIT); stamped apps drop out so the rest drains on later ticks.
767 out.truncate(limit as usize);
768 Ok(out)
769 }
770
771 /// Compute the highest WARNING_THRESHOLDS_PCT band currently exceeded by
772 /// `used / limit` whose value is strictly above `last_warning_pct`.
773 ///
774 /// Returns `None` if no new band has been breached.
775 pub fn highest_breached_threshold(used: i64, limit: i64, last_warning_pct: i16) -> Option<i16> {
776 if limit <= 0 {
777 return None;
778 }
779 // Compute current percentage as integer; saturate at 100+.
780 // Use f64 to avoid 32-bit overflow for very large byte counts.
781 let pct_f = (used as f64 / limit as f64) * 100.0;
782 let pct = pct_f.floor() as i64;
783 crate::synckit_billing::WARNING_THRESHOLDS_PCT
784 .iter()
785 .rev()
786 .copied()
787 .find(|&t| pct >= t as i64 && t > last_warning_pct)
788 }
789
790 /// Stamp `last_warning_pct` to record that a warning at `pct` has fired.
791 #[tracing::instrument(skip_all)]
792 pub async fn update_warning_pct(pool: &PgPool, app_id: SyncAppId, pct: i16) -> Result<()> {
793 sqlx::query(
794 r"
795 UPDATE sync_app_usage_current
796 SET last_warning_pct = $2, updated_at = NOW()
797 WHERE app_id = $1
798 ",
799 )
800 .bind(app_id)
801 .bind(pct)
802 .execute(pool)
803 .await?;
804 Ok(())
805 }
806
807 /// Per-key analogue of `update_warning_pct`. Stamps the band on
808 /// `sync_key_usage_current` so subsequent ticks don't re-fire the same band
809 /// for the same key.
810 #[tracing::instrument(skip_all)]
811 pub async fn update_key_warning_pct(
812 pool: &PgPool,
813 app_id: SyncAppId,
814 key: &str,
815 pct: i16,
816 ) -> Result<()> {
817 sqlx::query(
818 r"
819 UPDATE sync_key_usage_current
820 SET last_warning_pct = $3, updated_at = NOW()
821 WHERE app_id = $1 AND key = $2
822 ",
823 )
824 .bind(app_id)
825 .bind(key)
826 .bind(pct)
827 .execute(pool)
828 .await?;
829 Ok(())
830 }
831
832 /// Recalculate `bytes_stored` from the authoritative `sync_blobs` table for
833 /// every app and every per-key counter. Weekly drift correction. Returns
834 /// total count of rows updated across both tables.
835 #[tracing::instrument(skip_all)]
836 pub async fn recalculate_synckit_app_storage(pool: &PgPool) -> Result<u64> {
837 let mut tx = pool.begin().await?;
838
839 let app_res = sqlx::query(
840 r"
841 UPDATE sync_app_usage_current u
842 SET bytes_stored = COALESCE(s.total, 0), updated_at = NOW()
843 FROM (
844 SELECT app_id, SUM(size_bytes)::BIGINT AS total
845 FROM sync_blobs
846 GROUP BY app_id
847 ) s
848 WHERE u.app_id = s.app_id
849 AND u.bytes_stored <> COALESCE(s.total, 0)
850 ",
851 )
852 .execute(&mut *tx)
853 .await?;
854
855 // Per-key reconciliation: upsert every (app_id, key) found in sync_blobs.
856 // Keys that disappear (all blobs deleted) aren't pruned here, bytes_stored
857 // would just stay at whatever it last drifted to. Blob delete isn't shipped
858 // yet, so this is fine. When it ships, add a step that resets rows whose
859 // sync_blobs total is zero (or just delete them, the upsert recreates).
860 let key_res = sqlx::query(
861 r"
862 INSERT INTO sync_key_usage_current (app_id, key, bytes_stored, updated_at)
863 SELECT app_id, key, SUM(size_bytes)::BIGINT, NOW()
864 FROM sync_blobs
865 GROUP BY app_id, key
866 ON CONFLICT (app_id, key)
867 DO UPDATE SET
868 bytes_stored = EXCLUDED.bytes_stored,
869 updated_at = NOW()
870 WHERE sync_key_usage_current.bytes_stored <> EXCLUDED.bytes_stored
871 ",
872 )
873 .execute(&mut *tx)
874 .await?;
875
876 tx.commit().await?;
877 Ok(app_res.rows_affected() + key_res.rows_affected())
878 }
879