Skip to main content

max / makenotwork

29.9 KB · 1062 lines History Blame Raw
1 //! User account CRUD, profile updates, and lookup queries.
2
3 use sqlx::PgPool;
4
5 use super::enums::AppealDecision;
6 use super::models::*;
7 use super::validated_types::{Email, Username};
8 use super::UserId;
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 /// Fetch a user by primary key. Returns `None` if not found.
36 #[tracing::instrument(skip_all)]
37 pub async fn get_user_by_id(pool: &PgPool, id: UserId) -> Result<Option<DbUser>> {
38 let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE id = $1")
39 .bind(id)
40 .fetch_optional(pool)
41 .await?;
42
43 Ok(user)
44 }
45
46 /// Fetch multiple users by ID in a single query.
47 #[tracing::instrument(skip_all)]
48 pub async fn get_users_by_ids(pool: &PgPool, ids: &[UserId]) -> Result<Vec<DbUser>> {
49 let users = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE id = ANY($1)")
50 .bind(ids)
51 .fetch_all(pool)
52 .await?;
53 Ok(users)
54 }
55
56 /// Fetch a user by username. Returns `None` if not found.
57 #[tracing::instrument(skip_all)]
58 pub async fn get_user_by_username(pool: &PgPool, username: &Username) -> Result<Option<DbUser>> {
59 let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE username = $1")
60 .bind(username)
61 .fetch_optional(pool)
62 .await?;
63
64 Ok(user)
65 }
66
67 /// Fetch a user by email address. Returns `None` if not found.
68 #[tracing::instrument(skip_all)]
69 pub async fn get_user_by_email(pool: &PgPool, email: &Email) -> Result<Option<DbUser>> {
70 let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE email = $1")
71 .bind(email)
72 .fetch_optional(pool)
73 .await?;
74
75 Ok(user)
76 }
77
78 /// Update a user's display name and/or bio (COALESCE keeps existing values when `None`).
79 #[tracing::instrument(skip_all)]
80 pub async fn update_user_profile(
81 pool: &PgPool,
82 id: UserId,
83 display_name: Option<&str>,
84 bio: Option<&str>,
85 ) -> Result<DbUser> {
86 let user = sqlx::query_as::<_, DbUser>(
87 r#"
88 UPDATE users
89 SET display_name = COALESCE($2, display_name),
90 bio = COALESCE($3, bio)
91 WHERE id = $1
92 RETURNING *
93 "#,
94 )
95 .bind(id)
96 .bind(display_name)
97 .bind(bio)
98 .fetch_one(pool)
99 .await?;
100
101 Ok(user)
102 }
103
104 /// Replace a user's password hash and invalidate outstanding JWTs.
105 #[tracing::instrument(skip_all)]
106 pub async fn update_user_password(pool: &PgPool, id: UserId, password_hash: &str) -> Result<()> {
107 sqlx::query("UPDATE users SET password_hash = $2, jwt_invalidated_at = NOW() WHERE id = $1")
108 .bind(id)
109 .bind(password_hash)
110 .execute(pool)
111 .await?;
112
113 Ok(())
114 }
115
116 /// Self-deactivate an account (enter limbo state).
117 ///
118 /// Bumps `jwt_invalidated_at` so any outstanding SyncKit JWTs minted from
119 /// this account stop authenticating immediately.
120 #[tracing::instrument(skip_all)]
121 pub async fn deactivate_user(pool: &PgPool, id: UserId) -> Result<()> {
122 sqlx::query(
123 "UPDATE users SET deactivated_at = NOW(), jwt_invalidated_at = NOW(), updated_at = NOW() WHERE id = $1",
124 )
125 .bind(id)
126 .execute(pool)
127 .await?;
128
129 Ok(())
130 }
131
132 /// Reactivate a self-deactivated account.
133 #[tracing::instrument(skip_all)]
134 pub async fn reactivate_user(pool: &PgPool, id: UserId) -> Result<()> {
135 sqlx::query(
136 "UPDATE users SET deactivated_at = NULL, updated_at = NOW() WHERE id = $1",
137 )
138 .bind(id)
139 .execute(pool)
140 .await?;
141
142 Ok(())
143 }
144
145 /// Admin: permanently terminate an account (enforcement ladder step 4).
146 /// The user has 30 days to export data. After that, the scheduler deletes the account.
147 /// The account must already be suspended.
148 #[tracing::instrument(skip_all)]
149 pub async fn terminate_user(pool: &PgPool, id: UserId) -> Result<()> {
150 sqlx::query(
151 "UPDATE users SET terminated_at = NOW(), jwt_invalidated_at = NOW(), updated_at = NOW() WHERE id = $1",
152 )
153 .bind(id)
154 .execute(pool)
155 .await?;
156
157 Ok(())
158 }
159
160 /// Get user IDs of terminated accounts whose 30-day export window has expired.
161 #[tracing::instrument(skip_all)]
162 pub async fn get_expired_terminated_ids(pool: &PgPool) -> Result<Vec<UserId>> {
163 let ids: Vec<UserId> = sqlx::query_scalar(
164 r#"
165 SELECT id FROM users
166 WHERE terminated_at IS NOT NULL
167 AND terminated_at < NOW() - INTERVAL '30 days'
168 "#,
169 )
170 .fetch_all(pool)
171 .await?;
172
173 Ok(ids)
174 }
175
176 /// Permanently delete a user by ID.
177 #[tracing::instrument(skip_all)]
178 pub async fn delete_user(pool: &PgPool, id: UserId) -> Result<()> {
179 sqlx::query("DELETE FROM users WHERE id = $1")
180 .bind(id)
181 .execute(pool)
182 .await?;
183
184 Ok(())
185 }
186
187 /// Check whether this creator has any completed sales (transactions where they were the seller).
188 #[tracing::instrument(skip_all)]
189 pub async fn has_completed_sales(pool: &PgPool, id: UserId) -> Result<bool> {
190 let count: i64 = sqlx::query_scalar(
191 "SELECT COUNT(*) FROM transactions WHERE seller_id = $1 AND status = 'completed'",
192 )
193 .bind(id)
194 .fetch_one(pool)
195 .await?;
196
197 Ok(count > 0)
198 }
199
200 /// Schedule content removal 90 days from now. The user row is hidden from public
201 /// views but items remain accessible to buyers who previously purchased them.
202 /// After 90 days the scheduler deletes S3 objects and the user row.
203 #[tracing::instrument(skip_all)]
204 pub async fn schedule_content_removal(pool: &PgPool, id: UserId) -> Result<()> {
205 sqlx::query(
206 r#"
207 UPDATE users
208 SET content_removal_at = NOW() + INTERVAL '90 days',
209 deactivated_at = NOW(),
210 updated_at = NOW()
211 WHERE id = $1
212 "#,
213 )
214 .bind(id)
215 .execute(pool)
216 .await?;
217
218 Ok(())
219 }
220
221 /// Get user IDs whose 90-day content removal grace period has expired.
222 #[tracing::instrument(skip_all)]
223 pub async fn get_expired_content_removal_ids(pool: &PgPool) -> Result<Vec<UserId>> {
224 let ids: Vec<UserId> = sqlx::query_scalar(
225 r#"
226 SELECT id FROM users
227 WHERE content_removal_at IS NOT NULL
228 AND content_removal_at < NOW()
229 "#,
230 )
231 .fetch_all(pool)
232 .await?;
233
234 Ok(ids)
235 }
236
237 /// Create an ephemeral sandbox user. Returns the created row.
238 ///
239 /// The user gets `can_create_projects = true`, `email_verified = true`,
240 /// a SmallFiles creator tier, and a tight storage cap. The row is
241 /// automatically cleaned up by the scheduler after `sandbox_expires_at`.
242 #[tracing::instrument(skip_all)]
243 pub async fn create_sandbox_user(
244 pool: &PgPool,
245 username: &Username,
246 email: &Email,
247 password_hash: &str,
248 expiry_secs: i64,
249 ) -> Result<DbUser> {
250 let user = sqlx::query_as::<_, DbUser>(
251 r#"
252 INSERT INTO users (
253 username, email, password_hash,
254 is_sandbox, sandbox_expires_at,
255 can_create_projects, email_verified,
256 creator_tier
257 )
258 VALUES (
259 $1, $2, $3,
260 TRUE, NOW() + make_interval(secs => $4::float8),
261 TRUE, TRUE,
262 'small_files'
263 )
264 RETURNING *
265 "#,
266 )
267 .bind(username)
268 .bind(email)
269 .bind(password_hash)
270 .bind(expiry_secs as f64)
271 .fetch_one(pool)
272 .await?;
273
274 Ok(user)
275 }
276
277 /// Return IDs of sandbox users whose expiry has passed.
278 #[tracing::instrument(skip_all)]
279 pub async fn get_expired_sandbox_ids(pool: &PgPool) -> Result<Vec<UserId>> {
280 let ids = sqlx::query_scalar::<_, UserId>(
281 "SELECT id FROM users WHERE is_sandbox = TRUE AND sandbox_expires_at < NOW()",
282 )
283 .fetch_all(pool)
284 .await?;
285
286 Ok(ids)
287 }
288
289 /// Count active (non-expired) sandbox accounts created from a given IP.
290 /// Used to enforce the per-IP concurrent sandbox cap.
291 #[tracing::instrument(skip_all)]
292 pub async fn count_active_sandboxes_by_ip(pool: &PgPool, ip: &str) -> Result<i64> {
293 let count: i64 = sqlx::query_scalar(
294 r#"
295 SELECT COUNT(*) FROM users u
296 JOIN user_sessions us ON us.user_id = u.id
297 WHERE u.is_sandbox = TRUE
298 AND u.sandbox_expires_at > NOW()
299 AND us.ip_address = $1
300 "#,
301 )
302 .bind(ip)
303 .fetch_one(pool)
304 .await?;
305
306 Ok(count)
307 }
308
309 /// Update user's Stripe Connect account information after OAuth
310 #[tracing::instrument(skip_all)]
311 pub async fn update_user_stripe_account(
312 pool: &PgPool,
313 user_id: UserId,
314 stripe_account_id: &str,
315 onboarding_complete: bool,
316 payouts_enabled: bool,
317 charges_enabled: bool,
318 ) -> Result<DbUser> {
319 let user = sqlx::query_as::<_, DbUser>(
320 r#"
321 UPDATE users
322 SET stripe_account_id = $2,
323 stripe_onboarding_complete = $3,
324 stripe_payouts_enabled = $4,
325 stripe_charges_enabled = $5,
326 updated_at = NOW()
327 WHERE id = $1
328 RETURNING *
329 "#,
330 )
331 .bind(user_id)
332 .bind(stripe_account_id)
333 .bind(onboarding_complete)
334 .bind(payouts_enabled)
335 .bind(charges_enabled)
336 .fetch_one(pool)
337 .await?;
338
339 Ok(user)
340 }
341
342 /// Atomically set a user's Stripe Connect account ID, but only if one is not
343 /// already set. Returns `Some(user)` on success, or `None` if another request
344 /// already claimed the slot (race-condition guard).
345 #[tracing::instrument(skip_all)]
346 pub async fn try_set_stripe_account(
347 pool: &PgPool,
348 user_id: UserId,
349 stripe_account_id: &str,
350 ) -> Result<Option<DbUser>> {
351 let user = sqlx::query_as::<_, DbUser>(
352 r#"
353 UPDATE users
354 SET stripe_account_id = $2,
355 stripe_onboarding_complete = false,
356 stripe_payouts_enabled = false,
357 stripe_charges_enabled = false,
358 updated_at = NOW()
359 WHERE id = $1 AND (stripe_account_id IS NULL OR stripe_account_id = '')
360 RETURNING *
361 "#,
362 )
363 .bind(user_id)
364 .bind(stripe_account_id)
365 .fetch_optional(pool)
366 .await?;
367
368 Ok(user)
369 }
370
371 /// Update user's Stripe status from webhook (finds user by stripe_account_id)
372 #[tracing::instrument(skip_all)]
373 pub async fn update_user_stripe_status(
374 pool: &PgPool,
375 stripe_account_id: &str,
376 onboarding_complete: bool,
377 payouts_enabled: bool,
378 charges_enabled: bool,
379 ) -> Result<Option<DbUser>> {
380 let user = sqlx::query_as::<_, DbUser>(
381 r#"
382 UPDATE users
383 SET stripe_onboarding_complete = $2,
384 stripe_payouts_enabled = $3,
385 stripe_charges_enabled = $4,
386 updated_at = NOW()
387 WHERE stripe_account_id = $1
388 RETURNING *
389 "#,
390 )
391 .bind(stripe_account_id)
392 .bind(onboarding_complete)
393 .bind(payouts_enabled)
394 .bind(charges_enabled)
395 .fetch_optional(pool)
396 .await?;
397
398 Ok(user)
399 }
400
401 /// Mark a user's email as verified
402 #[tracing::instrument(skip_all)]
403 pub async fn verify_user_email(pool: &PgPool, user_id: UserId) -> Result<()> {
404 sqlx::query(
405 r#"
406 UPDATE users
407 SET email_verified = true,
408 email_verification_token = NULL,
409 updated_at = NOW()
410 WHERE id = $1
411 "#,
412 )
413 .bind(user_id)
414 .execute(pool)
415 .await?;
416
417 Ok(())
418 }
419
420 // ── Suspension / Appeals ──
421
422 /// Suspend a user account, clearing any prior appeal fields.
423 ///
424 /// Bumps `jwt_invalidated_at` so SyncKit JWTs minted for this user expire
425 /// at the next extractor check (subject to `SESSION_TOUCH_CACHE_SECS`).
426 #[tracing::instrument(skip_all)]
427 pub async fn suspend_user(pool: &PgPool, user_id: UserId, reason: &str) -> Result<()> {
428 sqlx::query(
429 r#"
430 UPDATE users
431 SET suspended_at = NOW(),
432 suspension_reason = $2,
433 jwt_invalidated_at = NOW(),
434 appeal_text = NULL,
435 appeal_submitted_at = NULL,
436 appeal_decision = NULL,
437 appeal_response = NULL,
438 appeal_decided_at = NULL,
439 updated_at = NOW()
440 WHERE id = $1
441 "#,
442 )
443 .bind(user_id)
444 .bind(reason)
445 .execute(pool)
446 .await?;
447
448 Ok(())
449 }
450
451 /// Remove suspension and clear all suspension/appeal fields.
452 #[tracing::instrument(skip_all)]
453 pub async fn unsuspend_user(pool: &PgPool, user_id: UserId) -> Result<()> {
454 sqlx::query(
455 r#"
456 UPDATE users
457 SET suspended_at = NULL,
458 suspension_reason = NULL,
459 appeal_text = NULL,
460 appeal_submitted_at = NULL,
461 appeal_decision = NULL,
462 appeal_response = NULL,
463 appeal_decided_at = NULL,
464 updated_at = NOW()
465 WHERE id = $1
466 "#,
467 )
468 .bind(user_id)
469 .execute(pool)
470 .await?;
471
472 Ok(())
473 }
474
475 // ── Creator pause (voluntary) ──
476
477 /// Set the creator_paused_at timestamp (voluntary pause).
478 #[tracing::instrument(skip_all)]
479 pub async fn pause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
480 sqlx::query(
481 "UPDATE users SET creator_paused_at = NOW(), updated_at = NOW() WHERE id = $1",
482 )
483 .bind(user_id)
484 .execute(pool)
485 .await?;
486
487 Ok(())
488 }
489
490 /// Clear the creator_paused_at timestamp (resume from voluntary pause).
491 #[tracing::instrument(skip_all)]
492 pub async fn unpause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
493 sqlx::query(
494 "UPDATE users SET creator_paused_at = NULL, updated_at = NOW() WHERE id = $1",
495 )
496 .bind(user_id)
497 .execute(pool)
498 .await?;
499
500 Ok(())
501 }
502
503 /// Submit an appeal for a suspended account, clearing any prior decision.
504 #[tracing::instrument(skip_all)]
505 pub async fn submit_appeal(pool: &PgPool, user_id: UserId, appeal_text: &str) -> Result<()> {
506 sqlx::query(
507 r#"
508 UPDATE users
509 SET appeal_text = $2,
510 appeal_submitted_at = NOW(),
511 appeal_decision = NULL,
512 appeal_response = NULL,
513 appeal_decided_at = NULL,
514 updated_at = NOW()
515 WHERE id = $1 AND suspended_at IS NOT NULL
516 "#,
517 )
518 .bind(user_id)
519 .bind(appeal_text)
520 .execute(pool)
521 .await?;
522
523 Ok(())
524 }
525
526 /// Resolve an appeal. If approved, also clears suspension.
527 #[tracing::instrument(skip_all)]
528 pub async fn resolve_appeal(
529 pool: &PgPool,
530 user_id: UserId,
531 decision: AppealDecision,
532 response: &str,
533 ) -> Result<()> {
534 if decision == AppealDecision::Approved {
535 // Approve: clear suspension entirely
536 sqlx::query(
537 r#"
538 UPDATE users
539 SET appeal_decision = $2,
540 appeal_response = $3,
541 appeal_decided_at = NOW(),
542 suspended_at = NULL,
543 suspension_reason = NULL,
544 updated_at = NOW()
545 WHERE id = $1
546 "#,
547 )
548 .bind(user_id)
549 .bind(decision)
550 .bind(response)
551 .execute(pool)
552 .await?;
553 } else {
554 // Deny: keep suspension, record decision
555 sqlx::query(
556 r#"
557 UPDATE users
558 SET appeal_decision = $2,
559 appeal_response = $3,
560 appeal_decided_at = NOW(),
561 updated_at = NOW()
562 WHERE id = $1
563 "#,
564 )
565 .bind(user_id)
566 .bind(decision)
567 .bind(response)
568 .execute(pool)
569 .await?;
570 }
571
572 Ok(())
573 }
574
575 /// Admin query: users with a pending appeal (submitted but not yet decided).
576 #[tracing::instrument(skip_all)]
577 pub async fn get_pending_appeals(pool: &PgPool) -> Result<Vec<DbUser>> {
578 let users = sqlx::query_as::<_, DbUser>(
579 r#"
580 SELECT * FROM users
581 WHERE appeal_submitted_at IS NOT NULL
582 AND appeal_decided_at IS NULL
583 ORDER BY appeal_submitted_at ASC
584 LIMIT 500
585 "#,
586 )
587 .fetch_all(pool)
588 .await?;
589
590 Ok(users)
591 }
592
593 /// Admin query: all users, optionally filtered by suspension status, with pagination.
594 #[tracing::instrument(skip_all)]
595 pub async fn get_all_users(
596 pool: &PgPool,
597 filter: Option<&str>,
598 limit: i64,
599 offset: i64,
600 ) -> Result<Vec<DbUser>> {
601 let limit = limit.min(200);
602 let users = match filter {
603 Some("suspended") => {
604 sqlx::query_as::<_, DbUser>(
605 "SELECT * FROM users WHERE suspended_at IS NOT NULL ORDER BY suspended_at DESC LIMIT $1 OFFSET $2",
606 )
607 .bind(limit)
608 .bind(offset)
609 .fetch_all(pool)
610 .await?
611 }
612 Some("active") => {
613 sqlx::query_as::<_, DbUser>(
614 "SELECT * FROM users WHERE suspended_at IS NULL ORDER BY created_at DESC LIMIT $1 OFFSET $2",
615 )
616 .bind(limit)
617 .bind(offset)
618 .fetch_all(pool)
619 .await?
620 }
621 _ => {
622 sqlx::query_as::<_, DbUser>(
623 "SELECT * FROM users ORDER BY created_at DESC LIMIT $1 OFFSET $2",
624 )
625 .bind(limit)
626 .bind(offset)
627 .fetch_all(pool)
628 .await?
629 }
630 };
631
632 Ok(users)
633 }
634
635 /// Count users matching a filter (for pagination totals).
636 #[tracing::instrument(skip_all)]
637 pub async fn count_users(pool: &PgPool, filter: Option<&str>) -> Result<i64> {
638 let count = match filter {
639 Some("suspended") => {
640 sqlx::query_scalar::<_, i64>(
641 "SELECT COUNT(*) FROM users WHERE suspended_at IS NOT NULL",
642 )
643 .fetch_one(pool)
644 .await?
645 }
646 Some("active") => {
647 sqlx::query_scalar::<_, i64>(
648 "SELECT COUNT(*) FROM users WHERE suspended_at IS NULL",
649 )
650 .fetch_one(pool)
651 .await?
652 }
653 _ => {
654 sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users")
655 .fetch_one(pool)
656 .await?
657 }
658 };
659
660 Ok(count)
661 }
662
663 /// Count total and suspended users in a single query.
664 #[tracing::instrument(skip_all)]
665 pub async fn count_users_summary(pool: &PgPool) -> Result<(i64, i64)> {
666 let (total, suspended): (i64, i64) = sqlx::query_as(
667 r#"
668 SELECT
669 COUNT(*),
670 COUNT(*) FILTER (WHERE suspended_at IS NOT NULL)
671 FROM users
672 "#,
673 )
674 .fetch_one(pool)
675 .await?;
676
677 Ok((total, suspended))
678 }
679
680 /// Get all user emails for bulk notifications (e.g. shutdown notice).
681 #[tracing::instrument(skip_all)]
682 pub async fn get_all_user_emails(pool: &PgPool) -> Result<Vec<(String, Option<String>)>> {
683 let rows = sqlx::query_as::<_, (String, Option<String>)>(
684 "SELECT email, display_name FROM users ORDER BY created_at ASC",
685 )
686 .fetch_all(pool)
687 .await?;
688
689 Ok(rows)
690 }
691
692 /// Update a user's email notification preferences.
693 #[allow(clippy::too_many_arguments)]
694 #[tracing::instrument(skip_all)]
695 pub async fn update_notification_preferences(
696 pool: &PgPool,
697 id: UserId,
698 notify_sale: bool,
699 notify_follower: bool,
700 notify_release: bool,
701 login_notification_enabled: bool,
702 notify_issues: bool,
703 notify_status: bool,
704 ) -> Result<()> {
705 sqlx::query(
706 r#"
707 UPDATE users
708 SET notify_sale = $2,
709 notify_follower = $3,
710 notify_release = $4,
711 login_notification_enabled = $5,
712 notify_issues = $6,
713 notify_status = $7,
714 updated_at = NOW()
715 WHERE id = $1
716 "#,
717 )
718 .bind(id)
719 .bind(notify_sale)
720 .bind(notify_follower)
721 .bind(notify_release)
722 .bind(login_notification_enabled)
723 .bind(notify_issues)
724 .bind(notify_status)
725 .execute(pool)
726 .await?;
727
728 Ok(())
729 }
730
731 /// Update a user's tip preferences (tips_enabled toggle and notification).
732 #[tracing::instrument(skip_all)]
733 pub async fn update_tip_preferences(
734 pool: &PgPool,
735 id: UserId,
736 tips_enabled: bool,
737 notify_tip: bool,
738 ) -> Result<()> {
739 sqlx::query(
740 r#"
741 UPDATE users
742 SET tips_enabled = $2,
743 notify_tip = $3,
744 updated_at = NOW()
745 WHERE id = $1
746 "#,
747 )
748 .bind(id)
749 .bind(tips_enabled)
750 .bind(notify_tip)
751 .execute(pool)
752 .await?;
753
754 Ok(())
755 }
756
757 /// Disable a single notification preference by column name.
758 ///
759 /// Used by the email unsubscribe handler. Only accepts known column names
760 /// to prevent SQL injection.
761 #[tracing::instrument(skip_all)]
762 pub async fn disable_notification(pool: &PgPool, user_id: UserId, preference: &str) -> Result<bool> {
763 let sql = match preference {
764 "notify_sale" => "UPDATE users SET notify_sale = false, updated_at = NOW() WHERE id = $1",
765 "notify_follower" => "UPDATE users SET notify_follower = false, updated_at = NOW() WHERE id = $1",
766 "notify_release" => "UPDATE users SET notify_release = false, updated_at = NOW() WHERE id = $1",
767 "login_notification_enabled" => "UPDATE users SET login_notification_enabled = false, updated_at = NOW() WHERE id = $1",
768 "notify_issues" => "UPDATE users SET notify_issues = false, updated_at = NOW() WHERE id = $1",
769 "notify_tip" => "UPDATE users SET notify_tip = false, updated_at = NOW() WHERE id = $1",
770 "notify_status" => "UPDATE users SET notify_status = false, updated_at = NOW() WHERE id = $1",
771 _ => return Ok(false),
772 };
773 let result = sqlx::query(sql).bind(user_id).execute(pool).await?;
774 Ok(result.rows_affected() > 0)
775 }
776
777 /// A user who opted into platform status notifications.
778 #[derive(sqlx::FromRow)]
779 pub struct StatusAlertSubscriber {
780 pub id: UserId,
781 pub email: Email,
782 pub display_name: Option<String>,
783 }
784
785 /// Get all users who opted into platform status notifications.
786 ///
787 /// Hard cap at 10k rows so the monitor's status-change fan-out can't unbox
788 /// an unbounded query into RAM. The 100ms pacing in `monitor.rs` already
789 /// limits fan-out throughput to ~600/minute — anything past 10k would
790 /// chew through Postmark rate limits anyway. If we ever hit the cap a
791 /// WARN fires so we know to switch to a paged dispatch model.
792 #[tracing::instrument(skip_all)]
793 pub async fn get_status_alert_subscribers(pool: &PgPool) -> Result<Vec<StatusAlertSubscriber>> {
794 const STATUS_SUBSCRIBER_CAP: i64 = 10_000;
795 let rows = sqlx::query_as::<_, StatusAlertSubscriber>(
796 "SELECT id, email, display_name FROM users \
797 WHERE notify_status = true AND deactivated_at IS NULL \
798 ORDER BY id LIMIT $1",
799 )
800 .bind(STATUS_SUBSCRIBER_CAP)
801 .fetch_all(pool)
802 .await?;
803 if rows.len() as i64 == STATUS_SUBSCRIBER_CAP {
804 tracing::warn!(
805 cap = STATUS_SUBSCRIBER_CAP,
806 "get_status_alert_subscribers hit hard cap; promote to paged dispatch"
807 );
808 }
809 Ok(rows)
810 }
811
812 /// Atomically check-and-set broadcast timestamp. Returns false if already sent within 24 hours.
813 #[tracing::instrument(skip_all)]
814 pub async fn try_set_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<bool> {
815 let result = sqlx::query(
816 r#"
817 UPDATE users
818 SET last_broadcast_at = NOW()
819 WHERE id = $1
820 AND (last_broadcast_at IS NULL OR last_broadcast_at < NOW() - INTERVAL '24 hours')
821 "#,
822 )
823 .bind(user_id)
824 .execute(pool)
825 .await?;
826
827 Ok(result.rows_affected() > 0)
828 }
829
830 /// Release the 24h broadcast slot. Used when a broadcast is refused after the
831 /// slot has already been claimed (e.g. recipient cap exceeded) so the creator
832 /// can retry without waiting a day.
833 #[tracing::instrument(skip_all)]
834 pub async fn clear_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<()> {
835 sqlx::query("UPDATE users SET last_broadcast_at = NULL WHERE id = $1")
836 .bind(user_id)
837 .execute(pool)
838 .await?;
839 Ok(())
840 }
841
842 // ── Upload Trust ──
843
844 /// Check if a user is trusted for uploads (bypasses review queue).
845 #[tracing::instrument(skip_all)]
846 pub async fn is_upload_trusted(pool: &PgPool, user_id: UserId) -> Result<bool> {
847 let trusted = sqlx::query_scalar::<_, bool>(
848 "SELECT upload_trusted FROM users WHERE id = $1",
849 )
850 .bind(user_id)
851 .fetch_one(pool)
852 .await?;
853
854 Ok(trusted)
855 }
856
857 /// Set a user's upload trust status.
858 #[tracing::instrument(skip_all)]
859 pub async fn set_upload_trusted(pool: &PgPool, user_id: UserId, trusted: bool) -> Result<()> {
860 sqlx::query(
861 r#"
862 UPDATE users
863 SET upload_trusted = $2,
864 updated_at = NOW()
865 WHERE id = $1
866 "#,
867 )
868 .bind(user_id)
869 .bind(trusted)
870 .execute(pool)
871 .await?;
872
873 Ok(())
874 }
875
876 // ── Onboarding email drip ──
877
878 /// Users who need the next onboarding email. Returns users at a given step
879 /// whose last email was sent more than `min_age` ago (or never).
880 #[tracing::instrument(skip_all)]
881 pub async fn get_onboarding_candidates(
882 pool: &PgPool,
883 step: i16,
884 min_age: chrono::Duration,
885 ) -> Result<Vec<DbUser>> {
886 let cutoff = chrono::Utc::now() - min_age;
887 let users = sqlx::query_as::<_, DbUser>(
888 "SELECT * FROM users
889 WHERE onboarding_email_step = $1
890 AND (onboarding_email_sent_at IS NULL OR onboarding_email_sent_at < $2)
891 AND suspended_at IS NULL",
892 )
893 .bind(step)
894 .bind(cutoff)
895 .fetch_all(pool)
896 .await?;
897 Ok(users)
898 }
899
900 /// Advance a user's onboarding email step and record the send time.
901 #[tracing::instrument(skip_all)]
902 pub async fn advance_onboarding_step(pool: &PgPool, user_id: UserId, new_step: i16) -> Result<()> {
903 sqlx::query(
904 "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = $1",
905 )
906 .bind(user_id)
907 .bind(new_step)
908 .execute(pool)
909 .await?;
910 Ok(())
911 }
912
913 /// Advance onboarding step for multiple users in a single query.
914 #[tracing::instrument(skip_all)]
915 pub async fn batch_advance_onboarding_step(
916 pool: &PgPool,
917 user_ids: &[UserId],
918 new_step: i16,
919 ) -> Result<()> {
920 sqlx::query(
921 "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = ANY($1)",
922 )
923 .bind(user_ids)
924 .bind(new_step)
925 .execute(pool)
926 .await?;
927 Ok(())
928 }
929
930 /// Mark a user as a founder. Called when they start a creator-tier
931 /// subscription while the founder pricing window is open. Sticky; never
932 /// reset, even on cancellation. Subsequent re-subscriptions during the
933 /// window keep their founder status. After the window closes, eligibility
934 /// is determined by `founder_locked_at` (stamped only for users with an
935 /// active subscription at the close-time snapshot).
936 ///
937 /// **DIY exclusion**: DIY-tier accounts are not full members and must not
938 /// qualify for founder pricing (`project_founder_pricing.md` § decision 5).
939 /// Only call this from creator-tier (Basic/SmallFiles/BigFiles/Everything)
940 /// checkout paths. When DIY ships, its checkout path must NOT invoke this.
941 #[tracing::instrument(skip_all)]
942 pub async fn mark_user_as_founder(pool: &PgPool, user_id: UserId) -> Result<()> {
943 sqlx::query(
944 r#"
945 UPDATE users
946 SET is_founder = TRUE,
947 updated_at = NOW()
948 WHERE id = $1 AND is_founder = FALSE
949 "#,
950 )
951 .bind(user_id)
952 .execute(pool)
953 .await?;
954 Ok(())
955 }
956
957 /// Close the founder pricing window by stamping `founder_locked_at` on every
958 /// user who is currently flagged `is_founder` AND has an active creator-tier
959 /// subscription. Returns the number of users locked in. Idempotent: skips
960 /// any user already locked. Intended to be called once from an admin tool
961 /// at the moment the founder window closes.
962 #[tracing::instrument(skip_all)]
963 pub async fn lock_in_founders_with_active_subscriptions(pool: &PgPool) -> Result<u64> {
964 let result = sqlx::query(
965 r#"
966 UPDATE users u
967 SET founder_locked_at = NOW(),
968 updated_at = NOW()
969 WHERE u.is_founder = TRUE
970 AND u.founder_locked_at IS NULL
971 AND EXISTS (
972 SELECT 1 FROM creator_subscriptions s
973 WHERE s.user_id = u.id
974 AND s.status = 'active'
975 )
976 "#,
977 )
978 .execute(pool)
979 .await?;
980 Ok(result.rows_affected())
981 }
982
983 /// Update a user's Stripe Tax toggle.
984 #[tracing::instrument(skip_all)]
985 pub async fn update_stripe_tax_enabled(pool: &PgPool, user_id: UserId, enabled: bool) -> Result<()> {
986 sqlx::query(
987 r#"
988 UPDATE users
989 SET stripe_tax_enabled = $2,
990 updated_at = NOW()
991 WHERE id = $1
992 "#,
993 )
994 .bind(user_id)
995 .bind(enabled)
996 .execute(pool)
997 .await?;
998
999 Ok(())
1000 }
1001
1002 /// Disconnect a user's Stripe account
1003 #[tracing::instrument(skip_all)]
1004 pub async fn disconnect_user_stripe(pool: &PgPool, user_id: UserId) -> Result<DbUser> {
1005 let user = sqlx::query_as::<_, DbUser>(
1006 r#"
1007 UPDATE users
1008 SET stripe_account_id = NULL,
1009 stripe_onboarding_complete = false,
1010 stripe_payouts_enabled = false,
1011 stripe_charges_enabled = false,
1012 updated_at = NOW()
1013 WHERE id = $1
1014 RETURNING *
1015 "#,
1016 )
1017 .bind(user_id)
1018 .fetch_one(pool)
1019 .await?;
1020
1021 Ok(user)
1022 }
1023
1024 /// Fetch the current cache generation for a user (cheap, indexed lookup).
1025 #[tracing::instrument(skip_all)]
1026 pub async fn get_cache_generation(pool: &PgPool, user_id: UserId) -> Result<i64> {
1027 let generation = sqlx::query_scalar::<_, i64>(
1028 "SELECT cache_generation FROM users WHERE id = $1",
1029 )
1030 .bind(user_id)
1031 .fetch_one(pool)
1032 .await?;
1033
1034 Ok(generation)
1035 }
1036
1037 /// Atomically increment the user's cache generation counter.
1038 /// Call after any write that changes user-visible dashboard data.
1039 #[tracing::instrument(skip_all)]
1040 pub async fn bump_cache_generation(pool: &PgPool, user_id: UserId) -> Result<()> {
1041 sqlx::query("UPDATE users SET cache_generation = cache_generation + 1 WHERE id = $1")
1042 .bind(user_id)
1043 .execute(pool)
1044 .await?;
1045
1046 Ok(())
1047 }
1048
1049 /// Look up a verified user by email (case-insensitive).
1050 /// Returns the user ID if a verified account exists with that email.
1051 #[tracing::instrument(skip_all)]
1052 pub async fn get_verified_user_id_by_email(pool: &PgPool, email: &Email) -> Result<Option<UserId>> {
1053 let id = sqlx::query_scalar(
1054 "SELECT id FROM users WHERE LOWER(email) = LOWER($1) AND email_verified = true",
1055 )
1056 .bind(email)
1057 .fetch_optional(pool)
1058 .await?;
1059
1060 Ok(id)
1061 }
1062