Skip to main content

max / makenotwork

32.0 KB · 1005 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 //!
6 //! Schema reference (kept here for offline-mode reviewers since the migration
7 //! is not yet applied at the time of writing):
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','app_wide')
19 //! key_cap INT
20 //! current_period_start TIMESTAMPTZ
21 //! current_period_end TIMESTAMPTZ
22 //!
23 //! sync_app_usage_current (one row per app):
24 //! app_id, bytes_stored, bytes_egress_period, keys_claimed,
25 //! last_warning_pct, period_started_at, updated_at
26 //! ```
27
28 use chrono::{DateTime, Utc};
29 use sqlx::PgPool;
30
31 use super::id_types::SyncAppId;
32 use super::models::{DbSyncAppBilling, DbSyncAppKey};
33 use crate::error::Result;
34
35 /// Outcome of a `claim_key` call.
36 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
37 pub struct ClaimResult {
38 /// `true` if this call inserted a new active claim row;
39 /// `false` if the key was already actively claimed (idempotent re-claim).
40 pub newly_claimed: bool,
41 /// Total active claims for this app after the operation.
42 pub total_claimed: i32,
43 }
44
45 /// Outcome of a `release_key` call.
46 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
47 pub struct ReleaseResult {
48 /// `true` if this call transitioned an active row to released;
49 /// `false` if no active row existed (idempotent release).
50 pub newly_released: bool,
51 /// Total active claims for this app after the operation.
52 pub total_claimed: i32,
53 }
54
55 /// Set the Stripe customer ID on a sync app (idempotent; re-sets the same id
56 /// without error). Called by the billing setup route once the developer-side
57 /// Customer object has been created in Stripe.
58 #[tracing::instrument(skip_all)]
59 pub async fn set_stripe_customer(
60 pool: &PgPool,
61 app_id: SyncAppId,
62 stripe_customer_id: &str,
63 ) -> Result<()> {
64 sqlx::query("UPDATE sync_apps SET stripe_customer_id = $2 WHERE id = $1")
65 .bind(app_id)
66 .bind(stripe_customer_id)
67 .execute(pool)
68 .await?;
69 Ok(())
70 }
71
72 /// Activate billing on a draft app: stamps subscription id, knobs, status,
73 /// and current period. The `egress_multiple` is passed in as `f64` and cast
74 /// to the column's `NUMERIC(6,2)` type in SQL.
75 #[tracing::instrument(skip_all)]
76 #[allow(clippy::too_many_arguments)]
77 pub async fn activate_billing(
78 pool: &PgPool,
79 app_id: SyncAppId,
80 enforcement_mode: &str,
81 storage_gb_cap: Option<i32>,
82 key_cap: Option<i32>,
83 gb_per_key: Option<i32>,
84 stripe_sub_id: &str,
85 period_start: DateTime<Utc>,
86 period_end: DateTime<Utc>,
87 ) -> Result<()> {
88 sqlx::query(
89 r#"
90 UPDATE sync_apps SET
91 billing_status = 'active',
92 stripe_subscription_id = $2,
93 enforcement_mode = $3,
94 storage_gb_cap = $4,
95 key_cap = $5,
96 gb_per_key = $6,
97 current_period_start = $7,
98 current_period_end = $8
99 WHERE id = $1
100 "#,
101 )
102 .bind(app_id)
103 .bind(stripe_sub_id)
104 .bind(enforcement_mode)
105 .bind(storage_gb_cap)
106 .bind(key_cap)
107 .bind(gb_per_key)
108 .bind(period_start)
109 .bind(period_end)
110 .execute(pool)
111 .await?;
112 Ok(())
113 }
114
115 /// Update the pricing knobs on an already-active app (no status change).
116 #[tracing::instrument(skip_all)]
117 pub async fn update_knobs(
118 pool: &PgPool,
119 app_id: SyncAppId,
120 enforcement_mode: &str,
121 storage_gb_cap: Option<i32>,
122 key_cap: Option<i32>,
123 gb_per_key: Option<i32>,
124 ) -> Result<()> {
125 sqlx::query(
126 r#"
127 UPDATE sync_apps SET
128 enforcement_mode = $2,
129 storage_gb_cap = $3,
130 key_cap = $4,
131 gb_per_key = $5
132 WHERE id = $1
133 "#,
134 )
135 .bind(app_id)
136 .bind(enforcement_mode)
137 .bind(storage_gb_cap)
138 .bind(key_cap)
139 .bind(gb_per_key)
140 .execute(pool)
141 .await?;
142 Ok(())
143 }
144
145 /// Set `billing_status` directly (used by webhook handlers for
146 /// suspended_unpaid / canceled transitions).
147 #[tracing::instrument(skip_all)]
148 pub async fn set_billing_status<'e>(
149 executor: impl sqlx::PgExecutor<'e>,
150 app_id: SyncAppId,
151 status: &str,
152 ) -> Result<()> {
153 sqlx::query("UPDATE sync_apps SET billing_status = $2 WHERE id = $1")
154 .bind(app_id)
155 .bind(status)
156 .execute(executor)
157 .await?;
158 Ok(())
159 }
160
161 /// Set the current period bounds (called by `invoice.paid` handler).
162 #[tracing::instrument(skip_all)]
163 pub async fn set_period<'e>(
164 executor: impl sqlx::PgExecutor<'e>,
165 app_id: SyncAppId,
166 start: DateTime<Utc>,
167 end: DateTime<Utc>,
168 ) -> Result<()> {
169 sqlx::query(
170 r#"
171 UPDATE sync_apps SET
172 current_period_start = $2,
173 current_period_end = $3
174 WHERE id = $1
175 "#,
176 )
177 .bind(app_id)
178 .bind(start)
179 .bind(end)
180 .execute(executor)
181 .await?;
182 Ok(())
183 }
184
185 /// Reset the per-period usage counters on `sync_app_usage_current`. Called
186 /// from the `invoice.paid` webhook handler at period rollover.
187 #[tracing::instrument(skip_all)]
188 pub async fn reset_period_usage<'e>(
189 executor: impl sqlx::PgExecutor<'e>,
190 app_id: SyncAppId,
191 ) -> Result<()> {
192 sqlx::query(
193 r#"
194 UPDATE sync_app_usage_current SET
195 bytes_egress_period = 0,
196 last_warning_pct = 0,
197 period_started_at = NOW(),
198 updated_at = NOW()
199 WHERE app_id = $1
200 "#,
201 )
202 .bind(app_id)
203 .execute(executor)
204 .await?;
205 Ok(())
206 }
207
208 /// Look up the sync app that owns a given Stripe subscription. Used by the
209 /// webhook router to distinguish SyncKit v2 subscriptions from
210 /// creator-tier / Fan+ subscriptions.
211 #[tracing::instrument(skip_all)]
212 pub async fn get_app_by_stripe_subscription(
213 pool: &PgPool,
214 stripe_sub_id: &str,
215 ) -> Result<Option<SyncAppId>> {
216 let row: Option<(SyncAppId,)> = sqlx::query_as(
217 "SELECT id FROM sync_apps WHERE stripe_subscription_id = $1",
218 )
219 .bind(stripe_sub_id)
220 .fetch_optional(pool)
221 .await?;
222 Ok(row.map(|(id,)| id))
223 }
224
225 /// Fetch the combined app+billing+usage view for a single app. `egress_multiple`
226 /// is cast from NUMERIC to DOUBLE PRECISION so it decodes into `f64` without
227 /// the `bigdecimal` sqlx feature. `sync_app_usage_current` is LEFT-joined; /// the row is normally inserted on app create (see migration 117), but
228 /// `Option` is used so a missing row doesn't blow up the query.
229 #[tracing::instrument(skip_all)]
230 pub async fn get_app_with_billing(
231 pool: &PgPool,
232 app_id: SyncAppId,
233 ) -> Result<Option<DbSyncAppBilling>> {
234 let app = sqlx::query_as::<_, DbSyncAppBilling>(
235 r#"
236 SELECT
237 sa.id,
238 sa.creator_id,
239 sa.name,
240 sa.is_internal,
241 sa.stripe_customer_id,
242 sa.stripe_subscription_id,
243 sa.billing_status,
244 sa.storage_gb_cap,
245 sa.enforcement_mode,
246 sa.key_cap,
247 sa.gb_per_key,
248 sa.current_period_start,
249 sa.current_period_end,
250 u.bytes_stored,
251 u.bytes_egress_period,
252 u.keys_claimed,
253 u.last_warning_pct,
254 u.period_started_at,
255 p.slug AS project_slug
256 FROM sync_apps sa
257 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
258 LEFT JOIN projects p ON p.id = sa.project_id
259 WHERE sa.id = $1
260 "#,
261 )
262 .bind(app_id)
263 .fetch_optional(pool)
264 .await?;
265 Ok(app)
266 }
267
268 /// Batch variant of `get_app_with_billing` that loads every app owned by a
269 /// creator with its billing+usage join in one query. Used by the user-level
270 /// SyncKit dashboard tab.
271 #[tracing::instrument(skip_all)]
272 pub async fn get_apps_with_billing_by_creator(
273 pool: &PgPool,
274 creator_id: super::id_types::UserId,
275 ) -> Result<Vec<DbSyncAppBilling>> {
276 let apps = sqlx::query_as::<_, DbSyncAppBilling>(
277 r#"
278 SELECT
279 sa.id,
280 sa.creator_id,
281 sa.name,
282 sa.is_internal,
283 sa.stripe_customer_id,
284 sa.stripe_subscription_id,
285 sa.billing_status,
286 sa.storage_gb_cap,
287 sa.enforcement_mode,
288 sa.key_cap,
289 sa.gb_per_key,
290 sa.current_period_start,
291 sa.current_period_end,
292 u.bytes_stored,
293 u.bytes_egress_period,
294 u.keys_claimed,
295 u.last_warning_pct,
296 u.period_started_at,
297 p.slug AS project_slug
298 FROM sync_apps sa
299 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
300 LEFT JOIN projects p ON p.id = sa.project_id
301 WHERE sa.creator_id = $1
302 "#,
303 )
304 .bind(creator_id)
305 .fetch_all(pool)
306 .await?;
307 Ok(apps)
308 }
309
310 /// Batch variant of `get_app_with_billing` for one project. Used by the
311 /// project-level SyncKit dashboard tab.
312 #[tracing::instrument(skip_all)]
313 pub async fn get_apps_with_billing_by_project(
314 pool: &PgPool,
315 project_id: super::id_types::ProjectId,
316 ) -> Result<Vec<DbSyncAppBilling>> {
317 let apps = sqlx::query_as::<_, DbSyncAppBilling>(
318 r#"
319 SELECT
320 sa.id,
321 sa.creator_id,
322 sa.name,
323 sa.is_internal,
324 sa.stripe_customer_id,
325 sa.stripe_subscription_id,
326 sa.billing_status,
327 sa.storage_gb_cap,
328 sa.enforcement_mode,
329 sa.key_cap,
330 sa.gb_per_key,
331 sa.current_period_start,
332 sa.current_period_end,
333 u.bytes_stored,
334 u.bytes_egress_period,
335 u.keys_claimed,
336 u.last_warning_pct,
337 u.period_started_at,
338 p.slug AS project_slug
339 FROM sync_apps sa
340 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
341 LEFT JOIN projects p ON p.id = sa.project_id
342 WHERE sa.project_id = $1
343 "#,
344 )
345 .bind(project_id)
346 .fetch_all(pool)
347 .await?;
348 Ok(apps)
349 }
350
351 /// Claim an SDK encryption key for an app. Idempotent: a re-claim of an
352 /// already-active key returns `newly_claimed = false` without inserting.
353 ///
354 /// The transaction locks `sync_app_usage_current` for this app first, so the
355 /// route handler can read `keys_claimed` and decide on the cap before this
356 /// runs without races. (The handler does its check pre-transaction; this
357 /// function re-checks idempotency inside the lock.)
358 #[tracing::instrument(skip_all)]
359 pub async fn claim_key(
360 pool: &sqlx::PgPool,
361 app_id: SyncAppId,
362 key: &str,
363 ) -> Result<ClaimResult> {
364 let mut tx = pool.begin().await?;
365
366 // Lock the usage row for this app. Returns the current keys_claimed.
367 let (mut keys_claimed,): (i32,) = sqlx::query_as(
368 "SELECT keys_claimed FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
369 )
370 .bind(app_id)
371 .fetch_one(&mut *tx)
372 .await?;
373
374 // Is there already an active claim for this key?
375 let existing: Option<(uuid::Uuid,)> = sqlx::query_as(
376 "SELECT id FROM sync_app_keys
377 WHERE app_id = $1 AND key = $2 AND released_at IS NULL",
378 )
379 .bind(app_id)
380 .bind(key)
381 .fetch_optional(&mut *tx)
382 .await?;
383
384 let newly_claimed = if existing.is_some() {
385 false
386 } else {
387 sqlx::query(
388 "INSERT INTO sync_app_keys (app_id, key) VALUES ($1, $2)",
389 )
390 .bind(app_id)
391 .bind(key)
392 .execute(&mut *tx)
393 .await?;
394
395 sqlx::query(
396 "UPDATE sync_app_usage_current
397 SET keys_claimed = keys_claimed + 1, updated_at = NOW()
398 WHERE app_id = $1",
399 )
400 .bind(app_id)
401 .execute(&mut *tx)
402 .await?;
403
404 keys_claimed += 1;
405 true
406 };
407
408 tx.commit().await?;
409 Ok(ClaimResult {
410 newly_claimed,
411 total_claimed: keys_claimed,
412 })
413 }
414
415 /// Release an SDK encryption key. Idempotent: releasing a key that is not
416 /// actively claimed returns `newly_released = false`.
417 #[tracing::instrument(skip_all)]
418 pub async fn release_key(
419 pool: &sqlx::PgPool,
420 app_id: SyncAppId,
421 key: &str,
422 ) -> Result<ReleaseResult> {
423 let mut tx = pool.begin().await?;
424
425 let (mut keys_claimed,): (i32,) = sqlx::query_as(
426 "SELECT keys_claimed FROM sync_app_usage_current WHERE app_id = $1 FOR UPDATE",
427 )
428 .bind(app_id)
429 .fetch_one(&mut *tx)
430 .await?;
431
432 let released: Option<(uuid::Uuid,)> = sqlx::query_as(
433 "UPDATE sync_app_keys SET released_at = NOW()
434 WHERE app_id = $1 AND key = $2 AND released_at IS NULL
435 RETURNING id",
436 )
437 .bind(app_id)
438 .bind(key)
439 .fetch_optional(&mut *tx)
440 .await?;
441
442 let newly_released = if released.is_some() {
443 sqlx::query(
444 "UPDATE sync_app_usage_current
445 SET keys_claimed = GREATEST(keys_claimed - 1, 0), updated_at = NOW()
446 WHERE app_id = $1",
447 )
448 .bind(app_id)
449 .execute(&mut *tx)
450 .await?;
451 keys_claimed = (keys_claimed - 1).max(0);
452 true
453 } else {
454 false
455 };
456
457 tx.commit().await?;
458 Ok(ReleaseResult {
459 newly_released,
460 total_claimed: keys_claimed,
461 })
462 }
463
464 /// List active (un-released) key claims for an app, ordered by `claimed_at DESC`.
465 /// Used by the dashboard "Active keys" view. `bytes_stored` is LEFT-joined
466 /// from `sync_key_usage_current` — `0` when no upload has landed yet.
467 #[tracing::instrument(skip_all)]
468 pub async fn list_active_keys(
469 pool: &sqlx::PgPool,
470 app_id: SyncAppId,
471 limit: i64,
472 offset: i64,
473 ) -> Result<Vec<DbSyncAppKey>> {
474 let rows = sqlx::query_as::<_, DbSyncAppKey>(
475 r#"
476 SELECT k.id, k.key, k.claimed_at,
477 COALESCE(u.bytes_stored, 0) AS bytes_stored
478 FROM sync_app_keys k
479 LEFT JOIN sync_key_usage_current u
480 ON u.app_id = k.app_id AND u.key = k.key
481 WHERE k.app_id = $1 AND k.released_at IS NULL
482 ORDER BY k.claimed_at DESC
483 LIMIT $2 OFFSET $3
484 "#,
485 )
486 .bind(app_id)
487 .bind(limit)
488 .bind(offset)
489 .fetch_all(pool)
490 .await?;
491 Ok(rows)
492 }
493
494 /// Per-app top-N key usage for the dashboard panel. Returns the highest-usage
495 /// active keys (sorted by `bytes_stored DESC`) for each app in `app_ids`,
496 /// batched into a single query so the integrations page doesn't N+1 across
497 /// every per_key-mode app.
498 #[tracing::instrument(skip_all)]
499 pub async fn get_top_keys_per_app(
500 pool: &PgPool,
501 app_ids: &[SyncAppId],
502 limit_per_app: i64,
503 ) -> Result<Vec<(SyncAppId, String, i64)>> {
504 if app_ids.is_empty() {
505 return Ok(Vec::new());
506 }
507 let rows: Vec<(SyncAppId, String, i64)> = sqlx::query_as(
508 r#"
509 SELECT app_id, key, bytes_stored
510 FROM (
511 SELECT u.app_id, u.key, u.bytes_stored,
512 ROW_NUMBER() OVER (PARTITION BY u.app_id ORDER BY u.bytes_stored DESC, u.key) AS rn
513 FROM sync_key_usage_current u
514 JOIN sync_app_keys k
515 ON k.app_id = u.app_id AND k.key = u.key AND k.released_at IS NULL
516 WHERE u.app_id = ANY($1)
517 ) ranked
518 WHERE rn <= $2
519 ORDER BY app_id, rn
520 "#,
521 )
522 .bind(app_ids)
523 .bind(limit_per_app)
524 .fetch_all(pool)
525 .await?;
526 Ok(rows)
527 }
528
529 // ── Phase 5: Usage counters and cap enforcement ──
530
531 /// A breached cap, returned by `would_exceed_*` checks.
532 #[derive(Debug, Clone, PartialEq, Eq)]
533 pub struct ExceededLimit {
534 /// `"storage"`, `"storage_per_key"`, or `"egress"`. (`"billing"` is reserved
535 /// for inactive-billing failure paths so a single 402 response shape can
536 /// carry every reason; the caller decides whether to thread that through
537 /// here or check `billing_status` separately.)
538 pub dimension: &'static str,
539 /// Current usage in bytes (before the would-be addition).
540 pub used: i64,
541 /// Configured cap in bytes.
542 pub limit: i64,
543 /// The SDK key that hit its cap, if `dimension == "storage_per_key"`.
544 /// `None` for app-wide dimensions.
545 pub key: Option<String>,
546 }
547
548 /// A row that may need a warning email sent.
549 ///
550 /// `threshold_pct` is the highest WARNING_THRESHOLDS_PCT band currently
551 /// breached above `last_warning_pct`. The caller is responsible for filtering
552 /// out apps where no new breach has occurred (i.e. usage hasn't crossed the
553 /// next threshold since the previous warning).
554 #[derive(Debug, Clone, PartialEq, Eq)]
555 pub struct WarningCandidate {
556 pub app_id: SyncAppId,
557 pub creator_id: super::id_types::UserId,
558 pub creator_email: String,
559 pub app_name: String,
560 pub threshold_pct: i16,
561 pub dimension: &'static str,
562 pub used: i64,
563 pub limit: i64,
564 /// SDK key that breached the threshold, when `dimension ==
565 /// "storage_per_key"`. `None` for app-wide breaches.
566 pub key: Option<String>,
567 }
568
569 /// Increment (or decrement, for negative `delta`) the rolling `bytes_stored`
570 /// counter on `sync_app_usage_current` and the per-key row on
571 /// `sync_key_usage_current` (upserted if missing). Returns the new app-level
572 /// total.
573 ///
574 /// NOTE: read-then-add elsewhere in the request handler is racy. Acceptable
575 /// overshoot under concurrent uploads in v1.
576 #[tracing::instrument(skip_all)]
577 pub async fn add_bytes_stored(
578 pool: &PgPool,
579 app_id: SyncAppId,
580 key: &str,
581 delta: i64,
582 ) -> Result<i64> {
583 let mut tx = pool.begin().await?;
584
585 let (new_val,): (i64,) = sqlx::query_as(
586 r#"
587 UPDATE sync_app_usage_current
588 SET bytes_stored = GREATEST(bytes_stored + $2, 0), updated_at = NOW()
589 WHERE app_id = $1
590 RETURNING bytes_stored
591 "#,
592 )
593 .bind(app_id)
594 .bind(delta)
595 .fetch_one(&mut *tx)
596 .await?;
597
598 sqlx::query(
599 r#"
600 INSERT INTO sync_key_usage_current (app_id, key, bytes_stored)
601 VALUES ($1, $2, GREATEST($3, 0))
602 ON CONFLICT (app_id, key)
603 DO UPDATE SET
604 bytes_stored = GREATEST(sync_key_usage_current.bytes_stored + $3, 0),
605 updated_at = NOW()
606 "#,
607 )
608 .bind(app_id)
609 .bind(key)
610 .bind(delta)
611 .execute(&mut *tx)
612 .await?;
613
614 tx.commit().await?;
615 Ok(new_val)
616 }
617
618 /// Increment (or decrement) the per-period `bytes_egress_period` counter.
619 /// Returns the new value.
620 #[tracing::instrument(skip_all)]
621 pub async fn add_bytes_egress(
622 pool: &PgPool,
623 app_id: SyncAppId,
624 delta: i64,
625 ) -> Result<i64> {
626 let (new_val,): (i64,) = sqlx::query_as(
627 r#"
628 UPDATE sync_app_usage_current
629 SET bytes_egress_period = GREATEST(bytes_egress_period + $2, 0), updated_at = NOW()
630 WHERE app_id = $1
631 RETURNING bytes_egress_period
632 "#,
633 )
634 .bind(app_id)
635 .bind(delta)
636 .fetch_one(pool)
637 .await?;
638 Ok(new_val)
639 }
640
641 /// Check whether storing `additional_bytes` more would exceed the storage cap.
642 ///
643 /// Returns `None` if no cap applies (internal apps, billing not active, or
644 /// the cap is unset). Returns `Some(ExceededLimit)` only when adding
645 /// `additional_bytes` would push usage past the cap.
646 ///
647 /// In `per_key` mode the per-key counter is checked first against
648 /// `gb_per_key × 1GB` — if exceeded, returns `dimension: "storage_per_key"`
649 /// with the offending `key`. The app-aggregate (`key_cap × gb_per_key`) is
650 /// also checked as a defensive hard ceiling; under normal operation this
651 /// can't trip before some key has tripped, but it guards against drift.
652 ///
653 /// In `bulk` mode `key` is unused (still passed in for call-site uniformity)
654 /// and the app-wide `storage_gb_cap` is the only check.
655 ///
656 /// Race-condition note: this reads, the caller adds. Concurrent uploads can
657 /// produce small overshoots. Acceptable for v1.
658 #[tracing::instrument(skip_all)]
659 pub async fn would_exceed_storage(
660 pool: &PgPool,
661 app_id: SyncAppId,
662 key: &str,
663 additional_bytes: i64,
664 ) -> Result<Option<ExceededLimit>> {
665 let row: Option<(bool, String, Option<i32>, Option<i32>, Option<i32>, Option<i64>)> = sqlx::query_as(
666 r#"
667 SELECT sa.is_internal, sa.enforcement_mode,
668 sa.storage_gb_cap, sa.key_cap, sa.gb_per_key,
669 u.bytes_stored
670 FROM sync_apps sa
671 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
672 WHERE sa.id = $1
673 "#,
674 )
675 .bind(app_id)
676 .fetch_optional(pool)
677 .await?;
678
679 let Some((is_internal, mode, storage_gb, key_cap, gb_per_key, bytes_stored)) = row else { return Ok(None); };
680 if is_internal { return Ok(None); }
681
682 let app_used = bytes_stored.unwrap_or(0);
683
684 match mode.as_str() {
685 "bulk" => {
686 let Some(gb) = storage_gb else { return Ok(None) };
687 let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
688 if app_used.saturating_add(additional_bytes) > limit {
689 return Ok(Some(ExceededLimit {
690 dimension: "storage",
691 used: app_used,
692 limit,
693 key: None,
694 }));
695 }
696 }
697 "per_key" => {
698 let (Some(k), Some(g)) = (key_cap, gb_per_key) else { return Ok(None) };
699 let per_key_limit = crate::synckit_billing::storage_cap_bytes(g as u32);
700 let app_limit = crate::synckit_billing::storage_cap_bytes(k.saturating_mul(g) as u32);
701
702 let key_used: Option<i64> = sqlx::query_scalar(
703 "SELECT bytes_stored FROM sync_key_usage_current
704 WHERE app_id = $1 AND key = $2",
705 )
706 .bind(app_id)
707 .bind(key)
708 .fetch_optional(pool)
709 .await?;
710 let key_used = key_used.unwrap_or(0);
711
712 if key_used.saturating_add(additional_bytes) > per_key_limit {
713 return Ok(Some(ExceededLimit {
714 dimension: "storage_per_key",
715 used: key_used,
716 limit: per_key_limit,
717 key: Some(key.to_string()),
718 }));
719 }
720 // Defensive app-aggregate ceiling. Tripping this before the per-key
721 // check means the per-key counters have drifted under the app
722 // counter — refuse rather than admit silent over-allocation.
723 if app_used.saturating_add(additional_bytes) > app_limit {
724 return Ok(Some(ExceededLimit {
725 dimension: "storage",
726 used: app_used,
727 limit: app_limit,
728 key: None,
729 }));
730 }
731 }
732 _ => return Ok(None),
733 }
734
735 Ok(None)
736 }
737
738 /// Fetch the list of active, non-internal apps that may need a warning email,
739 /// along with the data needed to compute which threshold (if any) has been
740 /// breached since the last notice. The per-app breach computation is done in
741 /// Rust by the caller (see `scheduler::synckit_warnings`).
742 #[tracing::instrument(skip_all)]
743 pub async fn get_apps_needing_warning(pool: &PgPool) -> Result<Vec<WarningCandidate>> {
744 use super::id_types::UserId;
745
746 // Egress is no longer a price input or an enforced cap (see migration 118),
747 // so the only dimension that warrants a usage warning is storage. The
748 // effective storage cap depends on enforcement_mode.
749 #[derive(sqlx::FromRow)]
750 struct Row {
751 app_id: SyncAppId,
752 creator_id: UserId,
753 creator_email: String,
754 app_name: String,
755 enforcement_mode: String,
756 storage_gb_cap: Option<i32>,
757 key_cap: Option<i32>,
758 gb_per_key: Option<i32>,
759 bytes_stored: i64,
760 last_warning_pct: i16,
761 }
762
763 let rows = sqlx::query_as::<_, Row>(
764 r#"
765 SELECT
766 sa.id AS app_id,
767 sa.creator_id AS creator_id,
768 u_user.email AS creator_email,
769 sa.name AS app_name,
770 sa.enforcement_mode AS enforcement_mode,
771 sa.storage_gb_cap AS storage_gb_cap,
772 sa.key_cap AS key_cap,
773 sa.gb_per_key AS gb_per_key,
774 COALESCE(u.bytes_stored, 0) AS bytes_stored,
775 COALESCE(u.last_warning_pct, 0::smallint) AS last_warning_pct
776 FROM sync_apps sa
777 JOIN users u_user ON u_user.id = sa.creator_id
778 LEFT JOIN sync_app_usage_current u ON u.app_id = sa.id
779 WHERE sa.billing_status = 'active'
780 AND sa.is_internal = FALSE
781 "#,
782 )
783 .fetch_all(pool)
784 .await?;
785
786 let mut out = Vec::new();
787 let mut per_key_apps: Vec<(SyncAppId, UserId, String, String, i32)> = Vec::new();
788 for r in rows {
789 match r.enforcement_mode.as_str() {
790 "bulk" => {
791 let Some(gb) = r.storage_gb_cap else { continue };
792 let limit = crate::synckit_billing::storage_cap_bytes(gb as u32);
793 if let Some(pct) =
794 highest_breached_threshold(r.bytes_stored, limit, r.last_warning_pct)
795 {
796 out.push(WarningCandidate {
797 app_id: r.app_id,
798 creator_id: r.creator_id,
799 creator_email: r.creator_email,
800 app_name: r.app_name,
801 threshold_pct: pct,
802 dimension: "storage",
803 used: r.bytes_stored,
804 limit,
805 key: None,
806 });
807 }
808 }
809 "per_key" => {
810 // Defer to a second query that fans out per (app, key); the
811 // per-key counter — not the app aggregate — is what we warn on.
812 let (Some(_), Some(g)) = (r.key_cap, r.gb_per_key) else { continue };
813 per_key_apps.push((r.app_id, r.creator_id, r.creator_email, r.app_name, g));
814 }
815 _ => {}
816 }
817 }
818
819 if !per_key_apps.is_empty() {
820 #[derive(sqlx::FromRow)]
821 struct KeyRow {
822 app_id: SyncAppId,
823 key: String,
824 bytes_stored: i64,
825 last_warning_pct: i16,
826 }
827 let app_ids: Vec<SyncAppId> = per_key_apps.iter().map(|t| t.0).collect();
828 let key_rows = sqlx::query_as::<_, KeyRow>(
829 r#"
830 SELECT u.app_id, u.key, u.bytes_stored, u.last_warning_pct
831 FROM sync_key_usage_current u
832 JOIN sync_app_keys k
833 ON k.app_id = u.app_id AND k.key = u.key AND k.released_at IS NULL
834 WHERE u.app_id = ANY($1)
835 "#,
836 )
837 .bind(&app_ids)
838 .fetch_all(pool)
839 .await?;
840
841 // Index app metadata by id for cheap join.
842 let meta: std::collections::HashMap<SyncAppId, (UserId, String, String, i32)> =
843 per_key_apps
844 .into_iter()
845 .map(|(a, c, e, n, g)| (a, (c, e, n, g)))
846 .collect();
847
848 for r in key_rows {
849 let Some((creator_id, creator_email, app_name, gb_per_key)) =
850 meta.get(&r.app_id).cloned()
851 else {
852 continue;
853 };
854 let limit = crate::synckit_billing::storage_cap_bytes(gb_per_key as u32);
855 if let Some(pct) =
856 highest_breached_threshold(r.bytes_stored, limit, r.last_warning_pct)
857 {
858 out.push(WarningCandidate {
859 app_id: r.app_id,
860 creator_id,
861 creator_email,
862 app_name,
863 threshold_pct: pct,
864 dimension: "storage_per_key",
865 used: r.bytes_stored,
866 limit,
867 key: Some(r.key),
868 });
869 }
870 }
871 }
872
873 Ok(out)
874 }
875
876 /// Compute the highest WARNING_THRESHOLDS_PCT band currently exceeded by
877 /// `used / limit` whose value is strictly above `last_warning_pct`.
878 ///
879 /// Returns `None` if no new band has been breached.
880 pub fn highest_breached_threshold(used: i64, limit: i64, last_warning_pct: i16) -> Option<i16> {
881 if limit <= 0 { return None; }
882 // Compute current percentage as integer; saturate at 100+.
883 // Use f64 to avoid 32-bit overflow for very large byte counts.
884 let pct_f = (used as f64 / limit as f64) * 100.0;
885 let pct = pct_f.floor() as i64;
886 crate::synckit_billing::WARNING_THRESHOLDS_PCT
887 .iter()
888 .rev()
889 .copied()
890 .find(|&t| pct >= t as i64 && t > last_warning_pct)
891 }
892
893 /// Stamp `last_warning_pct` to record that a warning at `pct` has fired.
894 #[tracing::instrument(skip_all)]
895 pub async fn update_warning_pct(
896 pool: &PgPool,
897 app_id: SyncAppId,
898 pct: i16,
899 ) -> Result<()> {
900 sqlx::query(
901 r#"
902 UPDATE sync_app_usage_current
903 SET last_warning_pct = $2, updated_at = NOW()
904 WHERE app_id = $1
905 "#,
906 )
907 .bind(app_id)
908 .bind(pct)
909 .execute(pool)
910 .await?;
911 Ok(())
912 }
913
914 /// Per-key analogue of `update_warning_pct`. Stamps the band on
915 /// `sync_key_usage_current` so subsequent ticks don't re-fire the same band
916 /// for the same key.
917 #[tracing::instrument(skip_all)]
918 pub async fn update_key_warning_pct(
919 pool: &PgPool,
920 app_id: SyncAppId,
921 key: &str,
922 pct: i16,
923 ) -> Result<()> {
924 sqlx::query(
925 r#"
926 UPDATE sync_key_usage_current
927 SET last_warning_pct = $3, updated_at = NOW()
928 WHERE app_id = $1 AND key = $2
929 "#,
930 )
931 .bind(app_id)
932 .bind(key)
933 .bind(pct)
934 .execute(pool)
935 .await?;
936 Ok(())
937 }
938
939 /// Recalculate `bytes_stored` from the authoritative `sync_blobs` table for
940 /// every app and every per-key counter. Weekly drift correction. Returns
941 /// total count of rows updated across both tables.
942 #[tracing::instrument(skip_all)]
943 pub async fn recalculate_synckit_app_storage(pool: &PgPool) -> Result<u64> {
944 let mut tx = pool.begin().await?;
945
946 let app_res = sqlx::query(
947 r#"
948 UPDATE sync_app_usage_current u
949 SET bytes_stored = COALESCE(s.total, 0), updated_at = NOW()
950 FROM (
951 SELECT app_id, SUM(size_bytes)::BIGINT AS total
952 FROM sync_blobs
953 GROUP BY app_id
954 ) s
955 WHERE u.app_id = s.app_id
956 AND u.bytes_stored <> COALESCE(s.total, 0)
957 "#,
958 )
959 .execute(&mut *tx)
960 .await?;
961
962 // Per-key reconciliation: upsert every (app_id, key) found in sync_blobs.
963 // Keys that disappear (all blobs deleted) aren't pruned here — bytes_stored
964 // would just stay at whatever it last drifted to. Blob delete isn't shipped
965 // yet, so this is fine. When it ships, add a step that resets rows whose
966 // sync_blobs total is zero (or just delete them — the upsert recreates).
967 let key_res = sqlx::query(
968 r#"
969 INSERT INTO sync_key_usage_current (app_id, key, bytes_stored, updated_at)
970 SELECT app_id, key, SUM(size_bytes)::BIGINT, NOW()
971 FROM sync_blobs
972 GROUP BY app_id, key
973 ON CONFLICT (app_id, key)
974 DO UPDATE SET
975 bytes_stored = EXCLUDED.bytes_stored,
976 updated_at = NOW()
977 WHERE sync_key_usage_current.bytes_stored <> EXCLUDED.bytes_stored
978 "#,
979 )
980 .execute(&mut *tx)
981 .await?;
982
983 tx.commit().await?;
984 Ok(app_res.rows_affected() + key_res.rows_affected())
985 }
986
987 /// Check whether a key is currently actively claimed. Used by the claim
988 /// handler to short-circuit the cap check for idempotent re-claims.
989 #[tracing::instrument(skip_all)]
990 pub async fn is_key_actively_claimed(
991 pool: &sqlx::PgPool,
992 app_id: SyncAppId,
993 key: &str,
994 ) -> Result<bool> {
995 let row: Option<(uuid::Uuid,)> = sqlx::query_as(
996 "SELECT id FROM sync_app_keys
997 WHERE app_id = $1 AND key = $2 AND released_at IS NULL",
998 )
999 .bind(app_id)
1000 .bind(key)
1001 .fetch_optional(pool)
1002 .await?;
1003 Ok(row.is_some())
1004 }
1005