Skip to main content

max / makenotwork

Split db/creator_tiers into subscriptions + storage_quota modules The 1172-line creator_tiers.rs mixed two domains on the creator-tier table cluster: subscription lifecycle and storage-quota accounting/upload gating. Split into a creator_tiers/ directory — subscriptions.rs (create, Stripe apply, tier resolution, grace/founder counts), storage_quota.rs (usage counters, breakdown, batch recalc, upload/presign gating), and tests.rs — with mod.rs re-exporting both via `pub use`. The db::creator_tiers::* path is preserved, so all call sites are unchanged. No behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 13:02 UTC
Commit: c43108711ff20569f65368a53a7edd1da6cf0153
Parent: 1d912dc
5 files changed, +1026 insertions, -500 deletions
D server/src/db/creator_tiers.rs -500
@@ -1,1172 +0,0 @@
1 - //! Creator tier subscription queries and storage enforcement.
2 -
3 - use chrono::{DateTime, Utc};
4 - use sqlx::PgPool;
5 -
6 - use super::enums::CreatorTier;
7 - use super::id_types::*;
8 - use super::models::{DbCreatorSubscription, StorageBreakdown};
9 - use crate::error::{AppError, Result};
10 - use crate::helpers::format_bytes;
11 - use crate::storage::FileType;
12 -
13 - /// Create or reactivate a creator tier subscription record.
14 - ///
15 - /// Uses ON CONFLICT DO UPDATE on the user_id unique index to handle
16 - /// both duplicate webhooks and re-subscription after cancellation.
17 - /// Returns `None` if the row already existed with the same stripe_subscription_id
18 - /// (duplicate webhook), `Some` if this was a fresh insert or a re-subscription
19 - /// with a different subscription ID.
20 - #[tracing::instrument(skip_all)]
21 - pub async fn create_creator_subscription<'e>(
22 - executor: impl sqlx::PgExecutor<'e>,
23 - user_id: UserId,
24 - stripe_subscription_id: &str,
25 - stripe_customer_id: &str,
26 - tier: CreatorTier,
27 - ) -> Result<Option<DbCreatorSubscription>> {
28 - // Use WHERE clause on the DO UPDATE to only update if the subscription_id
29 - // is different (new subscription) or status is not already active.
30 - // When the WHERE fails, DO UPDATE becomes a no-op and RETURNING yields no row.
31 - let sub = sqlx::query_as::<_, DbCreatorSubscription>(
32 - r#"
33 - INSERT INTO creator_subscriptions (user_id, stripe_subscription_id, stripe_customer_id, tier)
34 - VALUES ($1, $2, $3, $4)
35 - ON CONFLICT (user_id) DO UPDATE
36 - SET stripe_subscription_id = EXCLUDED.stripe_subscription_id,
37 - stripe_customer_id = EXCLUDED.stripe_customer_id,
38 - tier = EXCLUDED.tier,
39 - status = 'active',
40 - canceled_at = NULL,
41 - grace_enforced_at = NULL
42 - WHERE creator_subscriptions.stripe_subscription_id != EXCLUDED.stripe_subscription_id
43 - OR creator_subscriptions.status != 'active'
44 - RETURNING *
45 - "#,
46 - )
47 - .bind(user_id)
48 - .bind(stripe_subscription_id)
49 - .bind(stripe_customer_id)
50 - .bind(tier)
51 - .fetch_optional(executor)
52 - .await?;
53 -
54 - Ok(sub)
55 - }
56 -
57 - /// Look up a creator subscription by its Stripe subscription ID.
58 - #[tracing::instrument(skip_all)]
59 - pub async fn get_creator_sub_by_stripe_id(
60 - pool: &PgPool,
61 - stripe_subscription_id: &str,
62 - ) -> Result<Option<DbCreatorSubscription>> {
63 - let sub = sqlx::query_as::<_, DbCreatorSubscription>(
64 - "SELECT * FROM creator_subscriptions WHERE stripe_subscription_id = $1",
65 - )
66 - .bind(stripe_subscription_id)
67 - .fetch_optional(pool)
68 - .await?;
69 -
70 - Ok(sub)
71 - }
72 -
73 - /// Get a user's creator subscription (any status).
74 - #[tracing::instrument(skip_all)]
75 - pub async fn get_creator_sub_by_user(
76 - pool: &PgPool,
77 - user_id: UserId,
78 - ) -> Result<Option<DbCreatorSubscription>> {
79 - let sub = sqlx::query_as::<_, DbCreatorSubscription>(
80 - "SELECT * FROM creator_subscriptions WHERE user_id = $1",
81 - )
82 - .bind(user_id)
83 - .fetch_optional(pool)
84 - .await?;
85 -
86 - Ok(sub)
87 - }
88 -
89 - /// Get the active creator tier for a user (None if no active subscription).
90 - #[tracing::instrument(skip_all)]
91 - pub async fn get_active_creator_tier(
92 - pool: &PgPool,
93 - user_id: UserId,
94 - ) -> Result<Option<CreatorTier>> {
95 - let tier = sqlx::query_scalar::<_, String>(
96 - "SELECT tier FROM creator_subscriptions WHERE user_id = $1 AND status = 'active'",
97 - )
98 - .bind(user_id)
99 - .fetch_optional(pool)
100 - .await?;
101 -
102 - match tier {
103 - None => Ok(None),
104 - // A present-but-unparseable tier means the DB holds a tier string the
105 - // enum doesn't know (enum drift). Silently mapping that to `None` would
106 - // strip a paying creator's entitlements with no signal — surface it
107 - // loudly instead of swallowing it (Run 21). The `enum_drift` test and the
108 - // DB CHECK constraint make this unreachable in practice.
109 - Some(t) => match t.parse::<CreatorTier>() {
110 - Ok(parsed) => Ok(Some(parsed)),
111 - Err(_) => {
112 - tracing::error!(
113 - user_id = %user_id, tier = %t,
114 - "active creator subscription has an unrecognized tier string (enum drift)"
115 - );
116 - Err(crate::error::AppError::Internal(anyhow::anyhow!(
117 - "unrecognized creator tier '{t}' for user {user_id}"
118 - )))
119 - }
120 - },
121 - }
122 - }
123 -
124 - // Apply a Stripe-driven status and/or period update in one guarded statement.
125 - // `canceled` is terminal; reactivation runs through `create_creator_subscription`'s
126 - // `ON CONFLICT (user_id) DO UPDATE` at checkout, never through this path. Replaces
127 - // the old split status/period setters (the period half lacked the guard). See
128 - // `crate::db::subscription_writer`.
129 - crate::db::subscription_writer::define_stripe_subscription_writer!(
130 - apply_stripe_update,
131 - "creator_subscriptions",
132 - DbCreatorSubscription
133 - );
134 -
135 - /// Cancel a creator subscription (set status + canceled_at).
136 - #[tracing::instrument(skip_all)]
137 - pub async fn cancel_creator_sub(
138 - pool: &PgPool,
139 - stripe_subscription_id: &str,
140 - ) -> Result<Option<DbCreatorSubscription>> {
141 - let sub = sqlx::query_as::<_, DbCreatorSubscription>(
142 - r#"
143 - UPDATE creator_subscriptions
144 - SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
145 - WHERE stripe_subscription_id = $1
146 - RETURNING *
147 - "#,
148 - )
149 - .bind(stripe_subscription_id)
150 - .fetch_optional(pool)
151 - .await?;
152 -
153 - Ok(sub)
154 - }
155 -
156 - /// Sync the users.creator_tier column from the subscription status.
157 - /// Called after checkout/update/cancel to keep the denormalized column in sync.
158 - #[tracing::instrument(skip_all)]
159 - pub async fn sync_user_creator_tier(pool: &PgPool, user_id: UserId) -> Result<()> {
160 - sqlx::query(
161 - r#"
162 - UPDATE users SET creator_tier = (
163 - SELECT tier FROM creator_subscriptions
164 - WHERE user_id = $1 AND status = 'active'
165 - )
166 - WHERE id = $1
167 - "#,
168 - )
169 - .bind(user_id)
170 - .execute(pool)
171 - .await?;
172 -
173 - Ok(())
174 - }
175 -
176 - // ============================================================================
177 - // Storage tracking
178 - // ============================================================================
179 -
180 - /// Get the current storage_used_bytes for a user.
181 - #[tracing::instrument(skip_all)]
182 - pub async fn get_storage_used(pool: &PgPool, user_id: UserId) -> Result<i64> {
183 - let used: i64 = sqlx::query_scalar(
184 - "SELECT storage_used_bytes FROM users WHERE id = $1",
185 - )
186 - .bind(user_id)
187 - .fetch_one(pool)
188 - .await?;
189 -
190 - Ok(used)
191 - }
192 -
193 - /// Build the "storage cap exceeded" error, reading current usage on `executor`
194 - /// for the message. Shared by the cap-checked UPDATE helpers below so the
195 - /// wording (and the usage read) can't drift between them. Returns the error in
196 - /// `Ok`; a failed usage read propagates as `Err`, matching the prior inline code.
197 - async fn storage_cap_exceeded<'e>(
198 - executor: impl sqlx::PgExecutor<'e>,
199 - user_id: UserId,
200 - max_storage_bytes: i64,
201 - ) -> Result<AppError> {
202 - let used: i64 = sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1")
203 - .bind(user_id)
204 - .fetch_one(executor)
205 - .await?;
206 - Ok(AppError::BadRequest(format!(
207 - "You've used {} of {} storage. Delete files or upgrade your tier.",
208 - format_bytes(used),
209 - format_bytes(max_storage_bytes),
210 - )))
211 - }
212 -
213 - /// Atomically check storage cap and increment the user's storage counter.
214 - /// Returns an error if the increment would exceed `max_storage_bytes`.
215 - #[tracing::instrument(skip_all)]
216 - pub async fn try_increment_storage(
217 - pool: &PgPool,
218 - user_id: UserId,
219 - bytes: i64,
220 - max_storage_bytes: i64,
221 - ) -> Result<()> {
222 - let result = sqlx::query(
223 - "UPDATE users SET storage_used_bytes = storage_used_bytes + $2 \
224 - WHERE id = $1 AND storage_used_bytes + $2 <= $3",
225 - )
226 - .bind(user_id)
227 - .bind(bytes)
228 - .bind(max_storage_bytes)
229 - .execute(pool)
230 - .await?;
231 -
232 - if result.rows_affected() == 0 {
233 - return Err(storage_cap_exceeded(pool, user_id, max_storage_bytes).await?);
234 - }
235 -
236 - Ok(())
237 - }
238 -
239 - /// Transaction-friendly variant of [`try_increment_storage`]. Runs both the
240 - /// cap-checked UPDATE and the error-path SELECT against the supplied
241 - /// connection, so callers can wrap the increment in the same transaction as
242 - /// the follow-up entity write (e.g. `media_files::create`). On cap miss the
243 - /// transaction is left open for the caller to roll back via drop.
244 - #[tracing::instrument(skip_all)]
245 - pub async fn try_increment_storage_on(
246 - conn: &mut sqlx::PgConnection,
247 - user_id: UserId,
248 - bytes: i64,
249 - max_storage_bytes: i64,
250 - ) -> Result<()> {
251 - let result = sqlx::query(
252 - "UPDATE users SET storage_used_bytes = storage_used_bytes + $2 \
253 - WHERE id = $1 AND storage_used_bytes + $2 <= $3",
254 - )
255 - .bind(user_id)
256 - .bind(bytes)
257 - .bind(max_storage_bytes)
258 - .execute(&mut *conn)
259 - .await?;
260 -
261 - if result.rows_affected() == 0 {
262 - return Err(storage_cap_exceeded(&mut *conn, user_id, max_storage_bytes).await?);
263 - }
264 -
265 - Ok(())
266 - }
267 -
268 - /// Atomically replace storage: decrement old file size and increment new file
269 - /// size in a single cap-checked UPDATE, on a caller-supplied connection so it can
270 - /// be bundled with the row UPDATE in one transaction (no storage-drift window on
271 - /// a mid-write failure). Prevents drift by avoiding a separate decrement/increment.
272 - #[tracing::instrument(skip_all)]
273 - pub async fn try_replace_storage_on(
274 - conn: &mut sqlx::PgConnection,
275 - user_id: UserId,
276 - old_bytes: i64,
277 - new_bytes: i64,
278 - max_storage_bytes: i64,
279 - ) -> Result<()> {
280 - let result = sqlx::query(
281 - "UPDATE users SET storage_used_bytes = GREATEST(0, storage_used_bytes - $2) + $3 \
282 - WHERE id = $1 AND GREATEST(0, storage_used_bytes - $2) + $3 <= $4",
283 - )
284 - .bind(user_id)
285 - .bind(old_bytes)
286 - .bind(new_bytes)
287 - .bind(max_storage_bytes)
288 - .execute(&mut *conn)
289 - .await?;
290 -
291 - if result.rows_affected() == 0 {
292 - return Err(storage_cap_exceeded(&mut *conn, user_id, max_storage_bytes).await?);
293 - }
294 -
295 - Ok(())
296 - }
297 -
298 - /// Apply a confirmed upload's storage credit on a transaction connection:
299 - /// `replace_old_size = Some(old)` swaps an existing file's size for the new one
300 - /// (atomic decrement-old + increment-new); `None` is a fresh increment. Lets the
301 - /// four upload-confirm handlers share one call instead of each re-deriving the
302 - /// replace-vs-increment branch.
303 - #[tracing::instrument(skip_all)]
304 - pub async fn try_apply_storage_on(
305 - conn: &mut sqlx::PgConnection,
306 - user_id: UserId,
307 - replace_old_size: Option<i64>,
308 - new_bytes: i64,
309 - max_storage_bytes: i64,
310 - ) -> Result<()> {
311 - match replace_old_size {
312 - Some(old_bytes) => {
313 - try_replace_storage_on(conn, user_id, old_bytes, new_bytes, max_storage_bytes).await
314 - }
315 - None => try_increment_storage_on(conn, user_id, new_bytes, max_storage_bytes).await,
316 - }
317 - }
318 -
319 - /// Atomically decrement the user's storage counter (clamped to 0).
320 - #[tracing::instrument(skip_all)]
321 - pub async fn decrement_storage_used<'e>(
322 - executor: impl sqlx::PgExecutor<'e>,
323 - user_id: UserId,
324 - bytes: i64,
325 - ) -> Result<()> {
326 - sqlx::query(
327 - "UPDATE users SET storage_used_bytes = GREATEST(0, storage_used_bytes - $2) WHERE id = $1",
328 - )
329 - .bind(user_id)
330 - .bind(bytes)
331 - .execute(executor)
332 - .await?;
333 -
334 - Ok(())
335 - }
336 -
337 - /// Get the admin-set per-file size override for a user.
338 - #[tracing::instrument(skip_all)]
339 - pub async fn get_max_file_override(pool: &PgPool, user_id: UserId) -> Result<Option<i64>> {
340 - let val: Option<i64> = sqlx::query_scalar(
341 - "SELECT max_file_override_bytes FROM users WHERE id = $1",
342 - )
343 - .bind(user_id)
344 - .fetch_one(pool)
345 - .await?;
346 -
347 - Ok(val)
348 - }
349 -
350 - /// Set or clear the admin per-file size override.
351 - #[tracing::instrument(skip_all)]
352 - pub async fn set_max_file_override(
353 - pool: &PgPool,
354 - user_id: UserId,
355 - bytes: Option<i64>,
356 - ) -> Result<()> {
357 - sqlx::query(
358 - "UPDATE users SET max_file_override_bytes = $2 WHERE id = $1",
359 - )
360 - .bind(user_id)
361 - .bind(bytes)
362 - .execute(pool)
363 - .await?;
364 -
365 - Ok(())
366 - }
367 -
368 - /// Get the grandfathering deadline for a user.
369 - #[tracing::instrument(skip_all)]
370 - pub async fn get_grandfathered_until(
371 - pool: &PgPool,
372 - user_id: UserId,
373 - ) -> Result<Option<DateTime<Utc>>> {
374 - let val: Option<DateTime<Utc>> = sqlx::query_scalar(
375 - "SELECT grandfathered_until FROM users WHERE id = $1",
376 - )
377 - .bind(user_id)
378 - .fetch_one(pool)
379 - .await?;
380 -
381 - Ok(val)
382 - }
383 -
384 - /// Get a per-category storage breakdown for the creator dashboard (single query).
385 - #[tracing::instrument(skip_all)]
386 - pub async fn get_storage_breakdown(pool: &PgPool, user_id: UserId) -> Result<StorageBreakdown> {
387 - // Categories must partition every byte counted by `recalculate_all_storage_batch`
388 - // so the dashboard total reconciles with `storage_used_bytes`. `cover_bytes`
389 - // folds item covers + project covers; `gallery_bytes` covers the item/project
390 - // image carousels (Run #18 Storage B1).
391 - let row: (i64, i64, i64, i64, i64, i64, i64) = sqlx::query_as(
392 - r#"
393 - WITH audio_bytes AS (
394 - SELECT COALESCE(GREATEST(0, LEAST(SUM(i.audio_file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
395 - FROM items i JOIN projects p ON i.project_id = p.id
396 - WHERE p.user_id = $1 AND i.audio_file_size_bytes IS NOT NULL
397 - ),
398 - cover_bytes AS (
399 - SELECT COALESCE(GREATEST(0, LEAST(SUM(i.cover_file_size_bytes), 9223372036854775807))::BIGINT, 0)
400 - + COALESCE((
401 - SELECT GREATEST(0, LEAST(SUM(p.cover_image_size_bytes), 9223372036854775807))::BIGINT
402 - FROM projects p
403 - WHERE p.user_id = $1 AND p.cover_image_size_bytes IS NOT NULL
404 - ), 0) AS total
405 - FROM items i JOIN projects p ON i.project_id = p.id
406 - WHERE p.user_id = $1 AND i.cover_file_size_bytes IS NOT NULL
407 - ),
408 - version_bytes AS (
409 - SELECT COALESCE(GREATEST(0, LEAST(SUM(v.file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
410 - FROM versions v
411 - JOIN items i ON v.item_id = i.id
412 - JOIN projects p ON i.project_id = p.id
413 - WHERE p.user_id = $1 AND v.file_size_bytes IS NOT NULL
414 - ),
415 - insertion_bytes AS (
416 - SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size), 9223372036854775807))::BIGINT, 0) AS total
417 - FROM content_insertions WHERE user_id = $1
418 - ),
419 - video_bytes AS (
420 - SELECT COALESCE(GREATEST(0, LEAST(SUM(i.video_file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
421 - FROM items i JOIN projects p ON i.project_id = p.id
422 - WHERE p.user_id = $1 AND i.video_file_size_bytes IS NOT NULL
423 - ),
424 - media_bytes AS (
425 - SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
426 - FROM media_files WHERE user_id = $1
427 - ),
428 - gallery_bytes AS (
429 - SELECT COALESCE((
430 - SELECT GREATEST(0, LEAST(SUM(ii.file_size_bytes), 9223372036854775807))::BIGINT
431 - FROM item_images ii JOIN items i ON ii.item_id = i.id JOIN projects p ON i.project_id = p.id
432 - WHERE p.user_id = $1
433 - ), 0)
434 - + COALESCE((
435 - SELECT GREATEST(0, LEAST(SUM(pi.file_size_bytes), 9223372036854775807))::BIGINT
436 - FROM project_images pi JOIN projects p ON pi.project_id = p.id
437 - WHERE p.user_id = $1
438 - ), 0) AS total
439 - )
440 - SELECT
441 - (SELECT total FROM audio_bytes),
442 - (SELECT total FROM cover_bytes),
443 - (SELECT total FROM version_bytes),
444 - (SELECT total FROM insertion_bytes),
445 - (SELECT total FROM video_bytes),
446 - (SELECT total FROM media_bytes),
447 - (SELECT total FROM gallery_bytes)
448 - "#,
449 - )
450 - .bind(user_id)
451 - .fetch_one(pool)
452 - .await?;
453 -
454 - Ok(StorageBreakdown {
455 - audio_bytes: row.0,
456 - cover_bytes: row.1,
457 - download_bytes: row.2,
458 - insertion_bytes: row.3,
459 - video_bytes: row.4,
460 - media_bytes: row.5,
461 - gallery_bytes: row.6,
462 - // Each per-category SUM is SQL-clamped to [0, i64::MAX]; sum them
463 - // saturating so seven near-max categories can't overflow the total.
464 - total_bytes: [row.0, row.1, row.2, row.3, row.4, row.5, row.6]
465 - .into_iter()
466 - .fold(0i64, i64::saturating_add),
467 - })
468 - }
469 -
470 - /// Get user IDs of creators with canceled subscriptions 30+ days ago
471 - /// whose items have not yet been hidden.
472 - #[tracing::instrument(skip_all)]
473 - pub async fn get_expired_grace_creators(pool: &PgPool) -> Result<Vec<UserId>> {
474 - // Bounded batch per call. The scheduler enforces these inline on the tick
475 - // (two DB round-trips per creator), so an unbounded result set would let a
476 - // backlog stall the tick. `grace_enforced_at` is set as each creator is
477 - // processed, so successive ticks drain the rest; ORDER BY oldest-first keeps
478 - // it deterministic and starvation-free.
479 - let ids: Vec<UserId> = sqlx::query_scalar(
480 - r#"
481 - SELECT user_id FROM creator_subscriptions
482 - WHERE status = 'canceled'
483 - AND canceled_at IS NOT NULL
484 - AND canceled_at < NOW() - INTERVAL '30 days'
485 - AND grace_enforced_at IS NULL
486 - ORDER BY canceled_at ASC
487 - LIMIT 200
488 - "#,
489 - )
490 - .fetch_all(pool)
491 - .await?;
492 -
493 - Ok(ids)
494 - }
495 -
496 - /// Stamp `grace_enforced_at` for all given creators in one statement, paired with
497 - /// `items::hide_all_items_for_users` on the post-grace sweep (Perf-S4, Run 9).
498 - /// No-op on an empty slice.
499 - #[tracing::instrument(skip_all)]
500 - pub async fn mark_grace_enforced_batch(pool: &PgPool, user_ids: &[UserId]) -> Result<()> {
Lines truncated
A server/src/db/creator_tiers/mod.rs +20
@@ -0,0 +1,20 @@
1 + //! Creator tier subscription queries and storage enforcement.
2 +
3 + use chrono::{DateTime, Utc};
4 + use sqlx::PgPool;
5 +
6 + use super::enums::CreatorTier;
7 + use super::id_types::*;
8 + use super::models::{DbCreatorSubscription, StorageBreakdown};
9 + use crate::error::{AppError, Result};
10 + use crate::helpers::format_bytes;
11 + use crate::storage::FileType;
12 +
13 + mod storage_quota;
14 + mod subscriptions;
15 +
16 + pub use storage_quota::*;
17 + pub use subscriptions::*;
18 +
19 + #[cfg(test)]
20 + mod tests;
A server/src/db/creator_tiers/storage_quota.rs +500
@@ -0,0 +1,653 @@
1 + use super::*;
2 +
3 + // ============================================================================
4 + // Storage tracking
5 + // ============================================================================
6 +
7 + /// Get the current storage_used_bytes for a user.
8 + #[tracing::instrument(skip_all)]
9 + pub async fn get_storage_used(pool: &PgPool, user_id: UserId) -> Result<i64> {
10 + let used: i64 = sqlx::query_scalar(
11 + "SELECT storage_used_bytes FROM users WHERE id = $1",
12 + )
13 + .bind(user_id)
14 + .fetch_one(pool)
15 + .await?;
16 +
17 + Ok(used)
18 + }
19 +
20 + /// Build the "storage cap exceeded" error, reading current usage on `executor`
21 + /// for the message. Shared by the cap-checked UPDATE helpers below so the
22 + /// wording (and the usage read) can't drift between them. Returns the error in
23 + /// `Ok`; a failed usage read propagates as `Err`, matching the prior inline code.
24 + async fn storage_cap_exceeded<'e>(
25 + executor: impl sqlx::PgExecutor<'e>,
26 + user_id: UserId,
27 + max_storage_bytes: i64,
28 + ) -> Result<AppError> {
29 + let used: i64 = sqlx::query_scalar("SELECT storage_used_bytes FROM users WHERE id = $1")
30 + .bind(user_id)
31 + .fetch_one(executor)
32 + .await?;
33 + Ok(AppError::BadRequest(format!(
34 + "You've used {} of {} storage. Delete files or upgrade your tier.",
35 + format_bytes(used),
36 + format_bytes(max_storage_bytes),
37 + )))
38 + }
39 +
40 + /// Atomically check storage cap and increment the user's storage counter.
41 + /// Returns an error if the increment would exceed `max_storage_bytes`.
42 + #[tracing::instrument(skip_all)]
43 + pub async fn try_increment_storage(
44 + pool: &PgPool,
45 + user_id: UserId,
46 + bytes: i64,
47 + max_storage_bytes: i64,
48 + ) -> Result<()> {
49 + let result = sqlx::query(
50 + "UPDATE users SET storage_used_bytes = storage_used_bytes + $2 \
51 + WHERE id = $1 AND storage_used_bytes + $2 <= $3",
52 + )
53 + .bind(user_id)
54 + .bind(bytes)
55 + .bind(max_storage_bytes)
56 + .execute(pool)
57 + .await?;
58 +
59 + if result.rows_affected() == 0 {
60 + return Err(storage_cap_exceeded(pool, user_id, max_storage_bytes).await?);
61 + }
62 +
63 + Ok(())
64 + }
65 +
66 + /// Transaction-friendly variant of [`try_increment_storage`]. Runs both the
67 + /// cap-checked UPDATE and the error-path SELECT against the supplied
68 + /// connection, so callers can wrap the increment in the same transaction as
69 + /// the follow-up entity write (e.g. `media_files::create`). On cap miss the
70 + /// transaction is left open for the caller to roll back via drop.
71 + #[tracing::instrument(skip_all)]
72 + pub async fn try_increment_storage_on(
73 + conn: &mut sqlx::PgConnection,
74 + user_id: UserId,
75 + bytes: i64,
76 + max_storage_bytes: i64,
77 + ) -> Result<()> {
78 + let result = sqlx::query(
79 + "UPDATE users SET storage_used_bytes = storage_used_bytes + $2 \
80 + WHERE id = $1 AND storage_used_bytes + $2 <= $3",
81 + )
82 + .bind(user_id)
83 + .bind(bytes)
84 + .bind(max_storage_bytes)
85 + .execute(&mut *conn)
86 + .await?;
87 +
88 + if result.rows_affected() == 0 {
89 + return Err(storage_cap_exceeded(&mut *conn, user_id, max_storage_bytes).await?);
90 + }
91 +
92 + Ok(())
93 + }
94 +
95 + /// Atomically replace storage: decrement old file size and increment new file
96 + /// size in a single cap-checked UPDATE, on a caller-supplied connection so it can
97 + /// be bundled with the row UPDATE in one transaction (no storage-drift window on
98 + /// a mid-write failure). Prevents drift by avoiding a separate decrement/increment.
99 + #[tracing::instrument(skip_all)]
100 + pub async fn try_replace_storage_on(
101 + conn: &mut sqlx::PgConnection,
102 + user_id: UserId,
103 + old_bytes: i64,
104 + new_bytes: i64,
105 + max_storage_bytes: i64,
106 + ) -> Result<()> {
107 + let result = sqlx::query(
108 + "UPDATE users SET storage_used_bytes = GREATEST(0, storage_used_bytes - $2) + $3 \
109 + WHERE id = $1 AND GREATEST(0, storage_used_bytes - $2) + $3 <= $4",
110 + )
111 + .bind(user_id)
112 + .bind(old_bytes)
113 + .bind(new_bytes)
114 + .bind(max_storage_bytes)
115 + .execute(&mut *conn)
116 + .await?;
117 +
118 + if result.rows_affected() == 0 {
119 + return Err(storage_cap_exceeded(&mut *conn, user_id, max_storage_bytes).await?);
120 + }
121 +
122 + Ok(())
123 + }
124 +
125 + /// Apply a confirmed upload's storage credit on a transaction connection:
126 + /// `replace_old_size = Some(old)` swaps an existing file's size for the new one
127 + /// (atomic decrement-old + increment-new); `None` is a fresh increment. Lets the
128 + /// four upload-confirm handlers share one call instead of each re-deriving the
129 + /// replace-vs-increment branch.
130 + #[tracing::instrument(skip_all)]
131 + pub async fn try_apply_storage_on(
132 + conn: &mut sqlx::PgConnection,
133 + user_id: UserId,
134 + replace_old_size: Option<i64>,
135 + new_bytes: i64,
136 + max_storage_bytes: i64,
137 + ) -> Result<()> {
138 + match replace_old_size {
139 + Some(old_bytes) => {
140 + try_replace_storage_on(conn, user_id, old_bytes, new_bytes, max_storage_bytes).await
141 + }
142 + None => try_increment_storage_on(conn, user_id, new_bytes, max_storage_bytes).await,
143 + }
144 + }
145 +
146 + /// Atomically decrement the user's storage counter (clamped to 0).
147 + #[tracing::instrument(skip_all)]
148 + pub async fn decrement_storage_used<'e>(
149 + executor: impl sqlx::PgExecutor<'e>,
150 + user_id: UserId,
151 + bytes: i64,
152 + ) -> Result<()> {
153 + sqlx::query(
154 + "UPDATE users SET storage_used_bytes = GREATEST(0, storage_used_bytes - $2) WHERE id = $1",
155 + )
156 + .bind(user_id)
157 + .bind(bytes)
158 + .execute(executor)
159 + .await?;
160 +
161 + Ok(())
162 + }
163 +
164 + /// Get the admin-set per-file size override for a user.
165 + #[tracing::instrument(skip_all)]
166 + pub async fn get_max_file_override(pool: &PgPool, user_id: UserId) -> Result<Option<i64>> {
167 + let val: Option<i64> = sqlx::query_scalar(
168 + "SELECT max_file_override_bytes FROM users WHERE id = $1",
169 + )
170 + .bind(user_id)
171 + .fetch_one(pool)
172 + .await?;
173 +
174 + Ok(val)
175 + }
176 +
177 + /// Set or clear the admin per-file size override.
178 + #[tracing::instrument(skip_all)]
179 + pub async fn set_max_file_override(
180 + pool: &PgPool,
181 + user_id: UserId,
182 + bytes: Option<i64>,
183 + ) -> Result<()> {
184 + sqlx::query(
185 + "UPDATE users SET max_file_override_bytes = $2 WHERE id = $1",
186 + )
187 + .bind(user_id)
188 + .bind(bytes)
189 + .execute(pool)
190 + .await?;
191 +
192 + Ok(())
193 + }
194 +
195 + /// Get the grandfathering deadline for a user.
196 + #[tracing::instrument(skip_all)]
197 + pub async fn get_grandfathered_until(
198 + pool: &PgPool,
199 + user_id: UserId,
200 + ) -> Result<Option<DateTime<Utc>>> {
201 + let val: Option<DateTime<Utc>> = sqlx::query_scalar(
202 + "SELECT grandfathered_until FROM users WHERE id = $1",
203 + )
204 + .bind(user_id)
205 + .fetch_one(pool)
206 + .await?;
207 +
208 + Ok(val)
209 + }
210 +
211 + /// Get a per-category storage breakdown for the creator dashboard (single query).
212 + #[tracing::instrument(skip_all)]
213 + pub async fn get_storage_breakdown(pool: &PgPool, user_id: UserId) -> Result<StorageBreakdown> {
214 + // Categories must partition every byte counted by `recalculate_all_storage_batch`
215 + // so the dashboard total reconciles with `storage_used_bytes`. `cover_bytes`
216 + // folds item covers + project covers; `gallery_bytes` covers the item/project
217 + // image carousels (Run #18 Storage B1).
218 + let row: (i64, i64, i64, i64, i64, i64, i64) = sqlx::query_as(
219 + r#"
220 + WITH audio_bytes AS (
221 + SELECT COALESCE(GREATEST(0, LEAST(SUM(i.audio_file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
222 + FROM items i JOIN projects p ON i.project_id = p.id
223 + WHERE p.user_id = $1 AND i.audio_file_size_bytes IS NOT NULL
224 + ),
225 + cover_bytes AS (
226 + SELECT COALESCE(GREATEST(0, LEAST(SUM(i.cover_file_size_bytes), 9223372036854775807))::BIGINT, 0)
227 + + COALESCE((
228 + SELECT GREATEST(0, LEAST(SUM(p.cover_image_size_bytes), 9223372036854775807))::BIGINT
229 + FROM projects p
230 + WHERE p.user_id = $1 AND p.cover_image_size_bytes IS NOT NULL
231 + ), 0) AS total
232 + FROM items i JOIN projects p ON i.project_id = p.id
233 + WHERE p.user_id = $1 AND i.cover_file_size_bytes IS NOT NULL
234 + ),
235 + version_bytes AS (
236 + SELECT COALESCE(GREATEST(0, LEAST(SUM(v.file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
237 + FROM versions v
238 + JOIN items i ON v.item_id = i.id
239 + JOIN projects p ON i.project_id = p.id
240 + WHERE p.user_id = $1 AND v.file_size_bytes IS NOT NULL
241 + ),
242 + insertion_bytes AS (
243 + SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size), 9223372036854775807))::BIGINT, 0) AS total
244 + FROM content_insertions WHERE user_id = $1
245 + ),
246 + video_bytes AS (
247 + SELECT COALESCE(GREATEST(0, LEAST(SUM(i.video_file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
248 + FROM items i JOIN projects p ON i.project_id = p.id
249 + WHERE p.user_id = $1 AND i.video_file_size_bytes IS NOT NULL
250 + ),
251 + media_bytes AS (
252 + SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size_bytes), 9223372036854775807))::BIGINT, 0) AS total
253 + FROM media_files WHERE user_id = $1
254 + ),
255 + gallery_bytes AS (
256 + SELECT COALESCE((
257 + SELECT GREATEST(0, LEAST(SUM(ii.file_size_bytes), 9223372036854775807))::BIGINT
258 + FROM item_images ii JOIN items i ON ii.item_id = i.id JOIN projects p ON i.project_id = p.id
259 + WHERE p.user_id = $1
260 + ), 0)
261 + + COALESCE((
262 + SELECT GREATEST(0, LEAST(SUM(pi.file_size_bytes), 9223372036854775807))::BIGINT
263 + FROM project_images pi JOIN projects p ON pi.project_id = p.id
264 + WHERE p.user_id = $1
265 + ), 0) AS total
266 + )
267 + SELECT
268 + (SELECT total FROM audio_bytes),
269 + (SELECT total FROM cover_bytes),
270 + (SELECT total FROM version_bytes),
271 + (SELECT total FROM insertion_bytes),
272 + (SELECT total FROM video_bytes),
273 + (SELECT total FROM media_bytes),
274 + (SELECT total FROM gallery_bytes)
275 + "#,
276 + )
277 + .bind(user_id)
278 + .fetch_one(pool)
279 + .await?;
280 +
281 + Ok(StorageBreakdown {
282 + audio_bytes: row.0,
283 + cover_bytes: row.1,
284 + download_bytes: row.2,
285 + insertion_bytes: row.3,
286 + video_bytes: row.4,
287 + media_bytes: row.5,
288 + gallery_bytes: row.6,
289 + // Each per-category SUM is SQL-clamped to [0, i64::MAX]; sum them
290 + // saturating so seven near-max categories can't overflow the total.
291 + total_bytes: [row.0, row.1, row.2, row.3, row.4, row.5, row.6]
292 + .into_iter()
293 + .fold(0i64, i64::saturating_add),
294 + })
295 + }
296 +
297 + /// Creators per batch for the weekly storage recalc. Bounds each UPDATE's
298 + /// LATERAL SUM fan-out so a large creator base can't produce one multi-minute
299 + /// statement that trips the scheduler overrun alert; the connection is released
300 + /// between batches (ultra-fuzz Run 10 Perf S3).
301 + const STORAGE_RECALC_BATCH: usize = 500;
302 +
303 + /// IDs of every creator the storage recalc covers.
304 + async fn get_storage_recalc_creator_ids(pool: &PgPool) -> Result<Vec<UserId>> {
305 + let ids: Vec<UserId> =
306 + sqlx::query_scalar("SELECT id FROM users WHERE can_create_projects = true ORDER BY id")
307 + .fetch_all(pool)
308 + .await?;
309 + Ok(ids)
310 + }
311 +
312 + /// Weekly storage drift correction. Chunks creators into bounded batches so each
313 + /// statement stays short, returning the total number of corrected rows.
314 + #[tracing::instrument(skip_all)]
315 + pub async fn recalculate_all_storage_batch(pool: &PgPool) -> Result<u64> {
316 + let creator_ids = get_storage_recalc_creator_ids(pool).await?;
317 + let mut corrected = 0u64;
318 + for batch in creator_ids.chunks(STORAGE_RECALC_BATCH) {
319 + corrected += recalculate_storage_for_user_batch(pool, batch).await?;
320 + }
321 + Ok(corrected)
322 + }
323 +
324 + /// Recalculate `storage_used_bytes` for a specific batch of creators.
325 + async fn recalculate_storage_for_user_batch(pool: &PgPool, user_ids: &[UserId]) -> Result<u64> {
326 + if user_ids.is_empty() {
327 + return Ok(0);
328 + }
329 + let result = sqlx::query(
330 + r#"
331 + UPDATE users SET storage_used_bytes = totals.total
332 + FROM (
333 + SELECT u.id AS user_id,
334 + COALESCE(audio.total, 0)
335 + + COALESCE(cover.total, 0)
336 + + COALESCE(video.total, 0)
337 + + COALESCE(versions.total, 0)
338 + + COALESCE(insertions.total, 0)
339 + + COALESCE(media.total, 0)
340 + + COALESCE(project_cover.total, 0)
341 + + COALESCE(item_gallery.total, 0)
342 + + COALESCE(project_gallery.total, 0) AS total
343 + FROM users u
344 + LEFT JOIN LATERAL (
345 + SELECT GREATEST(0, LEAST(SUM(i.audio_file_size_bytes), 9223372036854775807))::BIGINT AS total
346 + FROM items i JOIN projects p ON i.project_id = p.id
347 + WHERE p.user_id = u.id AND i.audio_file_size_bytes IS NOT NULL
348 + ) audio ON true
349 + LEFT JOIN LATERAL (
350 + SELECT GREATEST(0, LEAST(SUM(i.cover_file_size_bytes), 9223372036854775807))::BIGINT AS total
351 + FROM items i JOIN projects p ON i.project_id = p.id
352 + WHERE p.user_id = u.id AND i.cover_file_size_bytes IS NOT NULL
353 + ) cover ON true
354 + LEFT JOIN LATERAL (
355 + SELECT GREATEST(0, LEAST(SUM(i.video_file_size_bytes), 9223372036854775807))::BIGINT AS total
356 + FROM items i JOIN projects p ON i.project_id = p.id
357 + WHERE p.user_id = u.id AND i.video_file_size_bytes IS NOT NULL
358 + ) video ON true
359 + LEFT JOIN LATERAL (
360 + SELECT GREATEST(0, LEAST(SUM(v.file_size_bytes), 9223372036854775807))::BIGINT AS total
361 + FROM versions v JOIN items i ON v.item_id = i.id JOIN projects p ON i.project_id = p.id
362 + WHERE p.user_id = u.id AND v.file_size_bytes IS NOT NULL
363 + ) versions ON true
364 + LEFT JOIN LATERAL (
365 + SELECT GREATEST(0, LEAST(SUM(ci.file_size), 9223372036854775807))::BIGINT AS total
366 + FROM content_insertions ci WHERE ci.user_id = u.id
367 + ) insertions ON true
368 + LEFT JOIN LATERAL (
369 + SELECT GREATEST(0, LEAST(SUM(mf.file_size_bytes), 9223372036854775807))::BIGINT AS total
370 + FROM media_files mf WHERE mf.user_id = u.id
371 + ) media ON true
372 + -- Project cover images charge storage at confirm but live in their own
373 + -- column (projects.cover_image_size_bytes, migration 126), not on items;
374 + -- the gallery carousels (item_images / project_images, migration 135)
375 + -- likewise charge at confirm and decrement on delete. All three were
376 + -- omitted from this recalc, so the weekly drift-correction zeroed every
377 + -- gallery/project-cover charge and the counter oscillated week to week
378 + -- (Run #18 Storage B1). Reconcile them from the same rows the charge writes.
379 + LEFT JOIN LATERAL (
380 + SELECT GREATEST(0, LEAST(SUM(p.cover_image_size_bytes), 9223372036854775807))::BIGINT AS total
381 + FROM projects p
382 + WHERE p.user_id = u.id AND p.cover_image_size_bytes IS NOT NULL
383 + ) project_cover ON true
384 + LEFT JOIN LATERAL (
385 + SELECT GREATEST(0, LEAST(SUM(ii.file_size_bytes), 9223372036854775807))::BIGINT AS total
386 + FROM item_images ii JOIN items i ON ii.item_id = i.id JOIN projects p ON i.project_id = p.id
387 + WHERE p.user_id = u.id
388 + ) item_gallery ON true
389 + LEFT JOIN LATERAL (
390 + SELECT GREATEST(0, LEAST(SUM(pi.file_size_bytes), 9223372036854775807))::BIGINT AS total
391 + FROM project_images pi JOIN projects p ON pi.project_id = p.id
392 + WHERE p.user_id = u.id
393 + ) project_gallery ON true
394 + WHERE u.can_create_projects = true AND u.id = ANY($1)
395 + ) totals
396 + WHERE users.id = totals.user_id AND users.storage_used_bytes IS DISTINCT FROM totals.total
397 + "#,
398 + )
399 + .bind(user_ids)
400 + .execute(pool)
401 + .await?;
402 +
403 + Ok(result.rows_affected())
404 + }
405 +
406 + // ============================================================================
407 + // Enforcement
408 + // ============================================================================
409 +
410 + /// Check whether a file upload is allowed based on the user's tier, storage,
411 + /// and grandfathering status. Returns the tier's `max_storage_bytes` on success
412 + /// (for use with `try_increment_storage`), or an appropriate `AppError` if rejected.
413 + ///
414 + /// Covers and media images bypass tier checks but respect their size limits
415 + /// enforced separately in the storage module.
416 + #[tracing::instrument(skip_all)]
417 + pub async fn check_upload_allowed(
418 + pool: &PgPool,
419 + user_id: UserId,
420 + file_type: FileType,
421 + file_size_bytes: i64,
422 + ) -> Result<i64> {
423 + // Covers and media images bypass per-file tier checks but still respect
424 + // the storage cap. Look up the active tier (fallback to Basic cap).
425 + if file_type == FileType::Cover || file_type == FileType::MediaImage {
426 + let active_tier = get_active_creator_tier(pool, user_id).await?;
427 + let max_storage = active_tier
428 + .map(|t| t.max_storage_bytes())
429 + .unwrap_or_else(|| CreatorTier::Basic.max_storage_bytes());
430 + return Ok(max_storage);
431 + }
432 +
433 + // Resolve effective tier. These two lookups are independent, so run them
434 + // concurrently rather than sequentially on the per-upload hot path.
435 + let (active_tier, grandfathered) = tokio::try_join!(
436 + get_active_creator_tier(pool, user_id),
437 + get_grandfathered_until(pool, user_id),
438 + )?;
439 +
440 + let effective_tier = match active_tier {
441 + Some(tier) => Some(tier),
442 + None => {
443 + // Check grandfathering
444 + if let Some(until) = grandfathered {
445 + if Utc::now() < until {
446 + Some(CreatorTier::SmallFiles) // grandfathered as SmallFiles-equivalent
447 + } else {
448 + None
449 + }
450 + } else {
451 + None
452 + }
453 + }
454 + };
455 +
456 + // Grace period check (canceled sub, within 30 days)
457 + // Use effective_tier so grandfathered users aren't blocked
458 + if effective_tier.is_none() {
459 + let in_grace = is_in_grace_period(pool, user_id).await?;
460 + if in_grace {
461 + return Err(AppError::BadRequest(
462 + "Your creator subscription has been canceled. Re-subscribe to upload files.".to_string(),
463 + ));
464 + }
465 + }
466 +
467 + // No tier and not grandfathered → reject
468 + let tier = match effective_tier {
469 + Some(t) => t,
470 + None => {
471 + return Err(AppError::BadRequest(
472 + "A creator tier subscription is required to upload files.".to_string(),
473 + ));
474 + }
475 + };
476 +
477 + // Basic tier is text-only (no non-cover uploads)
478 + if !tier.allows_file_uploads() {
479 + return Err(AppError::BadRequest(
480 + "Basic tier is text-only. Upgrade to Small Files or higher to upload files.".to_string(),
481 + ));
482 + }
483 +
484 + // The per-file-override and current-usage lookups are independent, so fetch
485 + // them concurrently rather than as two serial round-trips (audit Run 17 Perf).
486 + let (max_override, used) = tokio::try_join!(
487 + get_max_file_override(pool, user_id),
488 + get_storage_used(pool, user_id),
489 + )?;
490 +
491 + // Per-file size check
492 + let max_file = max_override.unwrap_or(tier.max_file_bytes());
493 + if file_size_bytes > max_file {
494 + return Err(AppError::FileTooLarge(format!(
495 + "File size ({}) exceeds the {} per-file limit of {}.",
496 + format_bytes(file_size_bytes),
497 + tier.label(),
498 + format_bytes(max_file),
499 + )));
500 + }
Lines truncated
A server/src/db/creator_tiers/subscriptions.rs +274
@@ -0,0 +1,274 @@
1 + use super::*;
2 +
3 + /// Create or reactivate a creator tier subscription record.
4 + ///
5 + /// Uses ON CONFLICT DO UPDATE on the user_id unique index to handle
6 + /// both duplicate webhooks and re-subscription after cancellation.
7 + /// Returns `None` if the row already existed with the same stripe_subscription_id
8 + /// (duplicate webhook), `Some` if this was a fresh insert or a re-subscription
9 + /// with a different subscription ID.
10 + #[tracing::instrument(skip_all)]
11 + pub async fn create_creator_subscription<'e>(
12 + executor: impl sqlx::PgExecutor<'e>,
13 + user_id: UserId,
14 + stripe_subscription_id: &str,
15 + stripe_customer_id: &str,
16 + tier: CreatorTier,
17 + ) -> Result<Option<DbCreatorSubscription>> {
18 + // Use WHERE clause on the DO UPDATE to only update if the subscription_id
19 + // is different (new subscription) or status is not already active.
20 + // When the WHERE fails, DO UPDATE becomes a no-op and RETURNING yields no row.
21 + let sub = sqlx::query_as::<_, DbCreatorSubscription>(
22 + r#"
23 + INSERT INTO creator_subscriptions (user_id, stripe_subscription_id, stripe_customer_id, tier)
24 + VALUES ($1, $2, $3, $4)
25 + ON CONFLICT (user_id) DO UPDATE
26 + SET stripe_subscription_id = EXCLUDED.stripe_subscription_id,
27 + stripe_customer_id = EXCLUDED.stripe_customer_id,
28 + tier = EXCLUDED.tier,
29 + status = 'active',
30 + canceled_at = NULL,
31 + grace_enforced_at = NULL
32 + WHERE creator_subscriptions.stripe_subscription_id != EXCLUDED.stripe_subscription_id
33 + OR creator_subscriptions.status != 'active'
34 + RETURNING *
35 + "#,
36 + )
37 + .bind(user_id)
38 + .bind(stripe_subscription_id)
39 + .bind(stripe_customer_id)
40 + .bind(tier)
41 + .fetch_optional(executor)
42 + .await?;
43 +
44 + Ok(sub)
45 + }
46 +
47 + /// Look up a creator subscription by its Stripe subscription ID.
48 + #[tracing::instrument(skip_all)]
49 + pub async fn get_creator_sub_by_stripe_id(
50 + pool: &PgPool,
51 + stripe_subscription_id: &str,
52 + ) -> Result<Option<DbCreatorSubscription>> {
53 + let sub = sqlx::query_as::<_, DbCreatorSubscription>(
54 + "SELECT * FROM creator_subscriptions WHERE stripe_subscription_id = $1",
55 + )
56 + .bind(stripe_subscription_id)
57 + .fetch_optional(pool)
58 + .await?;
59 +
60 + Ok(sub)
61 + }
62 +
63 + /// Get a user's creator subscription (any status).
64 + #[tracing::instrument(skip_all)]
65 + pub async fn get_creator_sub_by_user(
66 + pool: &PgPool,
67 + user_id: UserId,
68 + ) -> Result<Option<DbCreatorSubscription>> {
69 + let sub = sqlx::query_as::<_, DbCreatorSubscription>(
70 + "SELECT * FROM creator_subscriptions WHERE user_id = $1",
71 + )
72 + .bind(user_id)
73 + .fetch_optional(pool)
74 + .await?;
75 +
76 + Ok(sub)
77 + }
78 +
79 + /// Get the active creator tier for a user (None if no active subscription).
80 + #[tracing::instrument(skip_all)]
81 + pub async fn get_active_creator_tier(
82 + pool: &PgPool,
83 + user_id: UserId,
84 + ) -> Result<Option<CreatorTier>> {
85 + let tier = sqlx::query_scalar::<_, String>(
86 + "SELECT tier FROM creator_subscriptions WHERE user_id = $1 AND status = 'active'",
87 + )
88 + .bind(user_id)
89 + .fetch_optional(pool)
90 + .await?;
91 +
92 + match tier {
93 + None => Ok(None),
94 + // A present-but-unparseable tier means the DB holds a tier string the
95 + // enum doesn't know (enum drift). Silently mapping that to `None` would
96 + // strip a paying creator's entitlements with no signal — surface it
97 + // loudly instead of swallowing it (Run 21). The `enum_drift` test and the
98 + // DB CHECK constraint make this unreachable in practice.
99 + Some(t) => match t.parse::<CreatorTier>() {
100 + Ok(parsed) => Ok(Some(parsed)),
101 + Err(_) => {
102 + tracing::error!(
103 + user_id = %user_id, tier = %t,
104 + "active creator subscription has an unrecognized tier string (enum drift)"
105 + );
106 + Err(crate::error::AppError::Internal(anyhow::anyhow!(
107 + "unrecognized creator tier '{t}' for user {user_id}"
108 + )))
109 + }
110 + },
111 + }
112 + }
113 +
114 + // Apply a Stripe-driven status and/or period update in one guarded statement.
115 + // `canceled` is terminal; reactivation runs through `create_creator_subscription`'s
116 + // `ON CONFLICT (user_id) DO UPDATE` at checkout, never through this path. Replaces
117 + // the old split status/period setters (the period half lacked the guard). See
118 + // `crate::db::subscription_writer`.
119 + crate::db::subscription_writer::define_stripe_subscription_writer!(
120 + apply_stripe_update,
121 + "creator_subscriptions",
122 + DbCreatorSubscription
123 + );
124 +
125 + /// Cancel a creator subscription (set status + canceled_at).
126 + #[tracing::instrument(skip_all)]
127 + pub async fn cancel_creator_sub(
128 + pool: &PgPool,
129 + stripe_subscription_id: &str,
130 + ) -> Result<Option<DbCreatorSubscription>> {
131 + let sub = sqlx::query_as::<_, DbCreatorSubscription>(
132 + r#"
133 + UPDATE creator_subscriptions
134 + SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
135 + WHERE stripe_subscription_id = $1
136 + RETURNING *
137 + "#,
138 + )
139 + .bind(stripe_subscription_id)
140 + .fetch_optional(pool)
141 + .await?;
142 +
143 + Ok(sub)
144 + }
145 +
146 + /// Sync the users.creator_tier column from the subscription status.
147 + /// Called after checkout/update/cancel to keep the denormalized column in sync.
148 + #[tracing::instrument(skip_all)]
149 + pub async fn sync_user_creator_tier(pool: &PgPool, user_id: UserId) -> Result<()> {
150 + sqlx::query(
151 + r#"
152 + UPDATE users SET creator_tier = (
153 + SELECT tier FROM creator_subscriptions
154 + WHERE user_id = $1 AND status = 'active'
155 + )
156 + WHERE id = $1
157 + "#,
158 + )
159 + .bind(user_id)
160 + .execute(pool)
161 + .await?;
162 +
163 + Ok(())
164 + }
165 +
166 + /// Get user IDs of creators with canceled subscriptions 30+ days ago
167 + /// whose items have not yet been hidden.
168 + #[tracing::instrument(skip_all)]
169 + pub async fn get_expired_grace_creators(pool: &PgPool) -> Result<Vec<UserId>> {
170 + // Bounded batch per call. The scheduler enforces these inline on the tick
171 + // (two DB round-trips per creator), so an unbounded result set would let a
172 + // backlog stall the tick. `grace_enforced_at` is set as each creator is
173 + // processed, so successive ticks drain the rest; ORDER BY oldest-first keeps
174 + // it deterministic and starvation-free.
175 + let ids: Vec<UserId> = sqlx::query_scalar(
176 + r#"
177 + SELECT user_id FROM creator_subscriptions
178 + WHERE status = 'canceled'
179 + AND canceled_at IS NOT NULL
180 + AND canceled_at < NOW() - INTERVAL '30 days'
181 + AND grace_enforced_at IS NULL
182 + ORDER BY canceled_at ASC
183 + LIMIT 200
184 + "#,
185 + )
186 + .fetch_all(pool)
187 + .await?;
188 +
189 + Ok(ids)
190 + }
191 +
192 + /// Stamp `grace_enforced_at` for all given creators in one statement, paired with
193 + /// `items::hide_all_items_for_users` on the post-grace sweep (Perf-S4, Run 9).
194 + /// No-op on an empty slice.
195 + #[tracing::instrument(skip_all)]
196 + pub async fn mark_grace_enforced_batch(pool: &PgPool, user_ids: &[UserId]) -> Result<()> {
197 + if user_ids.is_empty() {
198 + return Ok(());
199 + }
200 + sqlx::query(
201 + "UPDATE creator_subscriptions SET grace_enforced_at = NOW() WHERE user_id = ANY($1)",
202 + )
203 + .bind(user_ids)
204 + .execute(pool)
205 + .await?;
206 +
207 + Ok(())
208 + }
209 +
210 + /// Count fully-paying creators — `status = 'active'` only.
211 + ///
212 + /// Excludes trialing (free trial), past_due (payment failed but not yet
213 + /// canceled), canceled-in-grace (winding down), and incomplete states.
214 + /// This is the number that goes on the runway disclosure as "paying
215 + /// creators today": revenue-bearing seats, no fudge.
216 + #[tracing::instrument(skip_all)]
217 + pub async fn count_active_paying(pool: &PgPool) -> Result<i64> {
218 + let count: (i64,) = sqlx::query_as(
219 + "SELECT COUNT(*) FROM creator_subscriptions WHERE status = 'active'",
220 + )
221 + .fetch_one(pool)
222 + .await?;
223 + Ok(count.0)
224 + }
225 +
226 + /// Count creators in a trial or 30-day cancellation grace period.
227 + ///
228 + /// These are not revenue-bearing today but represent the near-term
229 + /// pipeline: trialing seats may convert, grace seats may resubscribe
230 + /// before enforcement. Disclosed as a secondary number on the runway
231 + /// surface so the headline `count_active_paying` stays strict.
232 + #[tracing::instrument(skip_all)]
233 + pub async fn count_trialing_or_grace(pool: &PgPool) -> Result<i64> {
234 + let count: (i64,) = sqlx::query_as(
235 + r#"
236 + SELECT COUNT(*) FROM creator_subscriptions
237 + WHERE status = 'trialing'
238 + OR (
239 + status = 'canceled'
240 + AND canceled_at IS NOT NULL
241 + AND canceled_at > NOW() - INTERVAL '30 days'
242 + AND grace_enforced_at IS NULL
243 + )
244 + "#,
245 + )
246 + .fetch_one(pool)
247 + .await?;
248 + Ok(count.0)
249 + }
250 +
251 + /// Check whether a user is in the 30-day cancellation grace period.
252 + ///
253 + /// Returns `true` if the subscription is canceled but within 30 days of cancellation
254 + /// and enforcement has not yet been applied.
255 + #[tracing::instrument(skip_all)]
256 + pub async fn is_in_grace_period(pool: &PgPool, user_id: UserId) -> Result<bool> {
257 + let in_grace: bool = sqlx::query_scalar(
258 + r#"
259 + SELECT EXISTS(
260 + SELECT 1 FROM creator_subscriptions
261 + WHERE user_id = $1
262 + AND status = 'canceled'
263 + AND canceled_at IS NOT NULL
264 + AND canceled_at > NOW() - INTERVAL '30 days'
265 + AND grace_enforced_at IS NULL
266 + )
267 + "#,
268 + )
269 + .bind(user_id)
270 + .fetch_one(pool)
271 + .await?;
272 +
273 + Ok(in_grace)
274 + }
A server/src/db/creator_tiers/tests.rs +232
@@ -0,0 +1,232 @@
1 + use super::*;
2 +
3 + // ── CreatorTier::label ───────────────────────────────────────────────
4 +
5 + #[test]
6 + fn label_basic() {
7 + assert_eq!(CreatorTier::Basic.label(), "Basic");
8 + }
9 +
10 + #[test]
11 + fn label_small_files() {
12 + assert_eq!(CreatorTier::SmallFiles.label(), "Small Files");
13 + }
14 +
15 + #[test]
16 + fn label_big_files() {
17 + assert_eq!(CreatorTier::BigFiles.label(), "Big Files");
18 + }
19 +
20 + #[test]
21 + fn label_everything() {
22 + assert_eq!(CreatorTier::Everything.label(), "Everything");
23 + }
24 +
25 + // ── CreatorTier price + envelope invariants ────────────────────────
26 + //
27 + // Concrete cents/bytes come from `assumptions.toml` via the installed
28 + // `TierPrices` global. Tests pin only structural invariants (positive,
29 + // monotone, per-file ≤ storage) so a future toml edit doesn't rewrite
30 + // this file. Literal-value pins live in `docs/business/assumptions.toml`
31 + // itself and are guarded by the docengine `tier_bytes ↔ tier_limits`
32 + // validator.
33 +
34 + fn all_tiers() -> [CreatorTier; 4] {
35 + [
36 + CreatorTier::Basic,
37 + CreatorTier::SmallFiles,
38 + CreatorTier::BigFiles,
39 + CreatorTier::Everything,
40 + ]
41 + }
42 +
43 + #[test]
44 + fn prices_positive_and_strictly_increasing() {
45 + crate::tier_prices::TierPrices::install_test_default();
46 + let tiers = all_tiers();
47 + assert!(tiers[0].price_cents() > 0);
48 + for pair in tiers.windows(2) {
49 + assert!(
50 + pair[0].price_cents() < pair[1].price_cents(),
51 + "{:?} should cost less than {:?}",
52 + pair[0], pair[1],
53 + );
54 + }
55 + }
56 +
57 + #[test]
58 + fn max_file_bytes_non_decreasing() {
59 + crate::tier_prices::TierPrices::install_test_default();
60 + let tiers = all_tiers();
61 + assert!(tiers[0].max_file_bytes() > 0);
62 + for pair in tiers.windows(2) {
63 + assert!(
64 + pair[0].max_file_bytes() <= pair[1].max_file_bytes(),
65 + "{:?} file limit should not exceed {:?}",
66 + pair[0], pair[1],
67 + );
68 + }
69 + }
70 +
71 + #[test]
72 + fn max_storage_bytes_non_decreasing() {
73 + crate::tier_prices::TierPrices::install_test_default();
74 + let tiers = all_tiers();
75 + assert!(tiers[0].max_storage_bytes() > 0);
76 + for pair in tiers.windows(2) {
77 + assert!(
78 + pair[0].max_storage_bytes() <= pair[1].max_storage_bytes(),
79 + "{:?} storage limit should not exceed {:?}",
80 + pair[0], pair[1],
81 + );
82 + }
83 + }
84 +
85 + #[test]
86 + fn top_tier_matches_big_files_envelope() {
87 + // BigFiles and Everything share the same envelope by product design.
88 + // (Everything differentiates on features, not caps.)
89 + crate::tier_prices::TierPrices::install_test_default();
90 + assert_eq!(
91 + CreatorTier::Everything.max_file_bytes(),
92 + CreatorTier::BigFiles.max_file_bytes(),
93 + );
94 + assert_eq!(
95 + CreatorTier::Everything.max_storage_bytes(),
96 + CreatorTier::BigFiles.max_storage_bytes(),
97 + );
98 + }
99 +
100 + // ── CreatorTier::allows_file_uploads ─────────────────────────────────
101 +
102 + #[test]
103 + fn basic_tier_disallows_file_uploads() {
104 + assert!(!CreatorTier::Basic.allows_file_uploads());
105 + }
106 +
107 + #[test]
108 + fn small_files_allows_file_uploads() {
109 + assert!(CreatorTier::SmallFiles.allows_file_uploads());
110 + }
111 +
112 + #[test]
113 + fn big_files_allows_file_uploads() {
114 + assert!(CreatorTier::BigFiles.allows_file_uploads());
115 + }
116 +
117 + #[test]
118 + fn everything_allows_file_uploads() {
119 + assert!(CreatorTier::Everything.allows_file_uploads());
120 + }
121 +
122 + // ── format_bytes helper ─────────────────────────────────────────────
123 +
124 + #[test]
125 + fn format_bytes_zero() {
126 + assert_eq!(format_bytes(0), "0 B");
127 + }
128 +
129 + #[test]
130 + fn format_bytes_one_byte() {
131 + assert_eq!(format_bytes(1), "1 B");
132 + }
133 +
134 + #[test]
135 + fn format_bytes_below_kb() {
136 + assert_eq!(format_bytes(1023), "1023 B");
137 + }
138 +
139 + #[test]
140 + fn format_bytes_exactly_1kb() {
141 + assert_eq!(format_bytes(1024), "1.0 KB");
142 + }
143 +
144 + #[test]
145 + fn format_bytes_exactly_1mb() {
146 + assert_eq!(format_bytes(1024 * 1024), "1.0 MB");
147 + }
148 +
149 + #[test]
150 + fn format_bytes_exactly_1gb() {
151 + assert_eq!(format_bytes(1024 * 1024 * 1024), "1.0 GB");
152 + }
153 +
154 + #[test]
155 + fn format_bytes_negative_clamped_to_zero() {
156 + assert_eq!(format_bytes(-999), "0 B");
157 + }
158 +
159 + #[test]
160 + fn format_bytes_large_storage_cap() {
161 + // 500 GB (Everything tier cap)
162 + assert_eq!(format_bytes(500 * 1024 * 1024 * 1024), "500.0 GB");
163 + }
164 +
165 + // ── StorageBreakdown ────────────────────────────────────────────────
166 +
167 + #[test]
168 + fn storage_breakdown_default_is_all_zeros() {
169 + let sb = StorageBreakdown::default();
170 + assert_eq!(sb.audio_bytes, 0);
171 + assert_eq!(sb.cover_bytes, 0);
172 + assert_eq!(sb.download_bytes, 0);
173 + assert_eq!(sb.insertion_bytes, 0);
174 + assert_eq!(sb.video_bytes, 0);
175 + assert_eq!(sb.media_bytes, 0);
176 + assert_eq!(sb.total_bytes, 0);
177 + }
178 +
179 + #[test]
180 + fn storage_breakdown_total_is_sum_of_categories() {
181 + let sb = StorageBreakdown {
182 + audio_bytes: 100,
183 + cover_bytes: 200,
184 + download_bytes: 300,
185 + insertion_bytes: 400,
186 + video_bytes: 500,
187 + media_bytes: 600,
188 + gallery_bytes: 700,
189 + total_bytes: 100 + 200 + 300 + 400 + 500 + 600 + 700,
190 + };
191 + assert_eq!(
192 + sb.total_bytes,
193 + sb.audio_bytes + sb.cover_bytes + sb.download_bytes
194 + + sb.insertion_bytes + sb.video_bytes + sb.media_bytes + sb.gallery_bytes,
195 + );
196 + }
197 +
198 + #[test]
199 + fn storage_breakdown_single_category() {
200 + let sb = StorageBreakdown {
201 + audio_bytes: 1_000_000,
202 + total_bytes: 1_000_000,
203 + ..Default::default()
204 + };
205 + assert_eq!(sb.total_bytes, 1_000_000);
206 + assert_eq!(sb.cover_bytes, 0);
207 + }
208 +
209 + // ── Boundary / cross-cutting ────────────────────────────────────────
210 +
211 + #[test]
212 + fn basic_file_limit_less_than_storage_limit() {
213 + crate::tier_prices::TierPrices::install_test_default();
214 + assert!(CreatorTier::Basic.max_file_bytes() < CreatorTier::Basic.max_storage_bytes());
215 + }
216 +
217 + #[test]
218 + fn every_tier_file_limit_within_storage_limit() {
219 + crate::tier_prices::TierPrices::install_test_default();
220 + for tier in [
221 + CreatorTier::Basic,
222 + CreatorTier::SmallFiles,
223 + CreatorTier::BigFiles,
224 + CreatorTier::Everything,
225 + ] {
226 + assert!(
227 + tier.max_file_bytes() <= tier.max_storage_bytes(),
228 + "{:?} file limit exceeds its own storage limit",
229 + tier,
230 + );
231 + }
232 + }