Skip to main content

max / makenotwork

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