Skip to main content

max / makenotwork

37.2 KB · 1268 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 ) -> Result<Option<DbUser>> {
497 let user = sqlx::query_as::<_, DbUser>(
498 r"
499 UPDATE users
500 SET stripe_onboarding_complete = $2,
501 stripe_payouts_enabled = $3,
502 stripe_charges_enabled = $4,
503 updated_at = NOW()
504 WHERE stripe_account_id = $1
505 RETURNING *
506 ",
507 )
508 .bind(stripe_account_id)
509 .bind(onboarding_complete)
510 .bind(payouts_enabled)
511 .bind(charges_enabled)
512 .fetch_optional(pool)
513 .await?;
514
515 Ok(user)
516 }
517
518 /// Mark a user's email as verified
519 #[tracing::instrument(skip_all)]
520 pub async fn verify_user_email(pool: &PgPool, user_id: UserId) -> Result<()> {
521 sqlx::query(
522 r"
523 UPDATE users
524 SET email_verified = true,
525 email_verification_token = NULL,
526 updated_at = NOW()
527 WHERE id = $1
528 ",
529 )
530 .bind(user_id)
531 .execute(pool)
532 .await?;
533
534 Ok(())
535 }
536
537 // ── Suspension / Appeals ──
538
539 /// Suspend a user account, clearing any prior appeal fields.
540 ///
541 /// Bumps `jwt_invalidated_at` so SyncKit JWTs minted for this user expire
542 /// at the next extractor check (subject to `SESSION_TOUCH_CACHE_SECS`).
543 #[tracing::instrument(skip_all)]
544 pub async fn suspend_user(pool: &PgPool, user_id: UserId, reason: &str) -> Result<()> {
545 sqlx::query(
546 r"
547 UPDATE users
548 SET suspended_at = NOW(),
549 suspension_reason = $2,
550 jwt_invalidated_at = NOW(),
551 appeal_text = NULL,
552 appeal_submitted_at = NULL,
553 appeal_decision = NULL,
554 appeal_response = NULL,
555 appeal_decided_at = NULL,
556 updated_at = NOW()
557 WHERE id = $1
558 ",
559 )
560 .bind(user_id)
561 .bind(reason)
562 .execute(pool)
563 .await?;
564
565 Ok(())
566 }
567
568 /// Remove suspension and clear all suspension/appeal fields.
569 #[tracing::instrument(skip_all)]
570 pub async fn unsuspend_user(pool: &PgPool, user_id: UserId) -> Result<()> {
571 sqlx::query(
572 r"
573 UPDATE users
574 SET suspended_at = NULL,
575 suspension_reason = NULL,
576 appeal_text = NULL,
577 appeal_submitted_at = NULL,
578 appeal_decision = NULL,
579 appeal_response = NULL,
580 appeal_decided_at = NULL,
581 updated_at = NOW()
582 WHERE id = $1
583 ",
584 )
585 .bind(user_id)
586 .execute(pool)
587 .await?;
588
589 Ok(())
590 }
591
592 // ── Creator pause (voluntary) ──
593
594 /// Set the creator_paused_at timestamp (voluntary pause).
595 #[tracing::instrument(skip_all)]
596 pub async fn pause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
597 sqlx::query("UPDATE users SET creator_paused_at = NOW(), updated_at = NOW() WHERE id = $1")
598 .bind(user_id)
599 .execute(pool)
600 .await?;
601
602 Ok(())
603 }
604
605 /// Clear the creator_paused_at timestamp (resume from voluntary pause).
606 #[tracing::instrument(skip_all)]
607 pub async fn unpause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
608 sqlx::query("UPDATE users SET creator_paused_at = NULL, updated_at = NOW() WHERE id = $1")
609 .bind(user_id)
610 .execute(pool)
611 .await?;
612
613 Ok(())
614 }
615
616 /// Submit an appeal for a suspended account, clearing any prior decision.
617 #[tracing::instrument(skip_all)]
618 pub async fn submit_appeal(pool: &PgPool, user_id: UserId, appeal_text: &str) -> Result<()> {
619 sqlx::query(
620 r"
621 UPDATE users
622 SET appeal_text = $2,
623 appeal_submitted_at = NOW(),
624 appeal_decision = NULL,
625 appeal_response = NULL,
626 appeal_decided_at = NULL,
627 updated_at = NOW()
628 WHERE id = $1 AND suspended_at IS NOT NULL
629 ",
630 )
631 .bind(user_id)
632 .bind(appeal_text)
633 .execute(pool)
634 .await?;
635
636 Ok(())
637 }
638
639 /// Resolve an appeal. If approved, also clears suspension.
640 #[tracing::instrument(skip_all)]
641 pub async fn resolve_appeal(
642 pool: &PgPool,
643 user_id: UserId,
644 decision: AppealDecision,
645 response: &str,
646 ) -> Result<()> {
647 if decision == AppealDecision::Approved {
648 // Approve: clear suspension entirely
649 sqlx::query(
650 r"
651 UPDATE users
652 SET appeal_decision = $2,
653 appeal_response = $3,
654 appeal_decided_at = NOW(),
655 suspended_at = NULL,
656 suspension_reason = NULL,
657 updated_at = NOW()
658 WHERE id = $1
659 ",
660 )
661 .bind(user_id)
662 .bind(decision)
663 .bind(response)
664 .execute(pool)
665 .await?;
666 } else {
667 // Deny: keep suspension, record decision
668 sqlx::query(
669 r"
670 UPDATE users
671 SET appeal_decision = $2,
672 appeal_response = $3,
673 appeal_decided_at = NOW(),
674 updated_at = NOW()
675 WHERE id = $1
676 ",
677 )
678 .bind(user_id)
679 .bind(decision)
680 .bind(response)
681 .execute(pool)
682 .await?;
683 }
684
685 Ok(())
686 }
687
688 /// Admin query: users with a pending appeal (submitted but not yet decided).
689 #[tracing::instrument(skip_all)]
690 pub async fn get_pending_appeals(pool: &PgPool) -> Result<Vec<DbUser>> {
691 let users = sqlx::query_as::<_, DbUser>(
692 r"
693 SELECT * FROM users
694 WHERE appeal_submitted_at IS NOT NULL
695 AND appeal_decided_at IS NULL
696 ORDER BY appeal_submitted_at ASC
697 LIMIT 500
698 ",
699 )
700 .fetch_all(pool)
701 .await?;
702
703 Ok(users)
704 }
705
706 /// Admin query: all users, optionally filtered by suspension status, with pagination.
707 #[tracing::instrument(skip_all)]
708 pub async fn get_all_users(
709 pool: &PgPool,
710 filter: Option<&str>,
711 limit: i64,
712 offset: i64,
713 ) -> Result<Vec<DbUser>> {
714 let limit = limit.min(200);
715 let users = match filter {
716 Some("suspended") => {
717 sqlx::query_as::<_, DbUser>(
718 "SELECT * FROM users WHERE suspended_at IS NOT NULL ORDER BY suspended_at DESC LIMIT $1 OFFSET $2",
719 )
720 .bind(limit)
721 .bind(offset)
722 .fetch_all(pool)
723 .await?
724 }
725 Some("active") => {
726 sqlx::query_as::<_, DbUser>(
727 "SELECT * FROM users WHERE suspended_at IS NULL ORDER BY created_at DESC LIMIT $1 OFFSET $2",
728 )
729 .bind(limit)
730 .bind(offset)
731 .fetch_all(pool)
732 .await?
733 }
734 // Creators with a custom page; most recently changed first, so
735 // "recently changed" surfaces naturally at the top.
736 Some("custom_pages") => {
737 sqlx::query_as::<_, DbUser>(
738 "SELECT * FROM users WHERE custom_html <> '' OR custom_css <> '' \
739 ORDER BY custom_pages_updated_at DESC NULLS LAST, created_at DESC LIMIT $1 OFFSET $2",
740 )
741 .bind(limit)
742 .bind(offset)
743 .fetch_all(pool)
744 .await?
745 }
746 Some("pages_locked") => {
747 sqlx::query_as::<_, DbUser>(
748 "SELECT * FROM users WHERE custom_pages_locked = true ORDER BY created_at DESC LIMIT $1 OFFSET $2",
749 )
750 .bind(limit)
751 .bind(offset)
752 .fetch_all(pool)
753 .await?
754 }
755 _ => {
756 sqlx::query_as::<_, DbUser>(
757 "SELECT * FROM users ORDER BY created_at DESC LIMIT $1 OFFSET $2",
758 )
759 .bind(limit)
760 .bind(offset)
761 .fetch_all(pool)
762 .await?
763 }
764 };
765
766 Ok(users)
767 }
768
769 /// Count users matching a filter (for pagination totals).
770 #[tracing::instrument(skip_all)]
771 pub async fn count_users(pool: &PgPool, filter: Option<&str>) -> Result<i64> {
772 let count = match filter {
773 Some("suspended") => {
774 sqlx::query_scalar::<_, i64>(
775 "SELECT COUNT(*) FROM users WHERE suspended_at IS NOT NULL",
776 )
777 .fetch_one(pool)
778 .await?
779 }
780 Some("active") => {
781 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users WHERE suspended_at IS NULL")
782 .fetch_one(pool)
783 .await?
784 }
785 Some("custom_pages") => {
786 sqlx::query_scalar::<_, i64>(
787 "SELECT COUNT(*) FROM users WHERE custom_html <> '' OR custom_css <> ''",
788 )
789 .fetch_one(pool)
790 .await?
791 }
792 Some("pages_locked") => {
793 sqlx::query_scalar::<_, i64>(
794 "SELECT COUNT(*) FROM users WHERE custom_pages_locked = true",
795 )
796 .fetch_one(pool)
797 .await?
798 }
799 _ => {
800 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users")
801 .fetch_one(pool)
802 .await?
803 }
804 };
805
806 Ok(count)
807 }
808
809 /// Count total and suspended users in a single query.
810 #[tracing::instrument(skip_all)]
811 pub async fn count_users_summary(pool: &PgPool) -> Result<(i64, i64)> {
812 let (total, suspended): (i64, i64) = sqlx::query_as(
813 r"
814 SELECT
815 COUNT(*),
816 COUNT(*) FILTER (WHERE suspended_at IS NOT NULL)
817 FROM users
818 ",
819 )
820 .fetch_one(pool)
821 .await?;
822
823 Ok((total, suspended))
824 }
825
826 /// Get all user emails for bulk notifications (e.g. shutdown notice).
827 ///
828 /// Capped to bound memory on a full-table scan; if the cap is ever hit the WARN
829 /// fires so we know to switch to a paged dispatch model (mirrors
830 /// [`get_status_alert_subscribers`]).
831 #[tracing::instrument(skip_all)]
832 pub async fn get_all_user_emails(pool: &PgPool) -> Result<Vec<(String, Option<String>)>> {
833 const ALL_EMAILS_CAP: i64 = 50_000;
834 let rows = sqlx::query_as::<_, (String, Option<String>)>(
835 "SELECT email, display_name FROM users ORDER BY created_at ASC LIMIT $1",
836 )
837 .bind(ALL_EMAILS_CAP)
838 .fetch_all(pool)
839 .await?;
840
841 if rows.len() as i64 == ALL_EMAILS_CAP {
842 tracing::warn!(
843 cap = ALL_EMAILS_CAP,
844 "get_all_user_emails hit its cap; some users omitted, switch to paged dispatch"
845 );
846 }
847
848 Ok(rows)
849 }
850
851 /// Update a user's email notification preferences.
852 /// Email notification toggles for a user, one field per `users` column.
853 #[derive(Debug, Clone, Copy)]
854 pub struct NotificationPreferences {
855 pub notify_sale: bool,
856 pub notify_follower: bool,
857 pub notify_release: bool,
858 pub login_notification_enabled: bool,
859 pub notify_issues: bool,
860 pub notify_status: bool,
861 }
862
863 #[tracing::instrument(skip_all)]
864 pub async fn update_notification_preferences(
865 pool: &PgPool,
866 id: UserId,
867 prefs: NotificationPreferences,
868 ) -> Result<()> {
869 let NotificationPreferences {
870 notify_sale,
871 notify_follower,
872 notify_release,
873 login_notification_enabled,
874 notify_issues,
875 notify_status,
876 } = prefs;
877 sqlx::query(
878 r"
879 UPDATE users
880 SET notify_sale = $2,
881 notify_follower = $3,
882 notify_release = $4,
883 login_notification_enabled = $5,
884 notify_issues = $6,
885 notify_status = $7,
886 updated_at = NOW()
887 WHERE id = $1
888 ",
889 )
890 .bind(id)
891 .bind(notify_sale)
892 .bind(notify_follower)
893 .bind(notify_release)
894 .bind(login_notification_enabled)
895 .bind(notify_issues)
896 .bind(notify_status)
897 .execute(pool)
898 .await?;
899
900 Ok(())
901 }
902
903 /// Update a user's tip preferences (tips_enabled toggle and notification).
904 #[tracing::instrument(skip_all)]
905 pub async fn update_tip_preferences(
906 pool: &PgPool,
907 id: UserId,
908 tips_enabled: bool,
909 notify_tip: bool,
910 ) -> Result<()> {
911 sqlx::query(
912 r"
913 UPDATE users
914 SET tips_enabled = $2,
915 notify_tip = $3,
916 updated_at = NOW()
917 WHERE id = $1
918 ",
919 )
920 .bind(id)
921 .bind(tips_enabled)
922 .bind(notify_tip)
923 .execute(pool)
924 .await?;
925
926 Ok(())
927 }
928
929 /// Disable a single notification preference by column name.
930 ///
931 /// Used by the email unsubscribe handler. Only accepts known column names
932 /// to prevent SQL injection.
933 #[tracing::instrument(skip_all)]
934 pub async fn disable_notification(
935 pool: &PgPool,
936 user_id: UserId,
937 preference: &str,
938 ) -> Result<bool> {
939 let sql = match preference {
940 "notify_sale" => "UPDATE users SET notify_sale = false, updated_at = NOW() WHERE id = $1",
941 "notify_follower" => {
942 "UPDATE users SET notify_follower = false, updated_at = NOW() WHERE id = $1"
943 }
944 "notify_release" => {
945 "UPDATE users SET notify_release = false, updated_at = NOW() WHERE id = $1"
946 }
947 "login_notification_enabled" => {
948 "UPDATE users SET login_notification_enabled = false, updated_at = NOW() WHERE id = $1"
949 }
950 "notify_issues" => {
951 "UPDATE users SET notify_issues = false, updated_at = NOW() WHERE id = $1"
952 }
953 "notify_tip" => "UPDATE users SET notify_tip = false, updated_at = NOW() WHERE id = $1",
954 "notify_status" => {
955 "UPDATE users SET notify_status = false, updated_at = NOW() WHERE id = $1"
956 }
957 _ => return Ok(false),
958 };
959 let result = sqlx::query(sql).bind(user_id).execute(pool).await?;
960 Ok(result.rows_affected() > 0)
961 }
962
963 /// A user who opted into platform status notifications.
964 #[derive(sqlx::FromRow)]
965 pub struct StatusAlertSubscriber {
966 pub id: UserId,
967 pub email: Email,
968 pub display_name: Option<String>,
969 }
970
971 /// Get all users who opted into platform status notifications.
972 ///
973 /// Hard cap at 10k rows so the monitor's status-change fan-out can't unbox
974 /// an unbounded query into RAM. The 100ms pacing in `monitor.rs` already
975 /// limits fan-out throughput to ~600/minute, anything past 10k would
976 /// chew through Postmark rate limits anyway. If we ever hit the cap a
977 /// WARN fires so we know to switch to a paged dispatch model.
978 #[tracing::instrument(skip_all)]
979 pub async fn get_status_alert_subscribers(pool: &PgPool) -> Result<Vec<StatusAlertSubscriber>> {
980 const STATUS_SUBSCRIBER_CAP: i64 = 10_000;
981 let rows = sqlx::query_as::<_, StatusAlertSubscriber>(
982 "SELECT id, email, display_name FROM users \
983 WHERE notify_status = true AND deactivated_at IS NULL \
984 ORDER BY id LIMIT $1",
985 )
986 .bind(STATUS_SUBSCRIBER_CAP)
987 .fetch_all(pool)
988 .await?;
989 if rows.len() as i64 == STATUS_SUBSCRIBER_CAP {
990 tracing::warn!(
991 cap = STATUS_SUBSCRIBER_CAP,
992 "get_status_alert_subscribers hit hard cap; promote to paged dispatch"
993 );
994 }
995 Ok(rows)
996 }
997
998 /// Atomically check-and-set broadcast timestamp. Returns false if already sent within 24 hours.
999 #[tracing::instrument(skip_all)]
1000 pub async fn try_set_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<bool> {
1001 let result = sqlx::query(
1002 r"
1003 UPDATE users
1004 SET last_broadcast_at = NOW()
1005 WHERE id = $1
1006 AND (last_broadcast_at IS NULL OR last_broadcast_at < NOW() - INTERVAL '24 hours')
1007 ",
1008 )
1009 .bind(user_id)
1010 .execute(pool)
1011 .await?;
1012
1013 Ok(result.rows_affected() > 0)
1014 }
1015
1016 /// Release the 24h broadcast slot. Used when a broadcast is refused after the
1017 /// slot has already been claimed (e.g. recipient cap exceeded) so the creator
1018 /// can retry without waiting a day.
1019 #[tracing::instrument(skip_all)]
1020 pub async fn clear_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<()> {
1021 sqlx::query("UPDATE users SET last_broadcast_at = NULL WHERE id = $1")
1022 .bind(user_id)
1023 .execute(pool)
1024 .await?;
1025 Ok(())
1026 }
1027
1028 // ── Upload Trust ──
1029
1030 /// Check if a user is trusted for uploads (bypasses review queue).
1031 #[tracing::instrument(skip_all)]
1032 pub async fn is_upload_trusted(pool: &PgPool, user_id: UserId) -> Result<bool> {
1033 let trusted = sqlx::query_scalar::<_, bool>("SELECT upload_trusted FROM users WHERE id = $1")
1034 .bind(user_id)
1035 .fetch_one(pool)
1036 .await?;
1037
1038 Ok(trusted)
1039 }
1040
1041 /// Set a user's upload trust status.
1042 #[tracing::instrument(skip_all)]
1043 pub async fn set_upload_trusted(pool: &PgPool, user_id: UserId, trusted: bool) -> Result<()> {
1044 sqlx::query(
1045 r"
1046 UPDATE users
1047 SET upload_trusted = $2,
1048 updated_at = NOW()
1049 WHERE id = $1
1050 ",
1051 )
1052 .bind(user_id)
1053 .bind(trusted)
1054 .execute(pool)
1055 .await?;
1056
1057 Ok(())
1058 }
1059
1060 /// Moderation kill switch for custom pages. While locked, the creator can't
1061 /// edit their custom pages and the live pages render the platform default.
1062 /// Reversible: unlocking restores the (preserved) custom source.
1063 pub async fn set_custom_pages_locked(pool: &PgPool, user_id: UserId, locked: bool) -> Result<()> {
1064 sqlx::query("UPDATE users SET custom_pages_locked = $2, cache_generation = cache_generation + 1 WHERE id = $1")
1065 .bind(user_id)
1066 .bind(locked)
1067 .execute(pool)
1068 .await?;
1069 Ok(())
1070 }
1071
1072 // ── Onboarding email drip ──
1073
1074 /// Users who need the next onboarding email. Returns users at a given step
1075 /// whose last email was sent more than `min_age` ago (or never).
1076 #[tracing::instrument(skip_all)]
1077 pub async fn get_onboarding_candidates(
1078 pool: &PgPool,
1079 step: i16,
1080 min_age: chrono::Duration,
1081 ) -> Result<Vec<DbUser>> {
1082 let cutoff = chrono::Utc::now() - min_age;
1083 // Per-tick LIMIT bounds the scheduler's input list. The caller advances
1084 // each returned user's step (so the WHERE re-excludes them), meaning the
1085 // remainder is drained on the next tick, same re-tick pattern as the
1086 // sandbox/terminated/content-removal cleanup queries. Run #14 MEDIUM: a
1087 // signup surge must not load an unbounded user vec into the lock-held tick.
1088 let users = sqlx::query_as::<_, DbUser>(
1089 "SELECT * FROM users
1090 WHERE onboarding_email_step = $1
1091 AND (onboarding_email_sent_at IS NULL OR onboarding_email_sent_at < $2)
1092 AND suspended_at IS NULL
1093 ORDER BY onboarding_email_sent_at ASC NULLS FIRST
1094 LIMIT 1000",
1095 )
1096 .bind(step)
1097 .bind(cutoff)
1098 .fetch_all(pool)
1099 .await?;
1100 Ok(users)
1101 }
1102
1103 /// Advance a user's onboarding email step and record the send time.
1104 #[tracing::instrument(skip_all)]
1105 pub async fn advance_onboarding_step(pool: &PgPool, user_id: UserId, new_step: i16) -> Result<()> {
1106 sqlx::query(
1107 "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = $1",
1108 )
1109 .bind(user_id)
1110 .bind(new_step)
1111 .execute(pool)
1112 .await?;
1113 Ok(())
1114 }
1115
1116 /// Advance onboarding step for multiple users in a single query.
1117 #[tracing::instrument(skip_all)]
1118 pub async fn batch_advance_onboarding_step(
1119 pool: &PgPool,
1120 user_ids: &[UserId],
1121 new_step: i16,
1122 ) -> Result<()> {
1123 sqlx::query(
1124 "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = ANY($1)",
1125 )
1126 .bind(user_ids)
1127 .bind(new_step)
1128 .execute(pool)
1129 .await?;
1130 Ok(())
1131 }
1132
1133 /// Mark a user as a founder. Called when they start a creator-tier
1134 /// subscription while the founder pricing window is open. Sticky; never
1135 /// reset, even on cancellation. Subsequent re-subscriptions during the
1136 /// window keep their founder status. After the window closes, eligibility
1137 /// is determined by `founder_locked_at` (stamped only for users with an
1138 /// active subscription at the close-time snapshot).
1139 ///
1140 /// **DIY exclusion**: DIY-tier accounts are not full members and must not
1141 /// qualify for founder pricing (`project_founder_pricing.md` § decision 5).
1142 /// Only call this from creator-tier (Basic/SmallFiles/BigFiles/Everything)
1143 /// checkout paths. When DIY ships, its checkout path must NOT invoke this.
1144 #[tracing::instrument(skip_all)]
1145 pub async fn mark_user_as_founder(pool: &PgPool, user_id: UserId) -> Result<()> {
1146 sqlx::query(
1147 r"
1148 UPDATE users
1149 SET is_founder = TRUE,
1150 updated_at = NOW()
1151 WHERE id = $1 AND is_founder = FALSE
1152 ",
1153 )
1154 .bind(user_id)
1155 .execute(pool)
1156 .await?;
1157 Ok(())
1158 }
1159
1160 /// Close the founder pricing window by stamping `founder_locked_at` on every
1161 /// user who is currently flagged `is_founder` AND has an active creator-tier
1162 /// subscription. Returns the number of users locked in. Idempotent: skips
1163 /// any user already locked. Intended to be called once from an admin tool
1164 /// at the moment the founder window closes.
1165 #[tracing::instrument(skip_all)]
1166 pub async fn lock_in_founders_with_active_subscriptions(pool: &PgPool) -> Result<u64> {
1167 let result = sqlx::query(
1168 r"
1169 UPDATE users u
1170 SET founder_locked_at = NOW(),
1171 updated_at = NOW()
1172 WHERE u.is_founder = TRUE
1173 AND u.founder_locked_at IS NULL
1174 AND EXISTS (
1175 SELECT 1 FROM creator_subscriptions s
1176 WHERE s.user_id = u.id
1177 AND s.status = 'active'
1178 )
1179 ",
1180 )
1181 .execute(pool)
1182 .await?;
1183 Ok(result.rows_affected())
1184 }
1185
1186 /// Update a user's Stripe Tax toggle.
1187 #[tracing::instrument(skip_all)]
1188 pub async fn update_stripe_tax_enabled(
1189 pool: &PgPool,
1190 user_id: UserId,
1191 enabled: bool,
1192 ) -> Result<()> {
1193 sqlx::query(
1194 r"
1195 UPDATE users
1196 SET stripe_tax_enabled = $2,
1197 updated_at = NOW()
1198 WHERE id = $1
1199 ",
1200 )
1201 .bind(user_id)
1202 .bind(enabled)
1203 .execute(pool)
1204 .await?;
1205
1206 Ok(())
1207 }
1208
1209 /// Disconnect a user's Stripe account
1210 #[tracing::instrument(skip_all)]
1211 pub async fn disconnect_user_stripe(pool: &PgPool, user_id: UserId) -> Result<DbUser> {
1212 let user = sqlx::query_as::<_, DbUser>(
1213 r"
1214 UPDATE users
1215 SET stripe_account_id = NULL,
1216 stripe_onboarding_complete = false,
1217 stripe_payouts_enabled = false,
1218 stripe_charges_enabled = false,
1219 updated_at = NOW()
1220 WHERE id = $1
1221 RETURNING *
1222 ",
1223 )
1224 .bind(user_id)
1225 .fetch_one(pool)
1226 .await?;
1227
1228 Ok(user)
1229 }
1230
1231 /// Fetch the current cache generation for a user (cheap, indexed lookup).
1232 #[tracing::instrument(skip_all)]
1233 pub async fn get_cache_generation(pool: &PgPool, user_id: UserId) -> Result<i64> {
1234 let generation =
1235 sqlx::query_scalar::<_, i64>("SELECT cache_generation FROM users WHERE id = $1")
1236 .bind(user_id)
1237 .fetch_one(pool)
1238 .await?;
1239
1240 Ok(generation)
1241 }
1242
1243 /// Atomically increment the user's cache generation counter.
1244 /// Call after any write that changes user-visible dashboard data.
1245 #[tracing::instrument(skip_all)]
1246 pub async fn bump_cache_generation(pool: &PgPool, user_id: UserId) -> Result<()> {
1247 sqlx::query("UPDATE users SET cache_generation = cache_generation + 1 WHERE id = $1")
1248 .bind(user_id)
1249 .execute(pool)
1250 .await?;
1251
1252 Ok(())
1253 }
1254
1255 /// Look up a verified user by email (case-insensitive).
1256 /// Returns the user ID if a verified account exists with that email.
1257 #[tracing::instrument(skip_all)]
1258 pub async fn get_verified_user_id_by_email(pool: &PgPool, email: &Email) -> Result<Option<UserId>> {
1259 let id = sqlx::query_scalar(
1260 "SELECT id FROM users WHERE LOWER(email) = LOWER($1) AND email_verified = true",
1261 )
1262 .bind(email)
1263 .fetch_optional(pool)
1264 .await?;
1265
1266 Ok(id)
1267 }
1268