Skip to main content

max / makenotwork

29.8 KB · 880 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`, the synckit-billing arm of
182 /// the epoch-period bug class (CHRONIC C).
183 pub async fn apply_billing_update<'e>(
184 executor: impl sqlx::PgExecutor<'e>,
185 app_id: SyncAppId,
186 status: Option<&str>,
187 period: Option<(i64, i64)>,
188 ) -> Result<bool> {
189 let (period_start, period_end) = match period {
190 // Require a positive, non-inverted window. A non-positive `end` is the
191 // thin/zero-webhook guard (no 1970 period); `start <= end` additionally
192 // rejects an inverted range, so a malformed Stripe period writes nothing
193 // (the `COALESCE` keeps the existing values) rather than stamping an
194 // end-before-start window onto a live app.
195 Some((start, end)) if end > 0 && start <= end => (
196 DateTime::<Utc>::from_timestamp(start, 0),
197 DateTime::<Utc>::from_timestamp(end, 0),
198 ),
199 _ => (None, None),
200 };
201 let result = sqlx::query(
202 r"
203 UPDATE sync_apps SET
204 billing_status = COALESCE($2, billing_status),
205 current_period_start = COALESCE($3, current_period_start),
206 current_period_end = COALESCE($4, current_period_end)
207 WHERE id = $1
208 AND (billing_status != 'canceled' OR $2 = 'canceled')
209 ",
210 )
211 .bind(app_id)
212 .bind(status)
213 .bind(period_start)
214 .bind(period_end)
215 .execute(executor)
216 .await?;
217 Ok(result.rows_affected() > 0)
218 }
219
220 /// Reset the per-period usage counters on `sync_app_usage_current`. Called
221 /// from the `invoice.paid` webhook handler at period rollover.
222 #[tracing::instrument(skip_all)]
223 pub async fn reset_period_usage<'e>(
224 executor: impl sqlx::PgExecutor<'e>,
225 app_id: SyncAppId,
226 ) -> Result<()> {
227 sqlx::query(
228 r"
229 UPDATE sync_app_usage_current SET
230 bytes_egress_period = 0,
231 last_warning_pct = 0,
232 period_started_at = NOW(),
233 updated_at = NOW()
234 WHERE app_id = $1
235 ",
236 )
237 .bind(app_id)
238 .execute(executor)
239 .await?;
240 Ok(())
241 }
242
243 /// Look up the sync app that owns a given Stripe subscription. Used by the
244 /// webhook router to distinguish SyncKit v2 subscriptions from
245 /// creator-tier / Fan+ subscriptions.
246 #[tracing::instrument(skip_all)]
247 pub async fn get_app_by_stripe_subscription(
248 pool: &PgPool,
249 stripe_sub_id: &str,
250 ) -> Result<Option<SyncAppId>> {
251 let row: Option<(SyncAppId,)> =
252 sqlx::query_as("SELECT id FROM sync_apps WHERE stripe_subscription_id = $1")
253 .bind(stripe_sub_id)
254 .fetch_optional(pool)
255 .await?;
256 Ok(row.map(|(id,)| id))
257 }
258
259 /// Fetch the combined app+billing+usage view for a single app. `egress_multiple`
260 /// is cast from NUMERIC to DOUBLE PRECISION so it decodes into `f64` without
261 /// the `bigdecimal` sqlx feature. `sync_app_usage_current` is LEFT-joined: the
262 /// usage row is created transactionally with the app (`create_sync_app`) and
263 /// backfilled by migration 165, but the `Option`/LEFT JOIN stays as
264 /// defense-in-depth so a missing row can't blow up the read path.
265 #[tracing::instrument(skip_all)]
266 pub async fn get_app_with_billing(
267 pool: &PgPool,
268 app_id: SyncAppId,
269 ) -> Result<Option<DbSyncAppBilling>> {
270 let app = sqlx::query_as::<_, DbSyncAppBilling>(
271 r"
272 SELECT
273 sa.id,
274 sa.creator_id,
275 sa.name,
276 sa.is_internal,
277 sa.stripe_customer_id,
278 sa.stripe_subscription_id,
279 sa.billing_status,
280 sa.storage_gb_cap,
281 sa.enforcement_mode,
282 sa.key_cap,
283 sa.gb_per_key,
284 sa.current_period_start,
285 sa.current_period_end,
286 u.bytes_stored,
287 u.bytes_egress_period,
288 u.keys_claimed,
289 u.last_warning_pct,
290 u.period_started_at,
291 p.slug AS project_slug
292 FROM sync_apps sa
293 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
294 LEFT JOIN projects p ON p.id = sa.project_id
295 WHERE sa.id = $1
296 ",
297 )
298 .bind(app_id)
299 .fetch_optional(pool)
300 .await?;
301 Ok(app)
302 }
303
304 /// Batch variant of `get_app_with_billing` that loads every app owned by a
305 /// creator with its billing+usage join in one query. Used by the user-level
306 /// SyncKit dashboard tab.
307 #[tracing::instrument(skip_all)]
308 pub async fn get_apps_with_billing_by_creator(
309 pool: &PgPool,
310 creator_id: super::id_types::UserId,
311 ) -> Result<Vec<DbSyncAppBilling>> {
312 let apps = sqlx::query_as::<_, DbSyncAppBilling>(
313 r"
314 SELECT
315 sa.id,
316 sa.creator_id,
317 sa.name,
318 sa.is_internal,
319 sa.stripe_customer_id,
320 sa.stripe_subscription_id,
321 sa.billing_status,
322 sa.storage_gb_cap,
323 sa.enforcement_mode,
324 sa.key_cap,
325 sa.gb_per_key,
326 sa.current_period_start,
327 sa.current_period_end,
328 u.bytes_stored,
329 u.bytes_egress_period,
330 u.keys_claimed,
331 u.last_warning_pct,
332 u.period_started_at,
333 p.slug AS project_slug
334 FROM sync_apps sa
335 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
336 LEFT JOIN projects p ON p.id = sa.project_id
337 WHERE sa.creator_id = $1
338 ",
339 )
340 .bind(creator_id)
341 .fetch_all(pool)
342 .await?;
343 Ok(apps)
344 }
345
346 /// Batch variant of `get_app_with_billing` for one project. Used by the
347 /// project-level SyncKit dashboard tab.
348 #[tracing::instrument(skip_all)]
349 pub async fn get_apps_with_billing_by_project(
350 pool: &PgPool,
351 project_id: super::id_types::ProjectId,
352 ) -> Result<Vec<DbSyncAppBilling>> {
353 let apps = sqlx::query_as::<_, DbSyncAppBilling>(
354 r"
355 SELECT
356 sa.id,
357 sa.creator_id,
358 sa.name,
359 sa.is_internal,
360 sa.stripe_customer_id,
361 sa.stripe_subscription_id,
362 sa.billing_status,
363 sa.storage_gb_cap,
364 sa.enforcement_mode,
365 sa.key_cap,
366 sa.gb_per_key,
367 sa.current_period_start,
368 sa.current_period_end,
369 u.bytes_stored,
370 u.bytes_egress_period,
371 u.keys_claimed,
372 u.last_warning_pct,
373 u.period_started_at,
374 p.slug AS project_slug
375 FROM sync_apps sa
376 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
377 LEFT JOIN projects p ON p.id = sa.project_id
378 WHERE sa.project_id = $1
379 ",
380 )
381 .bind(project_id)
382 .fetch_all(pool)
383 .await?;
384 Ok(apps)
385 }
386
387 /// Claim an SDK encryption key for an app. Idempotent: a re-claim of an
388 /// already-active key returns `newly_claimed = false` without inserting.
389 ///
390 /// `key_cap` is the per-app active-key ceiling (`Some` only for `per_key`
391 /// developer apps; `None` means uncapped). The cap is checked **inside** the
392 /// transaction, under the `FOR UPDATE` lock on `sync_app_usage_current`, so
393 /// concurrent claims of distinct new keys can't collectively overshoot the cap
394 ///, the prior design checked the cap pre-transaction in the handler and could
395 /// over-allocate. A re-claim of an already-active key is admitted regardless of
396 /// the cap (it consumes no new slot).
397 #[tracing::instrument(skip_all)]
398 pub async fn claim_key(
399 pool: &sqlx::PgPool,
400 app_id: SyncAppId,
401 key: &str,
402 key_cap: Option<i32>,
403 ) -> Result<ClaimResult> {
404 let mut tx = pool.begin().await?;
405
406 // Lock the usage row for this app. Returns the current keys_claimed.
407 let (mut keys_claimed,): (i32,) = sqlx::query_as(
408 "SELECT keys_claimed FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
409 )
410 .bind(app_id)
411 .fetch_one(&mut *tx)
412 .await?;
413
414 // Is there already an active claim for this key?
415 let existing: Option<(uuid::Uuid,)> = sqlx::query_as(
416 "SELECT id FROM sync_app_keys
417 WHERE app_id = $1 AND key = $2 AND released_at IS NULL",
418 )
419 .bind(app_id)
420 .bind(key)
421 .fetch_optional(&mut *tx)
422 .await?;
423
424 if existing.is_some() {
425 // Idempotent re-claim, no new slot, cap not consulted.
426 tx.commit().await?;
427 return Ok(ClaimResult {
428 newly_claimed: false,
429 cap_reached: false,
430 total_claimed: keys_claimed,
431 });
432 }
433
434 // New claim: enforce the cap under the lock.
435 if let Some(cap) = key_cap
436 && keys_claimed >= cap
437 {
438 tx.commit().await?;
439 return Ok(ClaimResult {
440 newly_claimed: false,
441 cap_reached: true,
442 total_claimed: keys_claimed,
443 });
444 }
445
446 sqlx::query("INSERT INTO sync_app_keys (app_id, key) VALUES ($1, $2)")
447 .bind(app_id)
448 .bind(key)
449 .execute(&mut *tx)
450 .await?;
451 sqlx::query(
452 "UPDATE sync_app_usage_current
453 SET keys_claimed = keys_claimed + 1, updated_at = NOW()
454 WHERE app_id = $1",
455 )
456 .bind(app_id)
457 .execute(&mut *tx)
458 .await?;
459 keys_claimed += 1;
460
461 tx.commit().await?;
462 Ok(ClaimResult {
463 newly_claimed: true,
464 cap_reached: false,
465 total_claimed: keys_claimed,
466 })
467 }
468
469 /// Release an SDK encryption key. Idempotent: releasing a key that is not
470 /// actively claimed returns `newly_released = false`.
471 #[tracing::instrument(skip_all)]
472 pub async fn release_key(
473 pool: &sqlx::PgPool,
474 app_id: SyncAppId,
475 key: &str,
476 ) -> Result<ReleaseResult> {
477 let mut tx = pool.begin().await?;
478
479 let (mut keys_claimed,): (i32,) = sqlx::query_as(
480 "SELECT keys_claimed FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
481 )
482 .bind(app_id)
483 .fetch_one(&mut *tx)
484 .await?;
485
486 let released: Option<(uuid::Uuid,)> = sqlx::query_as(
487 "UPDATE sync_app_keys SET released_at = NOW()
488 WHERE app_id = $1 AND key = $2 AND released_at IS NULL
489 RETURNING id",
490 )
491 .bind(app_id)
492 .bind(key)
493 .fetch_optional(&mut *tx)
494 .await?;
495
496 let newly_released = if released.is_some() {
497 sqlx::query(
498 "UPDATE sync_app_usage_current
499 SET keys_claimed = GREATEST(keys_claimed - 1, 0), updated_at = NOW()
500 WHERE app_id = $1",
501 )
502 .bind(app_id)
503 .execute(&mut *tx)
504 .await?;
505 keys_claimed = (keys_claimed - 1).max(0);
506 true
507 } else {
508 false
509 };
510
511 tx.commit().await?;
512 Ok(ReleaseResult {
513 newly_released,
514 total_claimed: keys_claimed,
515 })
516 }
517
518 /// List active (un-released) key claims for an app, ordered by `claimed_at DESC`.
519 /// Used by the dashboard "Active keys" view. `bytes_stored` is LEFT-joined
520 /// from `sync_key_usage_current`, `0` when no upload has landed yet.
521 #[tracing::instrument(skip_all)]
522 pub async fn list_active_keys(
523 pool: &sqlx::PgPool,
524 app_id: SyncAppId,
525 limit: i64,
526 offset: i64,
527 ) -> Result<Vec<DbSyncAppKey>> {
528 let rows = sqlx::query_as::<_, DbSyncAppKey>(
529 r"
530 SELECT k.id, k.key, k.claimed_at,
531 COALESCE(u.bytes_stored, 0) AS bytes_stored
532 FROM sync_app_keys k
533 LEFT JOIN sync_key_usage_current u
534 ON u.app_id = k.app_id AND u.key = k.key
535 WHERE k.app_id = $1 AND k.released_at IS NULL
536 ORDER BY k.claimed_at DESC
537 LIMIT $2 OFFSET $3
538 ",
539 )
540 .bind(app_id)
541 .bind(limit)
542 .bind(offset)
543 .fetch_all(pool)
544 .await?;
545 Ok(rows)
546 }
547
548 /// Per-app top-N key usage for the dashboard panel. Returns the highest-usage
549 /// active keys (sorted by `bytes_stored DESC`) for each app in `app_ids`,
550 /// batched into a single query so the integrations page doesn't N+1 across
551 /// every per_key-mode app.
552 #[tracing::instrument(skip_all)]
553 pub async fn get_top_keys_per_app(
554 pool: &PgPool,
555 app_ids: &[SyncAppId],
556 limit_per_app: i64,
557 ) -> Result<Vec<(SyncAppId, String, i64)>> {
558 if app_ids.is_empty() {
559 return Ok(Vec::new());
560 }
561 let rows: Vec<(SyncAppId, String, i64)> = sqlx::query_as(
562 r"
563 SELECT app_id, key, bytes_stored
564 FROM (
565 SELECT u.app_id, u.key, u.bytes_stored,
566 ROW_NUMBER() OVER (PARTITION BY u.app_id ORDER BY u.bytes_stored DESC, u.key) AS rn
567 FROM sync_key_usage_current u
568 JOIN sync_app_keys k
569 ON k.app_id = u.app_id AND k.key = u.key AND k.released_at IS NULL
570 WHERE u.app_id = ANY($1)
571 ) ranked
572 WHERE rn <= $2
573 ORDER BY app_id, rn
574 ",
575 )
576 .bind(app_ids)
577 .bind(limit_per_app)
578 .fetch_all(pool)
579 .await?;
580 Ok(rows)
581 }
582
583 // ── Phase 5: Usage counters and cap enforcement ──
584
585 /// A row that may need a warning email sent.
586 ///
587 /// `threshold_pct` is the highest WARNING_THRESHOLDS_PCT band currently
588 /// breached above `last_warning_pct`. The caller is responsible for filtering
589 /// out apps where no new breach has occurred (i.e. usage hasn't crossed the
590 /// next threshold since the previous warning).
591 #[derive(Debug, Clone, PartialEq, Eq)]
592 pub struct WarningCandidate {
593 pub app_id: SyncAppId,
594 pub creator_id: super::id_types::UserId,
595 pub creator_email: String,
596 pub app_name: String,
597 pub threshold_pct: i16,
598 pub dimension: &'static str,
599 pub used: i64,
600 pub limit: i64,
601 /// SDK key that breached the threshold, when `dimension ==
602 /// "storage_per_key"`. `None` for app-wide breaches.
603 pub key: Option<String>,
604 }
605
606 /// Increment (or decrement) the per-period `bytes_egress_period` counter.
607 /// Returns the new value.
608 #[tracing::instrument(skip_all)]
609 pub async fn add_bytes_egress(pool: &PgPool, app_id: SyncAppId, delta: i64) -> Result<i64> {
610 let (new_val,): (i64,) = sqlx::query_as(
611 r"
612 UPDATE sync_app_usage_current
613 SET bytes_egress_period = GREATEST(bytes_egress_period + $2, 0), updated_at = NOW()
614 WHERE app_id = $1
615 RETURNING bytes_egress_period
616 ",
617 )
618 .bind(app_id)
619 .bind(delta)
620 .fetch_one(pool)
621 .await?;
622 Ok(new_val)
623 }
624
625 /// Fetch the list of active, non-internal apps that may need a warning email,
626 /// along with the data needed to compute which threshold (if any) has been
627 /// breached since the last notice. The per-app breach computation is done in
628 /// Rust by the caller (see `scheduler::synckit_warnings`).
629 /// `limit` bounds the app scan so a single scheduler tick does a bounded amount
630 /// of work; warned apps stamp `last_warning_pct` and drop out of the candidate
631 /// set, so any overflow drains on subsequent ticks.
632 #[tracing::instrument(skip_all)]
633 pub async fn get_apps_needing_warning(pool: &PgPool, limit: i64) -> Result<Vec<WarningCandidate>> {
634 use super::id_types::UserId;
635
636 // Egress is no longer a price input or an enforced cap (see migration 118),
637 // so the only dimension that warrants a usage warning is storage. The
638 // effective storage cap depends on enforcement_mode.
639 #[derive(sqlx::FromRow)]
640 struct Row {
641 app_id: SyncAppId,
642 creator_id: UserId,
643 creator_email: String,
644 app_name: String,
645 enforcement_mode: super::SyncEnforcementMode,
646 storage_gb_cap: Option<i32>,
647 key_cap: Option<i32>,
648 gb_per_key: Option<i32>,
649 bytes_stored: i64,
650 last_warning_pct: i16,
651 }
652
653 let rows = sqlx::query_as::<_, Row>(
654 r"
655 SELECT
656 sa.id AS app_id,
657 sa.creator_id AS creator_id,
658 u_user.email AS creator_email,
659 sa.name AS app_name,
660 sa.enforcement_mode AS enforcement_mode,
661 sa.storage_gb_cap AS storage_gb_cap,
662 sa.key_cap AS key_cap,
663 sa.gb_per_key AS gb_per_key,
664 COALESCE(u.bytes_stored, 0) AS bytes_stored,
665 COALESCE(u.last_warning_pct, 0::smallint) AS last_warning_pct
666 FROM sync_apps sa
667 JOIN users u_user ON u_user.id = sa.creator_id
668 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
669 WHERE sa.billing_status = 'active'
670 AND sa.is_internal = FALSE
671 ORDER BY sa.id
672 LIMIT $1
673 ",
674 )
675 .bind(limit)
676 .fetch_all(pool)
677 .await?;
678
679 let mut out = Vec::new();
680 let mut per_key_apps: Vec<(SyncAppId, UserId, String, String, i32)> = Vec::new();
681 for r in rows {
682 match r.enforcement_mode {
683 super::SyncEnforcementMode::Bulk => {
684 let Some(gb) = r.storage_gb_cap else { continue };
685 let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
686 if let Some(pct) =
687 highest_breached_threshold(r.bytes_stored, limit, r.last_warning_pct)
688 {
689 out.push(WarningCandidate {
690 app_id: r.app_id,
691 creator_id: r.creator_id,
692 creator_email: r.creator_email,
693 app_name: r.app_name,
694 threshold_pct: pct,
695 dimension: "storage",
696 used: r.bytes_stored,
697 limit,
698 key: None,
699 });
700 }
701 }
702 super::SyncEnforcementMode::PerKey => {
703 // Defer to a second query that fans out per (app, key); the
704 // per-key counter, not the app aggregate, is what we warn on.
705 let (Some(_), Some(g)) = (r.key_cap, r.gb_per_key) else {
706 continue;
707 };
708 per_key_apps.push((r.app_id, r.creator_id, r.creator_email, r.app_name, g));
709 }
710 }
711 }
712
713 if !per_key_apps.is_empty() {
714 #[derive(sqlx::FromRow)]
715 struct KeyRow {
716 app_id: SyncAppId,
717 key: String,
718 bytes_stored: i64,
719 last_warning_pct: i16,
720 }
721 let app_ids: Vec<SyncAppId> = per_key_apps.iter().map(|t| t.0).collect();
722 let key_rows = sqlx::query_as::<_, KeyRow>(
723 r"
724 SELECT u.app_id, u.key, u.bytes_stored, u.last_warning_pct
725 FROM sync_key_usage_current u
726 JOIN sync_app_keys k
727 ON k.app_id = u.app_id AND k.key = u.key AND k.released_at IS NULL
728 WHERE u.app_id = ANY($1)
729 ",
730 )
731 .bind(&app_ids)
732 .fetch_all(pool)
733 .await?;
734
735 // Index app metadata by id for cheap join.
736 let meta: std::collections::HashMap<SyncAppId, (UserId, String, String, i32)> =
737 per_key_apps
738 .into_iter()
739 .map(|(a, c, e, n, g)| (a, (c, e, n, g)))
740 .collect();
741
742 for r in key_rows {
743 let Some((creator_id, creator_email, app_name, gb_per_key)) =
744 meta.get(&r.app_id).cloned()
745 else {
746 continue;
747 };
748 let limit = crate::synckit_billing::storage_cap_bytes(gb_per_key as u32);
749 if let Some(pct) = highest_breached_threshold(r.bytes_stored, limit, r.last_warning_pct)
750 {
751 out.push(WarningCandidate {
752 app_id: r.app_id,
753 creator_id,
754 creator_email,
755 app_name,
756 threshold_pct: pct,
757 dimension: "storage_per_key",
758 used: r.bytes_stored,
759 limit,
760 key: Some(r.key),
761 });
762 }
763 }
764 }
765
766 // Bound the per-tick candidate count (the per-key fan-out can expand beyond
767 // the app LIMIT); stamped apps drop out so the rest drains on later ticks.
768 out.truncate(limit as usize);
769 Ok(out)
770 }
771
772 /// Compute the highest WARNING_THRESHOLDS_PCT band currently exceeded by
773 /// `used / limit` whose value is strictly above `last_warning_pct`.
774 ///
775 /// Returns `None` if no new band has been breached.
776 pub fn highest_breached_threshold(used: i64, limit: i64, last_warning_pct: i16) -> Option<i16> {
777 if limit <= 0 {
778 return None;
779 }
780 // Compute current percentage as integer; saturate at 100+.
781 // Use f64 to avoid 32-bit overflow for very large byte counts.
782 let pct_f = (used as f64 / limit as f64) * 100.0;
783 let pct = pct_f.floor() as i64;
784 crate::synckit_billing::WARNING_THRESHOLDS_PCT
785 .iter()
786 .rev()
787 .copied()
788 .find(|&t| pct >= t as i64 && t > last_warning_pct)
789 }
790
791 /// Stamp `last_warning_pct` to record that a warning at `pct` has fired.
792 #[tracing::instrument(skip_all)]
793 pub async fn update_warning_pct(pool: &PgPool, app_id: SyncAppId, pct: i16) -> Result<()> {
794 sqlx::query(
795 r"
796 UPDATE sync_app_usage_current
797 SET last_warning_pct = $2, updated_at = NOW()
798 WHERE app_id = $1
799 ",
800 )
801 .bind(app_id)
802 .bind(pct)
803 .execute(pool)
804 .await?;
805 Ok(())
806 }
807
808 /// Per-key analogue of `update_warning_pct`. Stamps the band on
809 /// `sync_key_usage_current` so subsequent ticks don't re-fire the same band
810 /// for the same key.
811 #[tracing::instrument(skip_all)]
812 pub async fn update_key_warning_pct(
813 pool: &PgPool,
814 app_id: SyncAppId,
815 key: &str,
816 pct: i16,
817 ) -> Result<()> {
818 sqlx::query(
819 r"
820 UPDATE sync_key_usage_current
821 SET last_warning_pct = $3, updated_at = NOW()
822 WHERE app_id = $1 AND key = $2
823 ",
824 )
825 .bind(app_id)
826 .bind(key)
827 .bind(pct)
828 .execute(pool)
829 .await?;
830 Ok(())
831 }
832
833 /// Recalculate `bytes_stored` from the authoritative `sync_blobs` table for
834 /// every app and every per-key counter. Weekly drift correction. Returns
835 /// total count of rows updated across both tables.
836 #[tracing::instrument(skip_all)]
837 pub async fn recalculate_synckit_app_storage(pool: &PgPool) -> Result<u64> {
838 let mut tx = pool.begin().await?;
839
840 let app_res = sqlx::query(
841 r"
842 UPDATE sync_app_usage_current u
843 SET bytes_stored = COALESCE(s.total, 0), updated_at = NOW()
844 FROM (
845 SELECT app_id, SUM(size_bytes)::BIGINT AS total
846 FROM sync_blobs
847 GROUP BY app_id
848 ) s
849 WHERE u.app_id = s.app_id
850 AND u.bytes_stored <> COALESCE(s.total, 0)
851 ",
852 )
853 .execute(&mut *tx)
854 .await?;
855
856 // Per-key reconciliation: upsert every (app_id, key) found in sync_blobs.
857 // Keys that disappear (all blobs deleted) aren't pruned here, bytes_stored
858 // would just stay at whatever it last drifted to. Blob delete isn't shipped
859 // yet, so this is fine. When it ships, add a step that resets rows whose
860 // sync_blobs total is zero (or just delete them, the upsert recreates).
861 let key_res = sqlx::query(
862 r"
863 INSERT INTO sync_key_usage_current (app_id, key, bytes_stored, updated_at)
864 SELECT app_id, key, SUM(size_bytes)::BIGINT, NOW()
865 FROM sync_blobs
866 GROUP BY app_id, key
867 ON CONFLICT (app_id, key)
868 DO UPDATE SET
869 bytes_stored = EXCLUDED.bytes_stored,
870 updated_at = NOW()
871 WHERE sync_key_usage_current.bytes_stored <> EXCLUDED.bytes_stored
872 ",
873 )
874 .execute(&mut *tx)
875 .await?;
876
877 tx.commit().await?;
878 Ok(app_res.rows_affected() + key_res.rows_affected())
879 }
880