Skip to main content

max / makenotwork

Split db/users.rs by what each query is for 1330 lines, 48 free functions against one table, in arrival order. Four late banner comments covered a quarter of them; the rest was chronology. Six siblings, each named for a reason a query exists: the account's arc, moderation, Stripe, notification preferences, onboarding, and the admin list. The core reads and writes stay in mod.rs. Every module is re-exported flat, so `crate::db::users::get_user_by_id` and every other call site is untouched. admin_queries is the one worth stating: `get_all_users` and `count_users` are two `match` ladders over the same filter set, and a filter added to one and not the other silently mispages the admin list. Their agreement was enforced by proximity, and sixty lines apart in a 1330-line file is not proximity. `delete_user` and the other restricted functions keep their `pub(crate)`; a glob re-export carries each item's own visibility through.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session
https://claude.ai/code/session_01EEmeiSJnmyL98QzA5Dwsvz
Author: Max Johnson <me@maxj.phd> · 2026-09-05 02:44 UTC
Signed with PGP, not checked
Commit: 2a58d2b2a71efed54ef2f954c6d33b7acb81fae3
Parent: 115ca79
8 files changed, +1393 insertions, -500 deletions
D server/src/db/users.rs -500
@@ -1,1330 +1,0 @@
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.
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 -
Lines truncated
A server/src/db/users/admin_queries.rs +131
@@ -1,0 +1,131 @@
1 + //! The admin user list and its counts.
2 + //!
3 + //! `get_all_users` and `count_users` are two `match` ladders over the same
4 + //! filter set, and a filter added to one and not the other silently mispages
5 + //! the list. Their agreement is enforced by sitting next to each other, which
6 + //! sixty lines apart in a 1330-line file was not.
7 +
8 + use sqlx::PgPool;
9 +
10 + use crate::db::models::DbUser;
11 + use crate::error::Result;
12 +
13 + /// Admin query: all users, optionally filtered by suspension status, with pagination.
14 + #[tracing::instrument(skip_all)]
15 + pub async fn get_all_users(
16 + pool: &PgPool,
17 + filter: Option<&str>,
18 + limit: i64,
19 + offset: i64,
20 + ) -> Result<Vec<DbUser>> {
21 + let limit = limit.min(200);
22 + let users = match filter {
23 + Some("suspended") => {
24 + sqlx::query_as::<_, DbUser>(
25 + "SELECT * FROM users WHERE suspended_at IS NOT NULL ORDER BY suspended_at DESC LIMIT $1 OFFSET $2",
26 + )
27 + .bind(limit)
28 + .bind(offset)
29 + .fetch_all(pool)
30 + .await?
31 + }
32 + Some("active") => {
33 + sqlx::query_as::<_, DbUser>(
34 + "SELECT * FROM users WHERE suspended_at IS NULL ORDER BY created_at DESC LIMIT $1 OFFSET $2",
35 + )
36 + .bind(limit)
37 + .bind(offset)
38 + .fetch_all(pool)
39 + .await?
40 + }
41 + // Creators with a custom page; most recently changed first, so
42 + // "recently changed" surfaces naturally at the top.
43 + Some("custom_pages") => {
44 + sqlx::query_as::<_, DbUser>(
45 + "SELECT * FROM users WHERE custom_html <> '' OR custom_css <> '' \
46 + ORDER BY custom_pages_updated_at DESC NULLS LAST, created_at DESC LIMIT $1 OFFSET $2",
47 + )
48 + .bind(limit)
49 + .bind(offset)
50 + .fetch_all(pool)
51 + .await?
52 + }
53 + Some("pages_locked") => {
54 + sqlx::query_as::<_, DbUser>(
55 + "SELECT * FROM users WHERE custom_pages_locked = true ORDER BY created_at DESC LIMIT $1 OFFSET $2",
56 + )
57 + .bind(limit)
58 + .bind(offset)
59 + .fetch_all(pool)
60 + .await?
61 + }
62 + _ => {
63 + sqlx::query_as::<_, DbUser>(
64 + "SELECT * FROM users ORDER BY created_at DESC LIMIT $1 OFFSET $2",
65 + )
66 + .bind(limit)
67 + .bind(offset)
68 + .fetch_all(pool)
69 + .await?
70 + }
71 + };
72 +
73 + Ok(users)
74 + }
75 +
76 + /// Count users matching a filter (for pagination totals).
77 + #[tracing::instrument(skip_all)]
78 + pub async fn count_users(pool: &PgPool, filter: Option<&str>) -> Result<i64> {
79 + let count = match filter {
80 + Some("suspended") => {
81 + sqlx::query_scalar::<_, i64>(
82 + "SELECT COUNT(*) FROM users WHERE suspended_at IS NOT NULL",
83 + )
84 + .fetch_one(pool)
85 + .await?
86 + }
87 + Some("active") => {
88 + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users WHERE suspended_at IS NULL")
89 + .fetch_one(pool)
90 + .await?
91 + }
92 + Some("custom_pages") => {
93 + sqlx::query_scalar::<_, i64>(
94 + "SELECT COUNT(*) FROM users WHERE custom_html <> '' OR custom_css <> ''",
95 + )
96 + .fetch_one(pool)
97 + .await?
98 + }
99 + Some("pages_locked") => {
100 + sqlx::query_scalar::<_, i64>(
101 + "SELECT COUNT(*) FROM users WHERE custom_pages_locked = true",
102 + )
103 + .fetch_one(pool)
104 + .await?
105 + }
106 + _ => {
107 + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM users")
108 + .fetch_one(pool)
109 + .await?
110 + }
111 + };
112 +
113 + Ok(count)
114 + }
115 +
116 + /// Count total and suspended users in a single query.
117 + #[tracing::instrument(skip_all)]
118 + pub async fn count_users_summary(pool: &PgPool) -> Result<(i64, i64)> {
119 + let (total, suspended): (i64, i64) = sqlx::query_as(
120 + r"
121 + SELECT
122 + COUNT(*),
123 + COUNT(*) FILTER (WHERE suspended_at IS NOT NULL)
124 + FROM users
125 + ",
126 + )
127 + .fetch_one(pool)
128 + .await?;
129 +
130 + Ok((total, suspended))
131 + }
A server/src/db/users/lifecycle.rs +236
@@ -1,0 +1,236 @@
1 + //! An account's arc after it exists: deactivate, terminate, delete, and the
2 + //! sandbox accounts that expire on their own.
3 +
4 + use sqlx::PgPool;
5 +
6 + use crate::db::UserId;
7 + use crate::db::models::DbUser;
8 + use crate::db::validated_types::{Email, Username};
9 + use crate::error::Result;
10 +
11 + /// Self-deactivate an account (enter limbo state).
12 + ///
13 + /// Bumps `jwt_invalidated_at` so any outstanding SyncKit JWTs minted from
14 + /// this account stop authenticating immediately.
15 + #[tracing::instrument(skip_all)]
16 + pub async fn deactivate_user(pool: &PgPool, id: UserId) -> Result<()> {
17 + sqlx::query(
18 + "UPDATE users SET deactivated_at = NOW(), jwt_invalidated_at = NOW(), updated_at = NOW() WHERE id = $1",
19 + )
20 + .bind(id)
21 + .execute(pool)
22 + .await?;
23 +
24 + Ok(())
25 + }
26 +
27 + /// Reactivate a self-deactivated account.
28 + #[tracing::instrument(skip_all)]
29 + pub async fn reactivate_user(pool: &PgPool, id: UserId) -> Result<()> {
30 + sqlx::query("UPDATE users SET deactivated_at = NULL, updated_at = NOW() WHERE id = $1")
31 + .bind(id)
32 + .execute(pool)
33 + .await?;
34 +
35 + Ok(())
36 + }
37 +
38 + /// Admin: permanently terminate an account (enforcement ladder step 4).
39 + /// The user has 30 days to export data. After that, the scheduler deletes the account.
40 + /// The account must already be suspended.
41 + #[tracing::instrument(skip_all)]
42 + pub async fn terminate_user(pool: &PgPool, id: UserId) -> Result<()> {
43 + sqlx::query(
44 + "UPDATE users SET terminated_at = NOW(), jwt_invalidated_at = NOW(), updated_at = NOW() WHERE id = $1",
45 + )
46 + .bind(id)
47 + .execute(pool)
48 + .await?;
49 +
50 + Ok(())
51 + }
52 +
53 + /// Get user IDs of terminated accounts whose 30-day export window has expired.
54 + #[tracing::instrument(skip_all)]
55 + pub async fn get_expired_terminated_ids(pool: &PgPool) -> Result<Vec<UserId>> {
56 + let ids: Vec<UserId> = sqlx::query_scalar(
57 + r"
58 + SELECT id FROM users
59 + WHERE terminated_at IS NOT NULL
60 + AND terminated_at < NOW() - INTERVAL '30 days'
61 + ORDER BY terminated_at
62 + LIMIT 1000
63 + ",
64 + )
65 + .fetch_all(pool)
66 + .await?;
67 +
68 + Ok(ids)
69 + }
70 +
71 + /// Permanently delete a user by ID.
72 + ///
73 + /// `pub(crate)` and not for direct handler use: go through
74 + /// [`crate::AppState::delete_user_account`], which also purges the in-memory
75 + /// caches keyed to the user (domain_cache). Deleting here alone would leave a
76 + /// stale, never-revalidated cache entry.
77 + #[tracing::instrument(skip_all)]
78 + pub(crate) async fn delete_user(pool: &PgPool, id: UserId) -> Result<()> {
79 + sqlx::query("DELETE FROM users WHERE id = $1")
80 + .bind(id)
81 + .execute(pool)
82 + .await?;
83 +
84 + Ok(())
85 + }
86 +
87 + /// Check whether this creator has any completed sales (transactions where they were the seller).
88 + #[tracing::instrument(skip_all)]
89 + pub async fn has_completed_sales(pool: &PgPool, id: UserId) -> Result<bool> {
90 + let count: i64 = sqlx::query_scalar(
91 + "SELECT COUNT(*) FROM transactions WHERE seller_id = $1 AND status = 'completed'",
92 + )
93 + .bind(id)
94 + .fetch_one(pool)
95 + .await?;
96 +
97 + Ok(count > 0)
98 + }
99 +
100 + /// Schedule content removal 90 days from now. The user row is hidden from public
101 + /// views but items remain accessible to buyers who previously purchased them.
102 + /// After 90 days the scheduler deletes S3 objects and the user row.
103 + #[tracing::instrument(skip_all)]
104 + pub async fn schedule_content_removal(pool: &PgPool, id: UserId) -> Result<()> {
105 + sqlx::query(
106 + r"
107 + UPDATE users
108 + SET content_removal_at = NOW() + INTERVAL '90 days',
109 + deactivated_at = NOW(),
110 + updated_at = NOW()
111 + WHERE id = $1
112 + ",
113 + )
114 + .bind(id)
115 + .execute(pool)
116 + .await?;
117 +
118 + Ok(())
119 + }
120 +
121 + /// Get user IDs whose 90-day content removal grace period has expired.
122 + #[tracing::instrument(skip_all)]
123 + pub async fn get_expired_content_removal_ids(pool: &PgPool) -> Result<Vec<UserId>> {
124 + let ids: Vec<UserId> = sqlx::query_scalar(
125 + r"
126 + SELECT id FROM users
127 + WHERE content_removal_at IS NOT NULL
128 + AND content_removal_at < NOW()
129 + ORDER BY content_removal_at
130 + LIMIT 1000
131 + ",
132 + )
133 + .fetch_all(pool)
134 + .await?;
135 +
136 + Ok(ids)
137 + }
138 +
139 + /// Create an ephemeral sandbox user. Returns the created row.
140 + ///
141 + /// The user gets `can_create_projects = true`, `email_verified = true`,
142 + /// a SmallFiles creator tier, and a tight storage cap. The row is
143 + /// automatically cleaned up by the scheduler after `sandbox_expires_at`.
144 + #[tracing::instrument(skip_all)]
145 + pub async fn create_sandbox_user(
146 + pool: &PgPool,
147 + username: &Username,
148 + email: &Email,
149 + password_hash: &str,
150 + expiry_secs: i64,
151 + ) -> Result<DbUser> {
152 + let user = sqlx::query_as::<_, DbUser>(
153 + r"
154 + INSERT INTO users (
155 + username, email, password_hash,
156 + is_sandbox, sandbox_expires_at,
157 + can_create_projects, email_verified,
158 + creator_tier
159 + )
160 + VALUES (
161 + $1, $2, $3,
162 + TRUE, NOW() + make_interval(secs => $4::float8),
163 + TRUE, TRUE,
164 + 'small_files'
165 + )
166 + RETURNING *
167 + ",
168 + )
169 + .bind(username)
170 + .bind(email)
171 + .bind(password_hash)
172 + .bind(expiry_secs as f64)
173 + .fetch_one(pool)
174 + .await?;
175 +
176 + Ok(user)
177 + }
178 +
179 + /// Return IDs of sandbox users whose expiry has passed.
180 + #[tracing::instrument(skip_all)]
181 + pub async fn get_expired_sandbox_ids(pool: &PgPool) -> Result<Vec<UserId>> {
182 + // Per-tick LIMIT bounds the supervisor's input list (the scheduler re-ticks
183 + // and the WHERE re-excludes already-deleted rows, so the remainder is picked
184 + // up next tick). Run #12 INFO, keeps a pathological mass-expiry from loading
185 + // an unbounded id vec even though concurrency is already capped at 4.
186 + let ids = sqlx::query_scalar::<_, UserId>(
187 + "SELECT id FROM users WHERE is_sandbox = TRUE AND sandbox_expires_at < NOW() \
188 + ORDER BY sandbox_expires_at LIMIT 1000",
189 + )
190 + .fetch_all(pool)
191 + .await?;
192 +
193 + Ok(ids)
194 + }
195 +
196 + /// Count active (non-expired) sandbox accounts created from a given IP.
197 + /// Used to enforce the per-IP concurrent sandbox cap.
198 + #[tracing::instrument(skip_all)]
199 + pub async fn count_active_sandboxes_by_ip(pool: &PgPool, ip: &str) -> Result<i64> {
200 + let count: i64 = sqlx::query_scalar(
201 + r"
202 + SELECT COUNT(*) FROM users u
203 + JOIN user_sessions us ON us.user_id = u.id
204 + WHERE u.is_sandbox = TRUE
205 + AND u.sandbox_expires_at > NOW()
206 + AND us.ip_address = $1
207 + ",
208 + )
209 + .bind(ip)
210 + .fetch_one(pool)
211 + .await?;
212 +
213 + Ok(count)
214 + }
215 +
216 + /// Set the creator_paused_at timestamp (voluntary pause).
217 + #[tracing::instrument(skip_all)]
218 + pub async fn pause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
219 + sqlx::query("UPDATE users SET creator_paused_at = NOW(), updated_at = NOW() WHERE id = $1")
220 + .bind(user_id)
221 + .execute(pool)
222 + .await?;
223 +
224 + Ok(())
225 + }
226 +
227 + /// Clear the creator_paused_at timestamp (resume from voluntary pause).
228 + #[tracing::instrument(skip_all)]
229 + pub async fn unpause_creator(pool: &PgPool, user_id: UserId) -> Result<()> {
230 + sqlx::query("UPDATE users SET creator_paused_at = NULL, updated_at = NOW() WHERE id = $1")
231 + .bind(user_id)
232 + .execute(pool)
233 + .await?;
234 +
235 + Ok(())
236 + }
A server/src/db/users/mod.rs +305
@@ -1,0 +1,305 @@
1 + //! User account CRUD, profile updates, and lookup queries.
2 + //!
3 + //! The core reads and writes are here; everything with a life of its own is a
4 + //! sibling.
5 +
6 + mod admin_queries;
7 + mod lifecycle;
8 + mod moderation;
9 + mod onboarding;
10 + mod preferences;
11 + mod stripe;
12 +
13 + pub use admin_queries::*;
14 + pub use lifecycle::*;
15 + pub use moderation::*;
16 + pub use onboarding::*;
17 + pub use preferences::*;
18 + pub use stripe::*;
19 +
20 + use sqlx::PgPool;
21 +
22 + use crate::db::UserId;
23 + use crate::db::models::DbUser;
24 + use crate::db::validated_types::{Email, Username};
25 + use crate::error::Result;
26 +
27 + /// Insert a new user and return the created row.
28 + #[tracing::instrument(skip_all)]
29 + pub async fn create_user(
30 + pool: &PgPool,
31 + username: &Username,
32 + email: &Email,
33 + password_hash: &str,
34 + ) -> Result<DbUser> {
35 + let user = sqlx::query_as::<_, DbUser>(
36 + r"
37 + INSERT INTO users (username, email, password_hash)
38 + VALUES ($1, $2, $3)
39 + RETURNING *
40 + ",
41 + )
42 + .bind(username)
43 + .bind(email)
44 + .bind(password_hash)
45 + .fetch_one(pool)
46 + .await?;
47 +
48 + Ok(user)
49 + }
50 +
51 + /// Insert a durable example-marketplace creator (see [`crate::seed`]).
52 + ///
53 + /// Unlike [`create_sandbox_user`], this leaves `is_sandbox` at its `FALSE`
54 + /// default so the account and its projects appear on every public surface
55 + /// (discover/browse/search gate only on `is_sandbox = FALSE`). It grants
56 + /// `can_create_projects`, marks the email verified, and pins the top
57 + /// `creator_tier` so no capability gate blocks the item spread seeded in later
58 + /// phases. Only ever called by the `--seed-examples` flow, which is itself
59 + /// confined to testnot/localhost by [`crate::seed::run`]'s guards.
60 + #[tracing::instrument(skip_all)]
61 + pub async fn create_example_creator(
62 + pool: &PgPool,
63 + username: &Username,
64 + email: &Email,
65 + password_hash: &str,
66 + ) -> Result<DbUser> {
67 + let user = sqlx::query_as::<_, DbUser>(
68 + r"
69 + INSERT INTO users (
70 + username, email, password_hash,
71 + can_create_projects, email_verified, creator_tier
72 + )
73 + VALUES ($1, $2, $3, TRUE, TRUE, 'everything')
74 + RETURNING *
75 + ",
76 + )
77 + .bind(username)
78 + .bind(email)
79 + .bind(password_hash)
80 + .fetch_one(pool)
81 + .await?;
82 +
83 + Ok(user)
84 + }
85 +
86 + /// Fetch a user by primary key. Returns `None` if not found.
87 + #[tracing::instrument(skip_all)]
88 + pub async fn get_user_by_id(pool: &PgPool, id: UserId) -> Result<Option<DbUser>> {
89 + let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE id = $1")
90 + .bind(id)
91 + .fetch_optional(pool)
92 + .await?;
93 +
94 + Ok(user)
95 + }
96 +
97 + /// Fetch multiple users by ID in a single query.
98 + #[tracing::instrument(skip_all)]
99 + pub async fn get_users_by_ids(pool: &PgPool, ids: &[UserId]) -> Result<Vec<DbUser>> {
100 + let users = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE id = ANY($1)")
101 + .bind(ids)
102 + .fetch_all(pool)
103 + .await?;
104 + Ok(users)
105 + }
106 +
107 + /// Fetch a user by username. Returns `None` if not found.
108 + #[tracing::instrument(skip_all)]
109 + pub async fn get_user_by_username(pool: &PgPool, username: &Username) -> Result<Option<DbUser>> {
110 + let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE username = $1")
111 + .bind(username)
112 + .fetch_optional(pool)
113 + .await?;
114 +
115 + Ok(user)
116 + }
117 +
118 + /// Fetch a user by email address. Returns `None` if not found.
119 + #[tracing::instrument(skip_all)]
120 + pub async fn get_user_by_email(pool: &PgPool, email: &Email) -> Result<Option<DbUser>> {
121 + let user = sqlx::query_as::<_, DbUser>("SELECT * FROM users WHERE email = $1")
122 + .bind(email)
123 + .fetch_optional(pool)
124 + .await?;
125 +
126 + Ok(user)
127 + }
128 +
129 + /// Look up a verified user by email (case-insensitive).
130 + /// Returns the user ID if a verified account exists with that email.
131 + #[tracing::instrument(skip_all)]
132 + pub async fn get_verified_user_id_by_email(pool: &PgPool, email: &Email) -> Result<Option<UserId>> {
133 + let id = sqlx::query_scalar(
134 + "SELECT id FROM users WHERE LOWER(email) = LOWER($1) AND email_verified = true",
135 + )
136 + .bind(email)
137 + .fetch_optional(pool)
138 + .await?;
139 +
140 + Ok(id)
141 + }
142 +
143 + /// Update a user's display name and/or bio (COALESCE keeps existing values when `None`).
144 + #[tracing::instrument(skip_all)]
145 + pub async fn update_user_profile(
146 + pool: &PgPool,
147 + id: UserId,
148 + display_name: Option<&str>,
149 + bio: Option<&str>,
150 + ) -> Result<DbUser> {
151 + let user = sqlx::query_as::<_, DbUser>(
152 + r"
153 + UPDATE users
154 + SET display_name = COALESCE($2, display_name),
155 + bio = COALESCE($3, bio)
156 + WHERE id = $1
157 + RETURNING *
158 + ",
159 + )
160 + .bind(id)
161 + .bind(display_name)
162 + .bind(bio)
163 + .fetch_one(pool)
164 + .await?;
165 +
166 + Ok(user)
167 + }
168 +
169 + /// Store a user's custom profile-page source (the original, pre-sanitization),
170 + /// stamp `custom_pages_updated_at`, and bump the cache generation. The source is
171 + /// re-sanitized on render; see [`crate::custom_pages`].
172 + #[tracing::instrument(skip_all, fields(user_id = %id))]
173 + pub async fn update_user_custom_page<'e>(
174 + executor: impl sqlx::PgExecutor<'e>,
175 + id: UserId,
176 + custom_html: &str,
177 + custom_css: &str,
178 + ) -> Result<DbUser> {
179 + let user = sqlx::query_as::<_, DbUser>(
180 + r"
181 + UPDATE users
182 + SET custom_html = $2,
183 + custom_css = $3,
184 + custom_pages_updated_at = now(),
185 + cache_generation = cache_generation + 1
186 + WHERE id = $1
187 + RETURNING *
188 + ",
189 + )
190 + .bind(id)
191 + .bind(custom_html)
192 + .bind(custom_css)
193 + .fetch_one(executor)
194 + .await?;
195 + Ok(user)
196 + }
197 +
198 + /// Clear a user's custom profile page back to the platform default.
199 + pub async fn reset_user_custom_page(pool: &PgPool, id: UserId) -> Result<()> {
200 + sqlx::query(
201 + "UPDATE users SET custom_html = '', custom_css = '', \
202 + custom_pages_updated_at = NULL, cache_generation = cache_generation + 1 WHERE id = $1",
203 + )
204 + .bind(id)
205 + .execute(pool)
206 + .await?;
207 + Ok(())
208 + }
209 +
210 + /// Set or clear a user's creator theme for their public profile. `None` clears
211 + /// to the platform default. The id is validated against the embedded registry
212 + /// before this call.
213 + #[tracing::instrument(skip_all)]
214 + pub async fn update_user_theme(pool: &PgPool, id: UserId, theme_id: Option<&str>) -> Result<()> {
215 + sqlx::query("UPDATE users SET theme_id = $2, updated_at = NOW() WHERE id = $1")
216 + .bind(id)
217 + .bind(theme_id)
218 + .execute(pool)
219 + .await?;
220 +
221 + Ok(())
222 + }
223 +
224 + /// Set a user's SSH console theme. Takes a `makeover::ThemeSelection` string
225 + /// (a bundled theme id, or `"system"`), validated before this call.
226 + #[tracing::instrument(skip_all)]
227 + pub async fn update_user_console_theme(pool: &PgPool, id: UserId, selection: &str) -> Result<()> {
228 + sqlx::query("UPDATE users SET console_theme = $2, updated_at = NOW() WHERE id = $1")
229 + .bind(id)
230 + .bind(selection)
231 + .execute(pool)
232 + .await?;
233 +
234 + Ok(())
235 + }
236 +
237 + /// Replace a user's password hash and invalidate outstanding JWTs.
238 + #[tracing::instrument(skip_all)]
239 + pub async fn update_user_password(pool: &PgPool, id: UserId, password_hash: &str) -> Result<()> {
240 + sqlx::query("UPDATE users SET password_hash = $2, jwt_invalidated_at = NOW() WHERE id = $1")
241 + .bind(id)
242 + .bind(password_hash)
243 + .execute(pool)
244 + .await?;
245 +
246 + Ok(())
247 + }
248 +
249 + /// Increment the user's feed key version, revoking their current personal-feed
250 + /// URL. Returns the new version (folded into the next URL's HMAC).
251 + #[tracing::instrument(skip_all)]
252 + pub async fn bump_feed_key_version(pool: &PgPool, id: UserId) -> Result<i32> {
253 + let (version,): (i32,) = sqlx::query_as(
254 + "UPDATE users SET feed_key_version = feed_key_version + 1, updated_at = NOW() \
255 + WHERE id = $1 RETURNING feed_key_version",
256 + )
257 + .bind(id)
258 + .fetch_one(pool)
259 + .await?;
260 +
261 + Ok(version)
262 + }
263 +
264 + /// Mark a user's email as verified
265 + #[tracing::instrument(skip_all)]
266 + pub async fn verify_user_email(pool: &PgPool, user_id: UserId) -> Result<()> {
267 + sqlx::query(
268 + r"
269 + UPDATE users
270 + SET email_verified = true,
271 + email_verification_token = NULL,
272 + updated_at = NOW()
273 + WHERE id = $1
274 + ",
275 + )
276 + .bind(user_id)
277 + .execute(pool)
278 + .await?;
279 +
280 + Ok(())
281 + }
282 +
283 + /// Fetch the current cache generation for a user (cheap, indexed lookup).
284 + #[tracing::instrument(skip_all)]
285 + pub async fn get_cache_generation(pool: &PgPool, user_id: UserId) -> Result<i64> {
286 + let generation =
287 + sqlx::query_scalar::<_, i64>("SELECT cache_generation FROM users WHERE id = $1")
288 + .bind(user_id)
289 + .fetch_one(pool)
290 + .await?;
291 +
292 + Ok(generation)
293 + }
294 +
295 + /// Atomically increment the user's cache generation counter.
296 + /// Call after any write that changes user-visible dashboard data.
297 + #[tracing::instrument(skip_all)]
298 + pub async fn bump_cache_generation(pool: &PgPool, user_id: UserId) -> Result<()> {
299 + sqlx::query("UPDATE users SET cache_generation = cache_generation + 1 WHERE id = $1")
300 + .bind(user_id)
301 + .execute(pool)
302 + .await?;
303 +
304 + Ok(())
305 + }
A server/src/db/users/moderation.rs +194
@@ -1,0 +1,194 @@
1 + //! Acting on an account: suspension, the appeal against it, and the two trust
2 + //! flags a moderator can set.
3 +
4 + use sqlx::PgPool;
5 +
6 + use crate::db::UserId;
7 + use crate::db::enums::AppealDecision;
8 + use crate::db::models::DbUser;
9 + use crate::error::Result;
10 +
11 + /// Suspend a user account, clearing any prior appeal fields.
12 + ///
13 + /// Bumps `jwt_invalidated_at` so SyncKit JWTs minted for this user expire
14 + /// at the next extractor check (subject to `SESSION_TOUCH_CACHE_SECS`).
15 + #[tracing::instrument(skip_all)]
16 + pub async fn suspend_user(pool: &PgPool, user_id: UserId, reason: &str) -> Result<()> {
17 + sqlx::query(
18 + r"
19 + UPDATE users
20 + SET suspended_at = NOW(),
21 + suspension_reason = $2,
22 + jwt_invalidated_at = NOW(),
23 + appeal_text = NULL,
24 + appeal_submitted_at = NULL,
25 + appeal_decision = NULL,
26 + appeal_response = NULL,
27 + appeal_decided_at = NULL,
28 + updated_at = NOW()
29 + WHERE id = $1
30 + ",
31 + )
32 + .bind(user_id)
33 + .bind(reason)
34 + .execute(pool)
35 + .await?;
36 +
37 + Ok(())
38 + }
39 +
40 + /// Remove suspension and clear all suspension/appeal fields.
41 + #[tracing::instrument(skip_all)]
42 + pub async fn unsuspend_user(pool: &PgPool, user_id: UserId) -> Result<()> {
43 + sqlx::query(
44 + r"
45 + UPDATE users
46 + SET suspended_at = NULL,
47 + suspension_reason = NULL,
48 + appeal_text = NULL,
49 + appeal_submitted_at = NULL,
50 + appeal_decision = NULL,
51 + appeal_response = NULL,
52 + appeal_decided_at = NULL,
53 + updated_at = NOW()
54 + WHERE id = $1
55 + ",
56 + )
57 + .bind(user_id)
58 + .execute(pool)
59 + .await?;
60 +
61 + Ok(())
62 + }
63 +
64 + /// Submit an appeal for a suspended account, clearing any prior decision.
65 + #[tracing::instrument(skip_all)]
66 + pub async fn submit_appeal(pool: &PgPool, user_id: UserId, appeal_text: &str) -> Result<()> {
67 + sqlx::query(
68 + r"
69 + UPDATE users
70 + SET appeal_text = $2,
71 + appeal_submitted_at = NOW(),
72 + appeal_decision = NULL,
73 + appeal_response = NULL,
74 + appeal_decided_at = NULL,
75 + updated_at = NOW()
76 + WHERE id = $1 AND suspended_at IS NOT NULL
77 + ",
78 + )
79 + .bind(user_id)
80 + .bind(appeal_text)
81 + .execute(pool)
82 + .await?;
83 +
84 + Ok(())
85 + }
86 +
87 + /// Resolve an appeal. If approved, also clears suspension.
88 + #[tracing::instrument(skip_all)]
89 + pub async fn resolve_appeal(
90 + pool: &PgPool,
91 + user_id: UserId,
92 + decision: AppealDecision,
93 + response: &str,
94 + ) -> Result<()> {
95 + if decision == AppealDecision::Approved {
96 + // Approve: clear suspension entirely
97 + sqlx::query(
98 + r"
99 + UPDATE users
100 + SET appeal_decision = $2,
101 + appeal_response = $3,
102 + appeal_decided_at = NOW(),
103 + suspended_at = NULL,
104 + suspension_reason = NULL,
105 + updated_at = NOW()
106 + WHERE id = $1
107 + ",
108 + )
109 + .bind(user_id)
110 + .bind(decision)
111 + .bind(response)
112 + .execute(pool)
113 + .await?;
114 + } else {
115 + // Deny: keep suspension, record decision
116 + sqlx::query(
117 + r"
118 + UPDATE users
119 + SET appeal_decision = $2,
120 + appeal_response = $3,
121 + appeal_decided_at = NOW(),
122 + updated_at = NOW()
123 + WHERE id = $1
124 + ",
125 + )
126 + .bind(user_id)
127 + .bind(decision)
128 + .bind(response)
129 + .execute(pool)
130 + .await?;
131 + }
132 +
133 + Ok(())
134 + }
135 +
136 + /// Admin query: users with a pending appeal (submitted but not yet decided).
137 + #[tracing::instrument(skip_all)]
138 + pub async fn get_pending_appeals(pool: &PgPool) -> Result<Vec<DbUser>> {
139 + let users = sqlx::query_as::<_, DbUser>(
140 + r"
141 + SELECT * FROM users
142 + WHERE appeal_submitted_at IS NOT NULL
143 + AND appeal_decided_at IS NULL
144 + ORDER BY appeal_submitted_at ASC
145 + LIMIT 500
146 + ",
147 + )
148 + .fetch_all(pool)
149 + .await?;
150 +
151 + Ok(users)
152 + }
153 +
154 + /// Check if a user is trusted for uploads (bypasses review queue).
155 + #[tracing::instrument(skip_all)]
156 + pub async fn is_upload_trusted(pool: &PgPool, user_id: UserId) -> Result<bool> {
157 + let trusted = sqlx::query_scalar::<_, bool>("SELECT upload_trusted FROM users WHERE id = $1")
158 + .bind(user_id)
159 + .fetch_one(pool)
160 + .await?;
161 +
162 + Ok(trusted)
163 + }
164 +
165 + /// Set a user's upload trust status.
166 + #[tracing::instrument(skip_all)]
167 + pub async fn set_upload_trusted(pool: &PgPool, user_id: UserId, trusted: bool) -> Result<()> {
168 + sqlx::query(
169 + r"
170 + UPDATE users
171 + SET upload_trusted = $2,
172 + updated_at = NOW()
173 + WHERE id = $1
174 + ",
175 + )
176 + .bind(user_id)
177 + .bind(trusted)
178 + .execute(pool)
179 + .await?;
180 +
181 + Ok(())
182 + }
183 +
184 + /// Moderation kill switch for custom pages. While locked, the creator can't
185 + /// edit their custom pages and the live pages render the platform default.
186 + /// Reversible: unlocking restores the (preserved) custom source.
187 + pub async fn set_custom_pages_locked(pool: &PgPool, user_id: UserId, locked: bool) -> Result<()> {
188 + sqlx::query("UPDATE users SET custom_pages_locked = $2, cache_generation = cache_generation + 1 WHERE id = $1")
189 + .bind(user_id)
190 + .bind(locked)
191 + .execute(pool)
192 + .await?;
193 + Ok(())
194 + }
A server/src/db/users/onboarding.rs +66
@@ -1,0 +1,66 @@
1 + //! Walking a new creator through the first-run steps.
2 +
3 + use sqlx::PgPool;
4 +
5 + use crate::db::UserId;
6 + use crate::db::models::DbUser;
7 + use crate::error::Result;
8 +
9 + /// Users who need the next onboarding email. Returns users at a given step
10 + /// whose last email was sent more than `min_age` ago (or never).
11 + #[tracing::instrument(skip_all)]
12 + pub async fn get_onboarding_candidates(
13 + pool: &PgPool,
14 + step: i16,
15 + min_age: chrono::Duration,
16 + ) -> Result<Vec<DbUser>> {
17 + let cutoff = chrono::Utc::now() - min_age;
18 + // Per-tick LIMIT bounds the scheduler's input list. The caller advances
19 + // each returned user's step (so the WHERE re-excludes them), meaning the
20 + // remainder is drained on the next tick, same re-tick pattern as the
21 + // sandbox/terminated/content-removal cleanup queries. Run #14 MEDIUM: a
22 + // signup surge must not load an unbounded user vec into the lock-held tick.
23 + let users = sqlx::query_as::<_, DbUser>(
24 + "SELECT * FROM users
25 + WHERE onboarding_email_step = $1
26 + AND (onboarding_email_sent_at IS NULL OR onboarding_email_sent_at < $2)
27 + AND suspended_at IS NULL
28 + ORDER BY onboarding_email_sent_at ASC NULLS FIRST
29 + LIMIT 1000",
30 + )
31 + .bind(step)
32 + .bind(cutoff)
33 + .fetch_all(pool)
34 + .await?;
35 + Ok(users)
36 + }
37 +
38 + /// Advance a user's onboarding email step and record the send time.
39 + #[tracing::instrument(skip_all)]
40 + pub async fn advance_onboarding_step(pool: &PgPool, user_id: UserId, new_step: i16) -> Result<()> {
41 + sqlx::query(
42 + "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = $1",
43 + )
44 + .bind(user_id)
45 + .bind(new_step)
46 + .execute(pool)
47 + .await?;
48 + Ok(())
49 + }
50 +
51 + /// Advance onboarding step for multiple users in a single query.
52 + #[tracing::instrument(skip_all)]
53 + pub async fn batch_advance_onboarding_step(
54 + pool: &PgPool,
55 + user_ids: &[UserId],
56 + new_step: i16,
57 + ) -> Result<()> {
58 + sqlx::query(
59 + "UPDATE users SET onboarding_email_step = $2, onboarding_email_sent_at = NOW() WHERE id = ANY($1)",
60 + )
61 + .bind(user_ids)
62 + .bind(new_step)
63 + .execute(pool)
64 + .await?;
65 + Ok(())
66 + }
A server/src/db/users/preferences.rs +205
@@ -1,0 +1,205 @@
1 + //! What a user asked to be told about.
2 + //!
3 + //! These forward to `crate::db::lists::sync_notification_subscription` rather
4 + //! than writing `users` directly. Grouping them puts that boundary
5 + //! (migration 189) somewhere a reader can see it.
6 +
7 + use sqlx::PgPool;
8 +
9 + use crate::db::UserId;
10 + use crate::db::validated_types::Email;
11 + use crate::error::Result;
12 +
13 + /// Get all user emails for bulk notifications (e.g. shutdown notice).
14 + ///
15 + /// Capped to bound memory on a full-table scan; if the cap is ever hit the WARN
16 + /// fires so we know to switch to a paged dispatch model (mirrors
17 + /// [`get_status_alert_subscribers`]).
18 + #[tracing::instrument(skip_all)]
19 + pub async fn get_all_user_emails(pool: &PgPool) -> Result<Vec<(String, Option<String>)>> {
20 + const ALL_EMAILS_CAP: i64 = 50_000;
21 + let rows = sqlx::query_as::<_, (String, Option<String>)>(
22 + "SELECT email, display_name FROM users ORDER BY created_at ASC LIMIT $1",
23 + )
24 + .bind(ALL_EMAILS_CAP)
25 + .fetch_all(pool)
26 + .await?;
27 +
28 + if rows.len() as i64 == ALL_EMAILS_CAP {
29 + tracing::warn!(
30 + cap = ALL_EMAILS_CAP,
31 + "get_all_user_emails hit its cap; some users omitted, switch to paged dispatch"
32 + );
33 + }
34 +
35 + Ok(rows)
36 + }
37 +
38 + /// Update a user's email notification preferences.
39 + ///
40 + /// Writes subscriptions only. The `users.notify_*` columns these used to set
41 + /// are gone (migration 189); `db::lists` is the single record, and the
42 + /// preferences page and the unsubscribe links write the same rows.
43 + #[derive(Debug, Clone, Copy)]
44 + pub struct NotificationPreferences {
45 + pub notify_sale: bool,
46 + pub notify_follower: bool,
47 + pub notify_release: bool,
48 + pub login_notification_enabled: bool,
49 + pub notify_issues: bool,
50 + pub notify_status: bool,
51 + pub notify_invite: bool,
52 + }
53 +
54 + #[tracing::instrument(skip_all)]
55 + pub async fn update_notification_preferences(
56 + pool: &PgPool,
57 + id: UserId,
58 + prefs: NotificationPreferences,
59 + ) -> Result<()> {
60 + let NotificationPreferences {
61 + notify_sale,
62 + notify_follower,
63 + notify_release,
64 + login_notification_enabled,
65 + notify_issues,
66 + notify_status,
67 + notify_invite,
68 + } = prefs;
69 +
70 + for (kind, enabled) in [
71 + ("sale", notify_sale),
72 + ("follower", notify_follower),
73 + ("releases", notify_release),
74 + ("issues", notify_issues),
75 + ("status", notify_status),
76 + ("login", login_notification_enabled),
77 + ("invite", notify_invite),
78 + ] {
79 + crate::db::lists::sync_notification_subscription(pool, id, kind, enabled).await?;
80 + }
81 +
82 + Ok(())
83 + }
84 +
85 + /// Update tip settings.
86 + ///
87 + /// `tips_enabled` is a capability (whether the creator accepts tips at all) and
88 + /// stays a column. `notify_tip` is a notification preference and now lives in
89 + /// subscriptions with the other six, so the two are written to different
90 + /// places despite arriving from the same form.
91 + #[tracing::instrument(skip_all)]
92 + pub async fn update_tip_preferences(
93 + pool: &PgPool,
94 + id: UserId,
95 + tips_enabled: bool,
96 + notify_tip: bool,
97 + ) -> Result<()> {
98 + sqlx::query("UPDATE users SET tips_enabled = $2, updated_at = NOW() WHERE id = $1")
99 + .bind(id)
100 + .bind(tips_enabled)
101 + .execute(pool)
102 + .await?;
103 +
104 + crate::db::lists::sync_notification_subscription(pool, id, "tip", notify_tip).await?;
105 + Ok(())
106 + }
107 +
108 + /// Turn one notification off, by the name the unsubscribe link carries.
109 + ///
110 + /// For the seven original preferences that name is the old `users.notify_*`
111 + /// column, because those names are baked into signed URLs already sitting in
112 + /// inboxes; they map back to list kinds here rather than being renamed, which
113 + /// would invalidate every link ever sent.
114 + ///
115 + /// A kind with no legacy column (`invite`, migration 195) carries its kind name
116 + /// instead. Nothing older is in an inbox to be broken, so there is no column
117 + /// name to preserve and inventing one would be cargo cult.
118 + #[tracing::instrument(skip_all)]
119 + pub async fn disable_notification(
120 + pool: &PgPool,
121 + user_id: UserId,
122 + preference: &str,
123 + ) -> Result<bool> {
124 + let legacy: Option<&str> = crate::db::lists::NOTIFICATION_LISTS
125 + .iter()
126 + .find(|(_, legacy)| *legacy == preference)
127 + .map(|(kind, _)| *kind);
128 + let Some(kind) = legacy.or_else(|| {
129 + preference
130 + .parse::<crate::db::ListKind>()
131 + .ok()
132 + .map(|_| preference)
133 + }) else {
134 + return Ok(false);
135 + };
136 + crate::db::lists::sync_notification_subscription(pool, user_id, kind, false).await?;
137 + Ok(true)
138 + }
139 +
140 + /// A user who opted into platform status notifications.
141 + #[derive(sqlx::FromRow)]
142 + pub struct StatusAlertSubscriber {
143 + pub id: UserId,
144 + pub email: Email,
145 + pub display_name: Option<String>,
146 + }
147 +
148 + /// Get all users who opted into platform status notifications.
149 + ///
150 + /// Hard cap at 10k rows so the monitor's status-change fan-out can't unbox
151 + /// an unbounded query into RAM. The 100ms pacing in `monitor.rs` already
152 + /// limits fan-out throughput to ~600/minute, anything past 10k would
153 + /// chew through Postmark rate limits anyway. If we ever hit the cap a
154 + /// WARN fires so we know to switch to a paged dispatch model.
155 + #[tracing::instrument(skip_all)]
156 + pub async fn get_status_alert_subscribers(pool: &PgPool) -> Result<Vec<StatusAlertSubscriber>> {
157 + const STATUS_SUBSCRIBER_CAP: i64 = 10_000;
158 + let rows = sqlx::query_as::<_, StatusAlertSubscriber>(
159 + "SELECT u.id, u.email, u.display_name FROM users u \
160 + JOIN list_subscriptions ls ON ls.user_id = u.id \
161 + JOIN lists l ON l.id = ls.list_id AND l.scope = 'platform' AND l.kind = 'status' \
162 + WHERE ls.state IN ('confirmed', 'imported') AND u.deactivated_at IS NULL \
163 + ORDER BY u.id LIMIT $1",
164 + )
165 + .bind(STATUS_SUBSCRIBER_CAP)
166 + .fetch_all(pool)
167 + .await?;
168 + if rows.len() as i64 == STATUS_SUBSCRIBER_CAP {
169 + tracing::warn!(
170 + cap = STATUS_SUBSCRIBER_CAP,
171 + "get_status_alert_subscribers hit hard cap; promote to paged dispatch"
172 + );
173 + }
174 + Ok(rows)
175 + }
176 +
177 + /// Atomically check-and-set broadcast timestamp. Returns false if already sent within 24 hours.
178 + #[tracing::instrument(skip_all)]
179 + pub async fn try_set_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<bool> {
180 + let result = sqlx::query(
181 + r"
182 + UPDATE users
183 + SET last_broadcast_at = NOW()
184 + WHERE id = $1
185 + AND (last_broadcast_at IS NULL OR last_broadcast_at < NOW() - INTERVAL '24 hours')
186 + ",
187 + )
188 + .bind(user_id)
189 + .execute(pool)
190 + .await?;
191 +
192 + Ok(result.rows_affected() > 0)
193 + }
194 +
195 + /// Release the 24h broadcast slot. Used when a broadcast is refused after the
196 + /// slot has already been claimed (e.g. recipient cap exceeded) so the creator
197 + /// can retry without waiting a day.
198 + #[tracing::instrument(skip_all)]
199 + pub async fn clear_broadcast_at(pool: &PgPool, user_id: UserId) -> Result<()> {
200 + sqlx::query("UPDATE users SET last_broadcast_at = NULL WHERE id = $1")
201 + .bind(user_id)
202 + .execute(pool)
203 + .await?;
204 + Ok(())
205 + }
A server/src/db/users/stripe.rs +256
@@ -1,0 +1,256 @@
1 + //! Everything tying a user to their Stripe account: the connect handshake, the
2 + //! currency they settle in, and the founder and tax flags that ride along.
3 +
4 + use sqlx::PgPool;
5 +
6 + use crate::db::UserId;
7 + use crate::db::models::DbUser;
8 + use crate::error::Result;
9 +
10 + /// Update user's Stripe Connect account information after OAuth
11 + #[tracing::instrument(skip_all)]
12 + pub async fn update_user_stripe_account(
13 + pool: &PgPool,
14 + user_id: UserId,
15 + stripe_account_id: &str,
16 + onboarding_complete: bool,
17 + payouts_enabled: bool,
18 + charges_enabled: bool,
19 + ) -> Result<DbUser> {
20 + let user = sqlx::query_as::<_, DbUser>(
21 + r"
22 + UPDATE users
23 + SET stripe_account_id = $2,
24 + stripe_onboarding_complete = $3,
25 + stripe_payouts_enabled = $4,
26 + stripe_charges_enabled = $5,
27 + updated_at = NOW()
28 + WHERE id = $1
29 + RETURNING *
30 + ",
31 + )
32 + .bind(user_id)
33 + .bind(stripe_account_id)
34 + .bind(onboarding_complete)
35 + .bind(payouts_enabled)
36 + .bind(charges_enabled)
37 + .fetch_one(pool)
38 + .await?;
39 +
40 + Ok(user)
41 + }
42 +
43 + /// Atomically set a user's Stripe Connect account ID, but only if one is not
44 + /// already set. Returns `Some(user)` on success, or `None` if another request
45 + /// already claimed the slot (race-condition guard).
46 + #[tracing::instrument(skip_all)]
47 + pub async fn try_set_stripe_account(
48 + pool: &PgPool,
49 + user_id: UserId,
50 + stripe_account_id: &str,
51 + ) -> Result<Option<DbUser>> {
52 + let user = sqlx::query_as::<_, DbUser>(
53 + r"
54 + UPDATE users
55 + SET stripe_account_id = $2,
56 + stripe_onboarding_complete = false,
57 + stripe_payouts_enabled = false,
58 + stripe_charges_enabled = false,
59 + updated_at = NOW()
60 + WHERE id = $1 AND (stripe_account_id IS NULL OR stripe_account_id = '')
61 + RETURNING *
62 + ",
63 + )
64 + .bind(user_id)
65 + .bind(stripe_account_id)
66 + .fetch_optional(pool)
67 + .await?;
68 +
69 + Ok(user)
70 + }
71 +
72 + /// Update user's Stripe status from webhook (finds user by stripe_account_id)
73 + #[tracing::instrument(skip_all)]
74 + pub async fn update_user_stripe_status(
75 + pool: &PgPool,
76 + stripe_account_id: &str,
77 + onboarding_complete: bool,
78 + payouts_enabled: bool,
79 + charges_enabled: bool,
80 + settlement_currency: Option<crate::currency::SettlementCurrency>,
81 + ) -> Result<Option<DbUser>> {
82 + // COALESCE, not a plain assignment: `None` means Stripe told us nothing
83 + // usable this time (too early in onboarding, or a currency outside our six),
84 + // and overwriting a known currency with USD on that signal would redenominate
85 + // every price the creator has set.
86 + let user = sqlx::query_as::<_, DbUser>(
87 + r"
88 + UPDATE users
89 + SET stripe_onboarding_complete = $2,
90 + stripe_payouts_enabled = $3,
91 + stripe_charges_enabled = $4,
92 + settlement_currency = COALESCE($5, settlement_currency),
93 + updated_at = NOW()
94 + WHERE stripe_account_id = $1
95 + RETURNING *
96 + ",
97 + )
98 + .bind(stripe_account_id)
99 + .bind(onboarding_complete)
100 + .bind(payouts_enabled)
101 + .bind(charges_enabled)
102 + .bind(settlement_currency)
103 + .fetch_optional(pool)
104 + .await?;
105 +
106 + Ok(user)
107 + }
108 +
109 + /// The settlement currency currently stored for a connected account, if any.
110 + ///
111 + /// Read before a webhook write so a *change* can be distinguished from a
112 + /// restatement of the same value. Stripe re-sends `account.updated` constantly,
113 + /// so alerting on every write would be noise; alerting on none of them would
114 + /// leave a creator's prices silently meaning different money.
115 + #[tracing::instrument(skip_all)]
116 + pub async fn get_settlement_currency_by_stripe_account(
117 + pool: &PgPool,
118 + stripe_account_id: &str,
119 + ) -> Result<Option<crate::currency::SettlementCurrency>> {
120 + let row: Option<(crate::currency::SettlementCurrency,)> =
121 + sqlx::query_as("SELECT settlement_currency FROM users WHERE stripe_account_id = $1")
122 + .bind(stripe_account_id)
123 + .fetch_optional(pool)
124 + .await?;
125 + Ok(row.map(|(c,)| c))
126 + }
127 +
128 + /// The account behind a Stripe Connect account id.
129 + #[tracing::instrument(skip_all)]
130 + pub async fn get_user_id_by_stripe_account(
131 + pool: &PgPool,
132 + stripe_account_id: &str,
133 + ) -> Result<Option<UserId>> {
134 + let id = sqlx::query_scalar::<_, UserId>("SELECT id FROM users WHERE stripe_account_id = $1")
135 + .bind(stripe_account_id)
136 + .fetch_optional(pool)
137 + .await?;
138 + Ok(id)
139 + }
140 +
141 + /// Store a buyer's cross-currency conversion preference.
142 + ///
143 + /// A preference, not a lock: the checkout form decides each purchase, and this
144 + /// only changes what that form comes back pre-selected with next time.
145 + #[tracing::instrument(skip_all)]
146 + pub async fn set_conversion_preference(
147 + pool: &PgPool,
148 + user_id: UserId,
149 + conversion: crate::currency::ConversionChoice,
150 + ) -> Result<()> {
151 + sqlx::query("UPDATE users SET conversion_preference = $2, updated_at = NOW() WHERE id = $1")
152 + .bind(user_id)
153 + .bind(conversion)
154 + .execute(pool)
155 + .await?;
156 + Ok(())
157 + }
158 +
159 + /// Mark a user as a founder. Called when they start a creator-tier
160 + /// subscription while the founder pricing window is open. Sticky; never
161 + /// reset, even on cancellation. Subsequent re-subscriptions during the
162 + /// window keep their founder status. After the window closes, eligibility
163 + /// is determined by `founder_locked_at` (stamped only for users with an
164 + /// active subscription at the close-time snapshot).
165 + ///
166 + /// **DIY exclusion**: DIY-tier accounts are not full members and must not
167 + /// qualify for founder pricing. This function does not enforce that, it sets
168 + /// `is_founder` unconditionally, so the exclusion is a caller obligation: only
169 + /// call this from creator-tier (Basic/SmallFiles/BigFiles/Everything) checkout
170 + /// paths. When DIY ships, its checkout path must NOT invoke this.
171 + #[tracing::instrument(skip_all)]
172 + pub async fn mark_user_as_founder(pool: &PgPool, user_id: UserId) -> Result<()> {
173 + sqlx::query(
174 + r"
175 + UPDATE users
176 + SET is_founder = TRUE,
177 + updated_at = NOW()
178 + WHERE id = $1 AND is_founder = FALSE
179 + ",
180 + )
181 + .bind(user_id)
182 + .execute(pool)
183 + .await?;
184 + Ok(())
185 + }
186 +
187 + /// Close the founder pricing window by stamping `founder_locked_at` on every
188 + /// user who is currently flagged `is_founder` AND has an active creator-tier
189 + /// subscription. Returns the number of users locked in. Idempotent: skips
190 + /// any user already locked. Intended to be called once from an admin tool
191 + /// at the moment the founder window closes.
192 + #[tracing::instrument(skip_all)]
193 + pub async fn lock_in_founders_with_active_subscriptions(pool: &PgPool) -> Result<u64> {
194 + let result = sqlx::query(
195 + r"
196 + UPDATE users u
197 + SET founder_locked_at = NOW(),
198 + updated_at = NOW()
199 + WHERE u.is_founder = TRUE
200 + AND u.founder_locked_at IS NULL
201 + AND EXISTS (
202 + SELECT 1 FROM creator_subscriptions s
203 + WHERE s.user_id = u.id
204 + AND s.status = 'active'
205 + )
206 + ",
207 + )
208 + .execute(pool)
209 + .await?;
210 + Ok(result.rows_affected())
211 + }
212 +
213 + /// Update a user's Stripe Tax toggle.
214 + #[tracing::instrument(skip_all)]
215 + pub async fn update_stripe_tax_enabled(
216 + pool: &PgPool,
217 + user_id: UserId,
218 + enabled: bool,
219 + ) -> Result<()> {
220 + sqlx::query(
221 + r"
222 + UPDATE users
223 + SET stripe_tax_enabled = $2,
224 + updated_at = NOW()
225 + WHERE id = $1
226 + ",
227 + )
228 + .bind(user_id)
229 + .bind(enabled)
230 + .execute(pool)
231 + .await?;
232 +
233 + Ok(())
234 + }
235 +
236 + /// Disconnect a user's Stripe account
237 + #[tracing::instrument(skip_all)]
238 + pub async fn disconnect_user_stripe(pool: &PgPool, user_id: UserId) -> Result<DbUser> {
239 + let user = sqlx::query_as::<_, DbUser>(
240 + r"
241 + UPDATE users
242 + SET stripe_account_id = NULL,
243 + stripe_onboarding_complete = false,
244 + stripe_payouts_enabled = false,
245 + stripe_charges_enabled = false,
246 + updated_at = NOW()
247 + WHERE id = $1
248 + RETURNING *
249 + ",
250 + )
251 + .bind(user_id)
252 + .fetch_one(pool)
253 + .await?;
254 +
255 + Ok(user)
256 + }