Skip to main content

max / makenotwork

39.8 KB · 1318 lines History Blame Raw
1 //! User account CRUD, profile updates, and lookup queries.
2
3 use sqlx::PgPool;
4
5 use super::UserId;
6 use super::enums::AppealDecision;
7 use super::models::DbUser;
8 use super::validated_types::{Email, Username};
9 use crate::error::Result;
10
11 /// Insert a new user and return the created row.
12 #[tracing::instrument(skip_all)]
13 pub async fn create_user(
14 pool: &PgPool,
15 username: &Username,
16 email: &Email,
17 password_hash: &str,
18 ) -> Result<DbUser> {
19 let user = sqlx::query_as::<_, DbUser>(
20 r"
21 INSERT INTO users (username, email, password_hash)
22 VALUES ($1, $2, $3)
23 RETURNING *
24 ",
25 )
26 .bind(username)
27 .bind(email)
28 .bind(password_hash)
29 .fetch_one(pool)
30 .await?;
31
32 Ok(user)
33 }
34
35 /// Insert a durable example-marketplace creator (see [`crate::seed`]).
36 ///
37 /// Unlike [`create_sandbox_user`], this leaves `is_sandbox` at its `FALSE`
38 /// default so the account and its projects appear on every public surface
39 /// (discover/browse/search gate only on `is_sandbox = FALSE`). It grants
40 /// `can_create_projects`, marks the email verified, and pins the top
41 /// `creator_tier` so no capability gate blocks the item spread seeded in later
42 /// phases. Only ever called by the `--seed-examples` flow, which is itself
43 /// confined to testnot/localhost by [`crate::seed::run`]'s guards.
44 #[tracing::instrument(skip_all)]
45 pub async fn create_example_creator(
46 pool: &PgPool,
47 username: &Username,
48 email: &Email,
49 password_hash: &str,
50 ) -> Result<DbUser> {
51 let user = sqlx::query_as::<_, DbUser>(
52 r"
53 INSERT INTO users (
54 username, email, password_hash,
55 can_create_projects, email_verified, creator_tier
56 )
57 VALUES ($1, $2, $3, TRUE, TRUE, 'everything')
58 RETURNING *
59 ",
60 )
61 .bind(username)
62 .bind(email)
63 .bind(password_hash)
64 .fetch_one(pool)
65 .await?;
66
67 Ok(user)
68 }
69
70 /// Fetch a user by primary key. Returns `None` if not found.
71 #[tracing::instrument(skip_all)]
72 pub async fn get_user_by_id(pool: &PgPool, id: UserId) -> Result<Option<DbUser>> {
73 let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE id = $1")
74 .bind(id)
75 .fetch_optional(pool)
76 .await?;
77
78 Ok(user)
79 }
80
81 /// Fetch multiple users by ID in a single query.
82 #[tracing::instrument(skip_all)]
83 pub async fn get_users_by_ids(pool: &PgPool, ids: &[UserId]) -> Result<Vec<DbUser>> {
84 let users = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE id = ANY($1)")
85 .bind(ids)
86 .fetch_all(pool)
87 .await?;
88 Ok(users)
89 }
90
91 /// Fetch a user by username. Returns `None` if not found.
92 #[tracing::instrument(skip_all)]
93 pub async fn get_user_by_username(pool: &PgPool, username: &Username) -> Result<Option<DbUser>> {
94 let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE username = $1")
95 .bind(username)
96 .fetch_optional(pool)
97 .await?;
98
99 Ok(user)
100 }
101
102 /// Fetch a user by email address. Returns `None` if not found.
103 #[tracing::instrument(skip_all)]
104 pub async fn get_user_by_email(pool: &PgPool, email: &Email) -> Result<Option<DbUser>> {
105 let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE email = $1")
106 .bind(email)
107 .fetch_optional(pool)
108 .await?;
109
110 Ok(user)
111 }
112
113 /// Update a user's display name and/or bio (COALESCE keeps existing values when `None`).
114 #[tracing::instrument(skip_all)]
115 pub async fn update_user_profile(
116 pool: &PgPool,
117 id: UserId,
118 display_name: Option<&str>,
119 bio: Option<&str>,
120 ) -> Result<DbUser> {
121 let user = sqlx::query_as::<_, DbUser>(
122 r"
123 UPDATE users
124 SET display_name = COALESCE($2, display_name),
125 bio = COALESCE($3, bio)
126 WHERE id = $1
127 RETURNING *
128 ",
129 )
130 .bind(id)
131 .bind(display_name)
132 .bind(bio)
133 .fetch_one(pool)
134 .await?;
135
136 Ok(user)
137 }
138
139 /// Store a user's custom profile-page source (the original, pre-sanitization),
140 /// stamp `custom_pages_updated_at`, and bump the cache generation. The source is
141 /// re-sanitized on render; see [`crate::custom_pages`].
142 #[tracing::instrument(skip_all, fields(user_id = %id))]
143 pub async fn update_user_custom_page<'e>(
144 executor: impl sqlx::PgExecutor<'e>,
145 id: UserId,
146 custom_html: &str,
147 custom_css: &str,
148 ) -> Result<DbUser> {
149 let user = sqlx::query_as::<_, DbUser>(
150 r"
151 UPDATE users
152 SET custom_html = $2,
153 custom_css = $3,
154 custom_pages_updated_at = now(),
155 cache_generation = cache_generation + 1
156 WHERE id = $1
157 RETURNING *
158 ",
159 )
160 .bind(id)
161 .bind(custom_html)
162 .bind(custom_css)
163 .fetch_one(executor)
164 .await?;
165 Ok(user)
166 }
167
168 /// Clear a user's custom profile page back to the platform default.
169 pub async fn reset_user_custom_page(pool: &PgPool, id: UserId) -> Result<()> {
170 sqlx::query(
171 "UPDATE users SET custom_html = '', custom_css = '', \
172 custom_pages_updated_at = NULL, cache_generation = cache_generation + 1 WHERE id = $1",
173 )
174 .bind(id)
175 .execute(pool)
176 .await?;
177 Ok(())
178 }
179
180 /// Set or clear a user's creator theme for their public profile. `None` clears
181 /// to the platform default. The id is validated against the embedded registry
182 /// before this call.
183 #[tracing::instrument(skip_all)]
184 pub async fn update_user_theme(pool: &PgPool, id: UserId, theme_id: Option<&str>) -> Result<()> {
185 sqlx::query("UPDATE users SET theme_id = $2, updated_at = NOW() WHERE id = $1")
186 .bind(id)
187 .bind(theme_id)
188 .execute(pool)
189 .await?;
190
191 Ok(())
192 }
193
194 /// Replace a user's password hash and invalidate outstanding JWTs.
195 #[tracing::instrument(skip_all)]
196 pub async fn update_user_password(pool: &PgPool, id: UserId, password_hash: &str) -> Result<()> {
197 sqlx::query("UPDATE users SET password_hash = $2, jwt_invalidated_at = NOW() WHERE id = $1")
198 .bind(id)
199 .bind(password_hash)
200 .execute(pool)
201 .await?;
202
203 Ok(())
204 }
205
206 /// Increment the user's feed key version, revoking their current personal-feed
207 /// URL. Returns the new version (folded into the next URL's HMAC).
208 #[tracing::instrument(skip_all)]
209 pub async fn bump_feed_key_version(pool: &PgPool, id: UserId) -> Result<i32> {
210 let (version,): (i32,) = sqlx::query_as(
211 "UPDATE users SET feed_key_version = feed_key_version + 1, updated_at = NOW() \
212 WHERE id = $1 RETURNING feed_key_version",
213 )
214 .bind(id)
215 .fetch_one(pool)
216 .await?;
217
218 Ok(version)
219 }
220
221 /// Self-deactivate an account (enter limbo state).
222 ///
223 /// Bumps `jwt_invalidated_at` so any outstanding SyncKit JWTs minted from
224 /// this account stop authenticating immediately.
225 #[tracing::instrument(skip_all)]
226 pub async fn deactivate_user(pool: &PgPool, id: UserId) -> Result<()> {
227 sqlx::query(
228 "UPDATE users SET deactivated_at = NOW(), jwt_invalidated_at = NOW(), updated_at = NOW() WHERE id = $1",
229 )
230 .bind(id)
231 .execute(pool)
232 .await?;
233
234 Ok(())
235 }
236
237 /// Reactivate a self-deactivated account.
238 #[tracing::instrument(skip_all)]
239 pub async fn reactivate_user(pool: &PgPool, id: UserId) -> Result<()> {
240 sqlx::query("UPDATE users SET deactivated_at = NULL, updated_at = NOW() WHERE id = $1")
241 .bind(id)
242 .execute(pool)
243 .await?;
244
245 Ok(())
246 }
247
248 /// Admin: permanently terminate an account (enforcement ladder step 4).
249 /// The user has 30 days to export data. After that, the scheduler deletes the account.
250 /// The account must already be suspended.
251 #[tracing::instrument(skip_all)]
252 pub async fn terminate_user(pool: &PgPool, id: UserId) -> Result<()> {
253 sqlx::query(
254 "UPDATE users SET terminated_at = NOW(), jwt_invalidated_at = NOW(), updated_at = NOW() WHERE id = $1",
255 )
256 .bind(id)
257 .execute(pool)
258 .await?;
259
260 Ok(())
261 }
262
263 /// Get user IDs of terminated accounts whose 30-day export window has expired.
264 #[tracing::instrument(skip_all)]
265 pub async fn get_expired_terminated_ids(pool: &PgPool) -> Result<Vec<UserId>> {
266 let ids: Vec<UserId> = sqlx::query_scalar(
267 r"
268 SELECT id FROM users
269 WHERE terminated_at IS NOT NULL
270 AND terminated_at < NOW() - INTERVAL '30 days'
271 ORDER BY terminated_at
272 LIMIT 1000
273 ",
274 )
275 .fetch_all(pool)
276 .await?;
277
278 Ok(ids)
279 }
280
281 /// Permanently delete a user by ID.
282 ///
283 /// `pub(crate)` and not for direct handler use: go through
284 /// [`crate::AppState::delete_user_account`], which also purges the in-memory
285 /// caches keyed to the user (domain_cache). Deleting here alone would leave a
286 /// stale, never-revalidated cache entry (ultra-fuzz Run 12 doubledown S-E).
287 #[tracing::instrument(skip_all)]
288 pub(crate) async fn delete_user(pool: &PgPool, id: UserId) -> Result<()> {
289 sqlx::query("DELETE FROM users WHERE id = $1")
290 .bind(id)
291 .execute(pool)
292 .await?;
293
294 Ok(())
295 }
296
297 /// Check whether this creator has any completed sales (transactions where they were the seller).
298 #[tracing::instrument(skip_all)]
299 pub async fn has_completed_sales(pool: &PgPool, id: UserId) -> Result<bool> {
300 let count: i64 = sqlx::query_scalar(
301 "SELECT COUNT(*) FROM transactions WHERE seller_id = $1 AND status = 'completed'",
302 )
303 .bind(id)
304 .fetch_one(pool)
305 .await?;
306
307 Ok(count > 0)
308 }
309
310 /// Schedule content removal 90 days from now. The user row is hidden from public
311 /// views but items remain accessible to buyers who previously purchased them.
312 /// After 90 days the scheduler deletes S3 objects and the user row.
313 #[tracing::instrument(skip_all)]
314 pub async fn schedule_content_removal(pool: &PgPool, id: UserId) -> Result<()> {
315 sqlx::query(
316 r"
317 UPDATE users
318 SET content_removal_at = NOW() + INTERVAL '90 days',
319 deactivated_at = NOW(),
320 updated_at = NOW()
321 WHERE id = $1
322 ",
323 )
324 .bind(id)
325 .execute(pool)
326 .await?;
327
328 Ok(())
329 }
330
331 /// Get user IDs whose 90-day content removal grace period has expired.
332 #[tracing::instrument(skip_all)]
333 pub async fn get_expired_content_removal_ids(pool: &PgPool) -> Result<Vec<UserId>> {
334 let ids: Vec<UserId> = sqlx::query_scalar(
335 r"
336 SELECT id FROM users
337 WHERE content_removal_at IS NOT NULL
338 AND content_removal_at < NOW()
339 ORDER BY content_removal_at
340 LIMIT 1000
341 ",
342 )
343 .fetch_all(pool)
344 .await?;
345
346 Ok(ids)
347 }
348
349 /// Create an ephemeral sandbox user. Returns the created row.
350 ///
351 /// The user gets `can_create_projects = true`, `email_verified = true`,
352 /// a SmallFiles creator tier, and a tight storage cap. The row is
353 /// automatically cleaned up by the scheduler after `sandbox_expires_at`.
354 #[tracing::instrument(skip_all)]
355 pub async fn create_sandbox_user(
356 pool: &PgPool,
357 username: &Username,
358 email: &Email,
359 password_hash: &str,
360 expiry_secs: i64,
361 ) -> Result<DbUser> {
362 let user = sqlx::query_as::<_, DbUser>(
363 r"
364 INSERT INTO users (
365 username, email, password_hash,
366 is_sandbox, sandbox_expires_at,
367 can_create_projects, email_verified,
368 creator_tier
369 )
370 VALUES (
371 $1, $2, $3,
372 TRUE, NOW() + make_interval(secs => $4::float8),
373 TRUE, TRUE,
374 'small_files'
375 )
376 RETURNING *
377 ",
378 )
379 .bind(username)
380 .bind(email)
381 .bind(password_hash)
382 .bind(expiry_secs as f64)
383 .fetch_one(pool)
384 .await?;
385
386 Ok(user)
387 }
388
389 /// Return IDs of sandbox users whose expiry has passed.
390 #[tracing::instrument(skip_all)]
391 pub async fn get_expired_sandbox_ids(pool: &PgPool) -> Result<Vec<UserId>> {
392 // Per-tick LIMIT bounds the supervisor's input list (the scheduler re-ticks
393 // and the WHERE re-excludes already-deleted rows, so the remainder is picked
394 // up next tick). Run #12 INFO, keeps a pathological mass-expiry from loading
395 // an unbounded id vec even though concurrency is already capped at 4.
396 let ids = sqlx::query_scalar::<_, UserId>(
397 "SELECT id FROM users WHERE is_sandbox = TRUE AND sandbox_expires_at < NOW() \
398 ORDER BY sandbox_expires_at LIMIT 1000",
399 )
400 .fetch_all(pool)
401 .await?;
402
403 Ok(ids)
404 }
405
406 /// Count active (non-expired) sandbox accounts created from a given IP.
407 /// Used to enforce the per-IP concurrent sandbox cap.
408 #[tracing::instrument(skip_all)]
409 pub async fn count_active_sandboxes_by_ip(pool: &PgPool, ip: &str) -> Result<i64> {
410 let count: i64 = sqlx::query_scalar(
411 r"
412 SELECT COUNT(*) FROM users u
413 JOIN user_sessions us ON us.user_id = u.id
414 WHERE u.is_sandbox = TRUE
415 AND u.sandbox_expires_at > NOW()
416 AND us.ip_address = $1
417 ",
418 )
419 .bind(ip)
420 .fetch_one(pool)
421 .await?;
422
423 Ok(count)
424 }
425
426 /// Update user's Stripe Connect account information after OAuth
427 #[tracing::instrument(skip_all)]
428 pub async fn update_user_stripe_account(
429 pool: &PgPool,
430 user_id: UserId,
431 stripe_account_id: &str,
432 onboarding_complete: bool,
433 payouts_enabled: bool,
434 charges_enabled: bool,
435 ) -> Result<DbUser> {
436 let user = sqlx::query_as::<_, DbUser>(
437 r"
438 UPDATE users
439 SET stripe_account_id = $2,
440 stripe_onboarding_complete = $3,
441 stripe_payouts_enabled = $4,
442 stripe_charges_enabled = $5,
443 updated_at = NOW()
444 WHERE id = $1
445 RETURNING *
446 ",
447 )
448 .bind(user_id)
449 .bind(stripe_account_id)
450 .bind(onboarding_complete)
451 .bind(payouts_enabled)
452 .bind(charges_enabled)
453 .fetch_one(pool)
454 .await?;
455
456 Ok(user)
457 }
458
459 /// Atomically set a user's Stripe Connect account ID, but only if one is not
460 /// already set. Returns `Some(user)` on success, or `None` if another request
461 /// already claimed the slot (race-condition guard).
462 #[tracing::instrument(skip_all)]
463 pub async fn try_set_stripe_account(
464 pool: &PgPool,
465 user_id: UserId,
466 stripe_account_id: &str,
467 ) -> Result<Option<DbUser>> {
468 let user = sqlx::query_as::<_, DbUser>(
469 r"
470 UPDATE users
471 SET stripe_account_id = $2,
472 stripe_onboarding_complete = false,
473 stripe_payouts_enabled = false,
474 stripe_charges_enabled = false,
475 updated_at = NOW()
476 WHERE id = $1 AND (stripe_account_id IS NULL OR stripe_account_id = '')
477 RETURNING *
478 ",
479 )
480 .bind(user_id)
481 .bind(stripe_account_id)
482 .fetch_optional(pool)
483 .await?;
484
485 Ok(user)
486 }
487
488 /// Update user's Stripe status from webhook (finds user by stripe_account_id)
489 #[tracing::instrument(skip_all)]
490 pub async fn update_user_stripe_status(
491 pool: &PgPool,
492 stripe_account_id: &str,
493 onboarding_complete: bool,
494 payouts_enabled: bool,
495 charges_enabled: bool,
496 settlement_currency: Option<crate::currency::SettlementCurrency>,
497 ) -> Result<Option<DbUser>> {
498 // COALESCE, not a plain assignment: `None` means Stripe told us nothing
499 // usable this time (too early in onboarding, or a currency outside our six),
500 // and overwriting a known currency with USD on that signal would redenominate
501 // every price the creator has set.
502 let user = sqlx::query_as::<_, DbUser>(
503 r"
504 UPDATE users
505 SET stripe_onboarding_complete = $2,
506 stripe_payouts_enabled = $3,
507 stripe_charges_enabled = $4,
508 settlement_currency = COALESCE($5, settlement_currency),
509 updated_at = NOW()
510 WHERE stripe_account_id = $1
511 RETURNING *
512 ",
513 )
514 .bind(stripe_account_id)
515 .bind(onboarding_complete)
516 .bind(payouts_enabled)
517 .bind(charges_enabled)
518 .bind(settlement_currency)
519 .fetch_optional(pool)
520 .await?;
521
522 Ok(user)
523 }
524
525 /// The settlement currency currently stored for a connected account, if any.
526 ///
527 /// Read before a webhook write so a *change* can be distinguished from a
528 /// restatement of the same value. Stripe re-sends `account.updated` constantly,
529 /// so alerting on every write would be noise; alerting on none of them would
530 /// leave a creator's prices silently meaning different money.
531 #[tracing::instrument(skip_all)]
532 pub async fn get_settlement_currency_by_stripe_account(
533 pool: &PgPool,
534 stripe_account_id: &str,
535 ) -> Result<Option<crate::currency::SettlementCurrency>> {
536 let row: Option<(crate::currency::SettlementCurrency,)> =
537 sqlx::query_as("SELECT settlement_currency FROM users WHERE stripe_account_id = $1")
538 .bind(stripe_account_id)
539 .fetch_optional(pool)
540 .await?;
541 Ok(row.map(|(c,)| c))
542 }
543
544 /// The account behind a Stripe Connect account id.
545 #[tracing::instrument(skip_all)]
546 pub async fn get_user_id_by_stripe_account(
547 pool: &PgPool,
548 stripe_account_id: &str,
549 ) -> Result<Option<UserId>> {
550 let id = sqlx::query_scalar::<_, UserId>("SELECT id FROM users WHERE stripe_account_id = $1")
551 .bind(stripe_account_id)
552 .fetch_optional(pool)
553 .await?;
554 Ok(id)
555 }
556
557 /// Store a buyer's cross-currency conversion preference.
558 ///
559 /// A preference, not a lock: the checkout form decides each purchase, and this
560 /// only changes what that form comes back pre-selected with next time.
561 #[tracing::instrument(skip_all)]
562 pub async fn set_conversion_preference(
563 pool: &PgPool,
564 user_id: UserId,
565 conversion: crate::currency::ConversionChoice,
566 ) -> Result<()> {
567 sqlx::query("UPDATE users SET conversion_preference = $2, updated_at = NOW() WHERE id = $1")
568 .bind(user_id)
569 .bind(conversion)
570 .execute(pool)
571 .await?;
572 Ok(())
573 }
574
575 /// Mark a user's email as verified
576 #[tracing::instrument(skip_all)]
577 pub async fn verify_user_email(pool: &PgPool, user_id: UserId) -> Result<()> {
578 sqlx::query(
579 r"
580 UPDATE users
581 SET email_verified = true,
582 email_verification_token = NULL,
583 updated_at = NOW()
584 WHERE id = $1
585 ",
586 )
587 .bind(user_id)
588 .execute(pool)
589 .await?;
590
591 Ok(())
592 }
593
594 // ── Suspension / Appeals ──
595
596 /// Suspend a user account, clearing any prior appeal fields.
597 ///
598 /// Bumps `jwt_invalidated_at` so SyncKit JWTs minted for this user expire
599 /// at the next extractor check (subject to `SESSION_TOUCH_CACHE_SECS`).
600 #[tracing::instrument(skip_all)]
601 pub async fn suspend_user(pool: &PgPool, user_id: UserId, reason: &str) -> Result<()> {
602 sqlx::query(
603 r"
604 UPDATE users
605 SET suspended_at = NOW(),
606 suspension_reason = $2,
607 jwt_invalidated_at = NOW(),
608 appeal_text = NULL,
609 appeal_submitted_at = NULL,
610 appeal_decision = NULL,
611 appeal_response = NULL,
612 appeal_decided_at = NULL,
613 updated_at = NOW()
614 WHERE id = $1
615 ",
616 )
617 .bind(user_id)
618 .bind(reason)
619 .execute(pool)
620 .await?;
621
622 Ok(())
623 }
624
625 /// Remove suspension and clear all suspension/appeal fields.
626 #[tracing::instrument(skip_all)]
627 pub async fn unsuspend_user(pool: &PgPool, user_id: UserId) -> Result<()> {
628 sqlx::query(
629 r"
630 UPDATE users
631 SET suspended_at = NULL,
632 suspension_reason = NULL,
633 appeal_text = NULL,
634 appeal_submitted_at = NULL,
635 appeal_decision = NULL,
636 appeal_response = NULL,
637 appeal_decided_at = NULL,
638 updated_at = NOW()
639 WHERE id = $1
640 ",
641 )
642 .bind(user_id)
643 .execute(pool)
644 .await?;
645
646 Ok(())
647 }
648
649 // ── Creator pause (voluntary) ──
650
651 /// Set the creator_paused_at timestamp (voluntary pause).
652 #[tracing::instrument(skip_all)]
653 pub async fn pause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
654 sqlx::query("UPDATE users SET creator_paused_at = NOW(), updated_at = NOW() WHERE id = $1")
655 .bind(user_id)
656 .execute(pool)
657 .await?;
658
659 Ok(())
660 }
661
662 /// Clear the creator_paused_at timestamp (resume from voluntary pause).
663 #[tracing::instrument(skip_all)]
664 pub async fn unpause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
665 sqlx::query("UPDATE users SET creator_paused_at = NULL, updated_at = NOW() WHERE id = $1")
666 .bind(user_id)
667 .execute(pool)
668 .await?;
669
670 Ok(())
671 }
672
673 /// Submit an appeal for a suspended account, clearing any prior decision.
674 #[tracing::instrument(skip_all)]
675 pub async fn submit_appeal(pool: &PgPool, user_id: UserId, appeal_text: &str) -> Result<()> {
676 sqlx::query(
677 r"
678 UPDATE users
679 SET appeal_text = $2,
680 appeal_submitted_at = NOW(),
681 appeal_decision = NULL,
682 appeal_response = NULL,
683 appeal_decided_at = NULL,
684 updated_at = NOW()
685 WHERE id = $1 AND suspended_at IS NOT NULL
686 ",
687 )
688 .bind(user_id)
689 .bind(appeal_text)
690 .execute(pool)
691 .await?;
692
693 Ok(())
694 }
695
696 /// Resolve an appeal. If approved, also clears suspension.
697 #[tracing::instrument(skip_all)]
698 pub async fn resolve_appeal(
699 pool: &PgPool,
700 user_id: UserId,
701 decision: AppealDecision,
702 response: &str,
703 ) -> Result<()> {
704 if decision == AppealDecision::Approved {
705 // Approve: clear suspension entirely
706 sqlx::query(
707 r"
708 UPDATE users
709 SET appeal_decision = $2,
710 appeal_response = $3,
711 appeal_decided_at = NOW(),
712 suspended_at = NULL,
713 suspension_reason = NULL,
714 updated_at = NOW()
715 WHERE id = $1
716 ",
717 )
718 .bind(user_id)
719 .bind(decision)
720 .bind(response)
721 .execute(pool)
722 .await?;
723 } else {
724 // Deny: keep suspension, record decision
725 sqlx::query(
726 r"
727 UPDATE users
728 SET appeal_decision = $2,
729 appeal_response = $3,
730 appeal_decided_at = NOW(),
731 updated_at = NOW()
732 WHERE id = $1
733 ",
734 )
735 .bind(user_id)
736 .bind(decision)
737 .bind(response)
738 .execute(pool)
739 .await?;
740 }
741
742 Ok(())
743 }
744
745 /// Admin query: users with a pending appeal (submitted but not yet decided).
746 #[tracing::instrument(skip_all)]
747 pub async fn get_pending_appeals(pool: &PgPool) -> Result<Vec<DbUser>> {
748 let users = sqlx::query_as::<_, DbUser>(
749 r"
750 SELECT * FROM users
751 WHERE appeal_submitted_at IS NOT NULL
752 AND appeal_decided_at IS NULL
753 ORDER BY appeal_submitted_at ASC
754 LIMIT 500
755 ",
756 )
757 .fetch_all(pool)
758 .await?;
759
760 Ok(users)
761 }
762
763 /// Admin query: all users, optionally filtered by suspension status, with pagination.
764 #[tracing::instrument(skip_all)]
765 pub async fn get_all_users(
766 pool: &PgPool,
767 filter: Option<&str>,
768 limit: i64,
769 offset: i64,
770 ) -> Result<Vec<DbUser>> {
771 let limit = limit.min(200);
772 let users = match filter {
773 Some("suspended") => {
774 sqlx::query_as::<_, DbUser>(
775 "SELECT * FROM users WHERE suspended_at IS NOT NULL ORDER BY suspended_at DESC LIMIT $1 OFFSET $2",
776 )
777 .bind(limit)
778 .bind(offset)
779 .fetch_all(pool)
780 .await?
781 }
782 Some("active") => {
783 sqlx::query_as::<_, DbUser>(
784 "SELECT * FROM users WHERE suspended_at IS NULL ORDER BY created_at DESC LIMIT $1 OFFSET $2",
785 )
786 .bind(limit)
787 .bind(offset)
788 .fetch_all(pool)
789 .await?
790 }
791 // Creators with a custom page; most recently changed first, so
792 // "recently changed" surfaces naturally at the top.
793 Some("custom_pages") => {
794 sqlx::query_as::<_, DbUser>(
795 "SELECT * FROM users WHERE custom_html <> '' OR custom_css <> '' \
796 ORDER BY custom_pages_updated_at DESC NULLS LAST, created_at DESC LIMIT $1 OFFSET $2",
797 )
798 .bind(limit)
799 .bind(offset)
800 .fetch_all(pool)
801 .await?
802 }
803 Some("pages_locked") => {
804 sqlx::query_as::<_, DbUser>(
805 "SELECT * FROM users WHERE custom_pages_locked = true ORDER BY created_at DESC LIMIT $1 OFFSET $2",
806 )
807 .bind(limit)
808 .bind(offset)
809 .fetch_all(pool)
810 .await?
811 }
812 _ => {
813 sqlx::query_as::<_, DbUser>(
814 "SELECT * FROM users ORDER BY created_at DESC LIMIT $1 OFFSET $2",
815 )
816 .bind(limit)
817 .bind(offset)
818 .fetch_all(pool)
819 .await?
820 }
821 };
822
823 Ok(users)
824 }
825
826 /// Count users matching a filter (for pagination totals).
827 #[tracing::instrument(skip_all)]
828 pub async fn count_users(pool: &PgPool, filter: Option<&str>) -> Result<i64> {
829 let count = match filter {
830 Some("suspended") => {
831 sqlx::query_scalar::<_, i64>(
832 "SELECT COUNT(*) FROM users WHERE suspended_at IS NOT NULL",
833 )
834 .fetch_one(pool)
835 .await?
836 }
837 Some("active") => {
838 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users WHERE suspended_at IS NULL")
839 .fetch_one(pool)
840 .await?
841 }
842 Some("custom_pages") => {
843 sqlx::query_scalar::<_, i64>(
844 "SELECT COUNT(*) FROM users WHERE custom_html <> '' OR custom_css <> ''",
845 )
846 .fetch_one(pool)
847 .await?
848 }
849 Some("pages_locked") => {
850 sqlx::query_scalar::<_, i64>(
851 "SELECT COUNT(*) FROM users WHERE custom_pages_locked = true",
852 )
853 .fetch_one(pool)
854 .await?
855 }
856 _ => {
857 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users")
858 .fetch_one(pool)
859 .await?
860 }
861 };
862
863 Ok(count)
864 }
865
866 /// Count total and suspended users in a single query.
867 #[tracing::instrument(skip_all)]
868 pub async fn count_users_summary(pool: &PgPool) -> Result<(i64, i64)> {
869 let (total, suspended): (i64, i64) = sqlx::query_as(
870 r"
871 SELECT
872 COUNT(*),
873 COUNT(*) FILTER (WHERE suspended_at IS NOT NULL)
874 FROM users
875 ",
876 )
877 .fetch_one(pool)
878 .await?;
879
880 Ok((total, suspended))
881 }
882
883 /// Get all user emails for bulk notifications (e.g. shutdown notice).
884 ///
885 /// Capped to bound memory on a full-table scan; if the cap is ever hit the WARN
886 /// fires so we know to switch to a paged dispatch model (mirrors
887 /// [`get_status_alert_subscribers`]).
888 #[tracing::instrument(skip_all)]
889 pub async fn get_all_user_emails(pool: &PgPool) -> Result<Vec<(String, Option<String>)>> {
890 const ALL_EMAILS_CAP: i64 = 50_000;
891 let rows = sqlx::query_as::<_, (String, Option<String>)>(
892 "SELECT email, display_name FROM users ORDER BY created_at ASC LIMIT $1",
893 )
894 .bind(ALL_EMAILS_CAP)
895 .fetch_all(pool)
896 .await?;
897
898 if rows.len() as i64 == ALL_EMAILS_CAP {
899 tracing::warn!(
900 cap = ALL_EMAILS_CAP,
901 "get_all_user_emails hit its cap; some users omitted, switch to paged dispatch"
902 );
903 }
904
905 Ok(rows)
906 }
907
908 /// Update a user's email notification preferences.
909 ///
910 /// Writes subscriptions only. The `users.notify_*` columns these used to set
911 /// are gone (migration 189); `db::lists` is the single record, and the
912 /// preferences page and the unsubscribe links write the same rows.
913 #[derive(Debug, Clone, Copy)]
914 pub struct NotificationPreferences {
915 pub notify_sale: bool,
916 pub notify_follower: bool,
917 pub notify_release: bool,
918 pub login_notification_enabled: bool,
919 pub notify_issues: bool,
920 pub notify_status: bool,
921 pub notify_invite: bool,
922 }
923
924 #[tracing::instrument(skip_all)]
925 pub async fn update_notification_preferences(
926 pool: &PgPool,
927 id: UserId,
928 prefs: NotificationPreferences,
929 ) -> Result<()> {
930 let NotificationPreferences {
931 notify_sale,
932 notify_follower,
933 notify_release,
934 login_notification_enabled,
935 notify_issues,
936 notify_status,
937 notify_invite,
938 } = prefs;
939
940 for (kind, enabled) in [
941 ("sale", notify_sale),
942 ("follower", notify_follower),
943 ("releases", notify_release),
944 ("issues", notify_issues),
945 ("status", notify_status),
946 ("login", login_notification_enabled),
947 ("invite", notify_invite),
948 ] {
949 crate::db::lists::sync_notification_subscription(pool, id, kind, enabled).await?;
950 }
951
952 Ok(())
953 }
954
955 /// Update tip settings.
956 ///
957 /// `tips_enabled` is a capability (whether the creator accepts tips at all) and
958 /// stays a column. `notify_tip` is a notification preference and now lives in
959 /// subscriptions with the other six, so the two are written to different
960 /// places despite arriving from the same form.
961 #[tracing::instrument(skip_all)]
962 pub async fn update_tip_preferences(
963 pool: &PgPool,
964 id: UserId,
965 tips_enabled: bool,
966 notify_tip: bool,
967 ) -> Result<()> {
968 sqlx::query("UPDATE users SET tips_enabled = $2, updated_at = NOW() WHERE id = $1")
969 .bind(id)
970 .bind(tips_enabled)
971 .execute(pool)
972 .await?;
973
974 crate::db::lists::sync_notification_subscription(pool, id, "tip", notify_tip).await?;
975 Ok(())
976 }
977
978 /// Turn one notification off, by the name the unsubscribe link carries.
979 ///
980 /// For the seven original preferences that name is the old `users.notify_*`
981 /// column, because those names are baked into signed URLs already sitting in
982 /// inboxes; they map back to list kinds here rather than being renamed, which
983 /// would invalidate every link ever sent.
984 ///
985 /// A kind with no legacy column (`invite`, migration 195) carries its kind name
986 /// instead. Nothing older is in an inbox to be broken, so there is no column
987 /// name to preserve and inventing one would be cargo cult.
988 #[tracing::instrument(skip_all)]
989 pub async fn disable_notification(
990 pool: &PgPool,
991 user_id: UserId,
992 preference: &str,
993 ) -> Result<bool> {
994 let legacy: Option<&str> = crate::db::lists::NOTIFICATION_LISTS
995 .iter()
996 .find(|(_, legacy)| *legacy == preference)
997 .map(|(kind, _)| *kind);
998 let Some(kind) = legacy.or_else(|| {
999 preference
1000 .parse::<crate::db::ListKind>()
1001 .ok()
1002 .map(|_| preference)
1003 }) else {
1004 return Ok(false);
1005 };
1006 crate::db::lists::sync_notification_subscription(pool, user_id, kind, false).await?;
1007 Ok(true)
1008 }
1009
1010 /// A user who opted into platform status notifications.
1011 #[derive(sqlx::FromRow)]
1012 pub struct StatusAlertSubscriber {
1013 pub id: UserId,
1014 pub email: Email,
1015 pub display_name: Option<String>,
1016 }
1017
1018 /// Get all users who opted into platform status notifications.
1019 ///
1020 /// Hard cap at 10k rows so the monitor's status-change fan-out can't unbox
1021 /// an unbounded query into RAM. The 100ms pacing in `monitor.rs` already
1022 /// limits fan-out throughput to ~600/minute, anything past 10k would
1023 /// chew through Postmark rate limits anyway. If we ever hit the cap a
1024 /// WARN fires so we know to switch to a paged dispatch model.
1025 #[tracing::instrument(skip_all)]
1026 pub async fn get_status_alert_subscribers(pool: &PgPool) -> Result<Vec<StatusAlertSubscriber>> {
1027 const STATUS_SUBSCRIBER_CAP: i64 = 10_000;
1028 let rows = sqlx::query_as::<_, StatusAlertSubscriber>(
1029 "SELECT u.id, u.email, u.display_name FROM users u \
1030 JOIN list_subscriptions ls ON ls.user_id = u.id \
1031 JOIN lists l ON l.id = ls.list_id AND l.scope = 'platform' AND l.kind = 'status' \
1032 WHERE ls.state IN ('confirmed', 'imported') AND u.deactivated_at IS NULL \
1033 ORDER BY u.id LIMIT $1",
1034 )
1035 .bind(STATUS_SUBSCRIBER_CAP)
1036 .fetch_all(pool)
1037 .await?;
1038 if rows.len() as i64 == STATUS_SUBSCRIBER_CAP {
1039 tracing::warn!(
1040 cap = STATUS_SUBSCRIBER_CAP,
1041 "get_status_alert_subscribers hit hard cap; promote to paged dispatch"
1042 );
1043 }
1044 Ok(rows)
1045 }
1046
1047 /// Atomically check-and-set broadcast timestamp. Returns false if already sent within 24 hours.
1048 #[tracing::instrument(skip_all)]
1049 pub async fn try_set_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<bool> {
1050 let result = sqlx::query(
1051 r"
1052 UPDATE users
1053 SET last_broadcast_at = NOW()
1054 WHERE id = $1
1055 AND (last_broadcast_at IS NULL OR last_broadcast_at < NOW() - INTERVAL '24 hours')
1056 ",
1057 )
1058 .bind(user_id)
1059 .execute(pool)
1060 .await?;
1061
1062 Ok(result.rows_affected() > 0)
1063 }
1064
1065 /// Release the 24h broadcast slot. Used when a broadcast is refused after the
1066 /// slot has already been claimed (e.g. recipient cap exceeded) so the creator
1067 /// can retry without waiting a day.
1068 #[tracing::instrument(skip_all)]
1069 pub async fn clear_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<()> {
1070 sqlx::query("UPDATE users SET last_broadcast_at = NULL WHERE id = $1")
1071 .bind(user_id)
1072 .execute(pool)
1073 .await?;
1074 Ok(())
1075 }
1076
1077 // ── Upload Trust ──
1078
1079 /// Check if a user is trusted for uploads (bypasses review queue).
1080 #[tracing::instrument(skip_all)]
1081 pub async fn is_upload_trusted(pool: &PgPool, user_id: UserId) -> Result<bool> {
1082 let trusted = sqlx::query_scalar::<_, bool>("SELECT upload_trusted FROM users WHERE id = $1")
1083 .bind(user_id)
1084 .fetch_one(pool)
1085 .await?;
1086
1087 Ok(trusted)
1088 }
1089
1090 /// Set a user's upload trust status.
1091 #[tracing::instrument(skip_all)]
1092 pub async fn set_upload_trusted(pool: &PgPool, user_id: UserId, trusted: bool) -> Result<()> {
1093 sqlx::query(
1094 r"
1095 UPDATE users
1096 SET upload_trusted = $2,
1097 updated_at = NOW()
1098 WHERE id = $1
1099 ",
1100 )
1101 .bind(user_id)
1102 .bind(trusted)
1103 .execute(pool)
1104 .await?;
1105
1106 Ok(())
1107 }
1108
1109 /// Moderation kill switch for custom pages. While locked, the creator can't
1110 /// edit their custom pages and the live pages render the platform default.
1111 /// Reversible: unlocking restores the (preserved) custom source.
1112 pub async fn set_custom_pages_locked(pool: &PgPool, user_id: UserId, locked: bool) -> Result<()> {
1113 sqlx::query("UPDATE users SET custom_pages_locked = $2, cache_generation = cache_generation + 1 WHERE id = $1")
1114 .bind(user_id)
1115 .bind(locked)
1116 .execute(pool)
1117 .await?;
1118 Ok(())
1119 }
1120
1121 // ── Onboarding email drip ──
1122
1123 /// Users who need the next onboarding email. Returns users at a given step
1124 /// whose last email was sent more than `min_age` ago (or never).
1125 #[tracing::instrument(skip_all)]
1126 pub async fn get_onboarding_candidates(
1127 pool: &PgPool,
1128 step: i16,
1129 min_age: chrono::Duration,
1130 ) -> Result<Vec<DbUser>> {
1131 let cutoff = chrono::Utc::now() - min_age;
1132 // Per-tick LIMIT bounds the scheduler's input list. The caller advances
1133 // each returned user's step (so the WHERE re-excludes them), meaning the
1134 // remainder is drained on the next tick, same re-tick pattern as the
1135 // sandbox/terminated/content-removal cleanup queries. Run #14 MEDIUM: a
1136 // signup surge must not load an unbounded user vec into the lock-held tick.
1137 let users = sqlx::query_as::<_, DbUser>(
1138 "SELECT * FROM users
1139 WHERE onboarding_email_step = $1
1140 AND (onboarding_email_sent_at IS NULL OR onboarding_email_sent_at < $2)
1141 AND suspended_at IS NULL
1142 ORDER BY onboarding_email_sent_at ASC NULLS FIRST
1143 LIMIT 1000",
1144 )
1145 .bind(step)
1146 .bind(cutoff)
1147 .fetch_all(pool)
1148 .await?;
1149 Ok(users)
1150 }
1151
1152 /// Advance a user's onboarding email step and record the send time.
1153 #[tracing::instrument(skip_all)]
1154 pub async fn advance_onboarding_step(pool: &PgPool, user_id: UserId, new_step: i16) -> Result<()> {
1155 sqlx::query(
1156 "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = $1",
1157 )
1158 .bind(user_id)
1159 .bind(new_step)
1160 .execute(pool)
1161 .await?;
1162 Ok(())
1163 }
1164
1165 /// Advance onboarding step for multiple users in a single query.
1166 #[tracing::instrument(skip_all)]
1167 pub async fn batch_advance_onboarding_step(
1168 pool: &PgPool,
1169 user_ids: &[UserId],
1170 new_step: i16,
1171 ) -> Result<()> {
1172 sqlx::query(
1173 "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = ANY($1)",
1174 )
1175 .bind(user_ids)
1176 .bind(new_step)
1177 .execute(pool)
1178 .await?;
1179 Ok(())
1180 }
1181
1182 /// Mark a user as a founder. Called when they start a creator-tier
1183 /// subscription while the founder pricing window is open. Sticky; never
1184 /// reset, even on cancellation. Subsequent re-subscriptions during the
1185 /// window keep their founder status. After the window closes, eligibility
1186 /// is determined by `founder_locked_at` (stamped only for users with an
1187 /// active subscription at the close-time snapshot).
1188 ///
1189 /// **DIY exclusion**: DIY-tier accounts are not full members and must not
1190 /// qualify for founder pricing. This function does not enforce that, it sets
1191 /// `is_founder` unconditionally, so the exclusion is a caller obligation: only
1192 /// call this from creator-tier (Basic/SmallFiles/BigFiles/Everything) checkout
1193 /// paths. When DIY ships, its checkout path must NOT invoke this.
1194 #[tracing::instrument(skip_all)]
1195 pub async fn mark_user_as_founder(pool: &PgPool, user_id: UserId) -> Result<()> {
1196 sqlx::query(
1197 r"
1198 UPDATE users
1199 SET is_founder = TRUE,
1200 updated_at = NOW()
1201 WHERE id = $1 AND is_founder = FALSE
1202 ",
1203 )
1204 .bind(user_id)
1205 .execute(pool)
1206 .await?;
1207 Ok(())
1208 }
1209
1210 /// Close the founder pricing window by stamping `founder_locked_at` on every
1211 /// user who is currently flagged `is_founder` AND has an active creator-tier
1212 /// subscription. Returns the number of users locked in. Idempotent: skips
1213 /// any user already locked. Intended to be called once from an admin tool
1214 /// at the moment the founder window closes.
1215 #[tracing::instrument(skip_all)]
1216 pub async fn lock_in_founders_with_active_subscriptions(pool: &PgPool) -> Result<u64> {
1217 let result = sqlx::query(
1218 r"
1219 UPDATE users u
1220 SET founder_locked_at = NOW(),
1221 updated_at = NOW()
1222 WHERE u.is_founder = TRUE
1223 AND u.founder_locked_at IS NULL
1224 AND EXISTS (
1225 SELECT 1 FROM creator_subscriptions s
1226 WHERE s.user_id = u.id
1227 AND s.status = 'active'
1228 )
1229 ",
1230 )
1231 .execute(pool)
1232 .await?;
1233 Ok(result.rows_affected())
1234 }
1235
1236 /// Update a user's Stripe Tax toggle.
1237 #[tracing::instrument(skip_all)]
1238 pub async fn update_stripe_tax_enabled(
1239 pool: &PgPool,
1240 user_id: UserId,
1241 enabled: bool,
1242 ) -> Result<()> {
1243 sqlx::query(
1244 r"
1245 UPDATE users
1246 SET stripe_tax_enabled = $2,
1247 updated_at = NOW()
1248 WHERE id = $1
1249 ",
1250 )
1251 .bind(user_id)
1252 .bind(enabled)
1253 .execute(pool)
1254 .await?;
1255
1256 Ok(())
1257 }
1258
1259 /// Disconnect a user's Stripe account
1260 #[tracing::instrument(skip_all)]
1261 pub async fn disconnect_user_stripe(pool: &PgPool, user_id: UserId) -> Result<DbUser> {
1262 let user = sqlx::query_as::<_, DbUser>(
1263 r"
1264 UPDATE users
1265 SET stripe_account_id = NULL,
1266 stripe_onboarding_complete = false,
1267 stripe_payouts_enabled = false,
1268 stripe_charges_enabled = false,
1269 updated_at = NOW()
1270 WHERE id = $1
1271 RETURNING *
1272 ",
1273 )
1274 .bind(user_id)
1275 .fetch_one(pool)
1276 .await?;
1277
1278 Ok(user)
1279 }
1280
1281 /// Fetch the current cache generation for a user (cheap, indexed lookup).
1282 #[tracing::instrument(skip_all)]
1283 pub async fn get_cache_generation(pool: &PgPool, user_id: UserId) -> Result<i64> {
1284 let generation =
1285 sqlx::query_scalar::<_, i64>("SELECT cache_generation FROM users WHERE id = $1")
1286 .bind(user_id)
1287 .fetch_one(pool)
1288 .await?;
1289
1290 Ok(generation)
1291 }
1292
1293 /// Atomically increment the user's cache generation counter.
1294 /// Call after any write that changes user-visible dashboard data.
1295 #[tracing::instrument(skip_all)]
1296 pub async fn bump_cache_generation(pool: &PgPool, user_id: UserId) -> Result<()> {
1297 sqlx::query("UPDATE users SET cache_generation = cache_generation + 1 WHERE id = $1")
1298 .bind(user_id)
1299 .execute(pool)
1300 .await?;
1301
1302 Ok(())
1303 }
1304
1305 /// Look up a verified user by email (case-insensitive).
1306 /// Returns the user ID if a verified account exists with that email.
1307 #[tracing::instrument(skip_all)]
1308 pub async fn get_verified_user_id_by_email(pool: &PgPool, email: &Email) -> Result<Option<UserId>> {
1309 let id = sqlx::query_scalar(
1310 "SELECT id FROM users WHERE LOWER(email) = LOWER($1) AND email_verified = true",
1311 )
1312 .bind(email)
1313 .fetch_optional(pool)
1314 .await?;
1315
1316 Ok(id)
1317 }
1318