Skip to main content

max / makenotwork

Decompose AppState Phase 3: clear handler holdouts Narrow the shared helpers the last holdout handlers depended on, then migrate those handlers off State<AppState>: - delete_user_account: AppState method -> free fn delete_user_account(&PgPool, &AppCaches, user_id). Clears profile::delete_account and account::confirm_delete_handler (now db+caches+email[+config]). - scheduler announcements (send_release_announcements/send_blog_post_ announcements + onboarding fns) -> (&PgPool, &EmailClient, &Config, ...). - scheduler mt_threads (spawn_mt_thread_for_item/blog_post + _by_lookup) -> (&PgPool, &BackgroundTx, &Integrations, &Config, ...). - Clears items/crud::update_item, blog::create/update_blog_post, wizards/item::step_save+save_preview (now db+email+config+bg+integrations). - scheduler mod/cleanup keep &AppState and project slice refs at the call sites (per-slice clones into the announce spawns). Routes tree now has exactly ONE State<AppState> extractor: require_admin_layer (the admin-gate middleware delegating to the AdminUser extractor, which needs &AppState) -- a correct composition root. No behavior change; blog 13 + mailing 11 + account 23 + wizard 23 + publish 18 + delete 41 tests pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-14 13:34 UTC
Commit: 0ea5f0ee05689ba7f333c2137b62d74e28394729
Parent: f6a7b23
11 files changed, +212 insertions, -136 deletions
M server/src/lib.rs +22 -13
@@ -431,19 +431,28 @@ impl AppState {
431 431 self.storage.require_public_s3()
432 432 }
433 433
434 - /// Delete a user account and purge every derived in-memory cache keyed to it.
435 - ///
436 - /// This is the single deletion entry point for handlers and the scheduler. The
437 - /// raw `db::users::delete_user` is `pub(crate)` so a caller cannot delete a user
438 - /// and forget to purge `domain_cache` — which has no TTL and never re-validates a
439 - /// hit, so a stale entry would otherwise linger until process restart (ultra-fuzz
440 - /// Run 12 doubledown S-E: couple cache invalidation to the mutation). The custom
441 - /// domain rows cascade at the DB layer; here we drop the matching cache entries.
442 - pub async fn delete_user_account(&self, user_id: db::UserId) -> error::Result<()> {
443 - db::users::delete_user(&self.db, user_id).await?;
444 - self.caches.domain_cache.retain(|_, uid| *uid != user_id);
445 - Ok(())
446 - }
434 + }
435 +
436 + /// Delete a user account and purge every derived in-memory cache keyed to it.
437 + ///
438 + /// This is the single deletion entry point for handlers and the scheduler. The
439 + /// raw `db::users::delete_user` is `pub(crate)` so a caller cannot delete a user
440 + /// and forget to purge `domain_cache` — which has no TTL and never re-validates a
441 + /// hit, so a stale entry would otherwise linger until process restart (ultra-fuzz
442 + /// Run 12 doubledown S-E: couple cache invalidation to the mutation). The custom
443 + /// domain rows cascade at the DB layer; here we drop the matching cache entries.
444 + ///
445 + /// Takes the `Db` and `AppCaches` slices rather than the whole `AppState` so a
446 + /// caller can hold just those (the decomposition seam); the coupling of the
447 + /// mutation to the invalidation is preserved by requiring both.
448 + pub async fn delete_user_account(
449 + db: &sqlx::PgPool,
450 + caches: &AppCaches,
451 + user_id: db::UserId,
452 + ) -> error::Result<()> {
453 + db::users::delete_user(db, user_id).await?;
454 + caches.domain_cache.retain(|_, uid| *uid != user_id);
455 + Ok(())
447 456 }
448 457
449 458 /// Build the app router with all routes and middleware (minus tracing/TCP).
M server/src/routes/api/blog.rs +27 -16
@@ -16,7 +16,8 @@ use crate::{
16 16 helpers::{htmx_toast_response, parse_schedule_datetime, slugify},
17 17 types::ListResponse,
18 18 validation,
19 - AppState,
19 + config::Config,
20 + Integrations,
20 21 };
21 22
22 23 use crate::extractors::ValidatedJson;
@@ -128,14 +129,19 @@ pub(super) async fn get_blog_post(
128 129
129 130 /// Create a new blog post under a project.
130 131 #[tracing::instrument(skip_all, name = "blog::create_blog_post")]
132 + #[allow(clippy::too_many_arguments)]
131 133 pub(super) async fn create_blog_post(
132 - State(state): State<AppState>,
134 + State(db): State<PgPool>,
135 + State(mailer): State<crate::email::EmailClient>,
136 + State(config): State<Config>,
137 + State(bg): State<crate::background::BackgroundTx>,
138 + State(integrations): State<Integrations>,
133 139 AuthUser(user): AuthUser,
134 140 Path(project_id): Path<ProjectId>,
135 141 ValidatedJson(req): ValidatedJson<CreateBlogPostRequest>,
136 142 ) -> Result<impl IntoResponse> {
137 143 user.check_not_suspended()?;
138 - verify_project_ownership(&state.db, project_id, user.id).await?;
144 + verify_project_ownership(&db, project_id, user.id).await?;
139 145
140 146 // Validate title
141 147 validation::validate_blog_post_title(&req.title)?;
@@ -149,7 +155,7 @@ pub(super) async fn create_blog_post(
149 155 let body_markdown = req.body_markdown.as_deref().unwrap_or("");
150 156 validation::validate_blog_post_body(body_markdown)?;
151 157
152 - let cdn_base = state.config.cdn_base_url.as_deref().unwrap_or("https://cdn.makenot.work");
158 + let cdn_base = config.cdn_base_url.as_deref().unwrap_or("https://cdn.makenot.work");
153 159 let body_html = crate::markdown::render_creator_markdown(body_markdown, user.id, cdn_base);
154 160 let is_published = req.is_published.unwrap_or(false);
155 161
@@ -161,7 +167,7 @@ pub(super) async fn create_blog_post(
161 167 // collision, covering the TOCTOU race where a concurrent request grabs the
162 168 // same slug.
163 169 let base = slug.to_string();
164 - let pool = &state.db;
170 + let pool = &db;
165 171 let (title_s, body_html_s) = (req.title.as_str(), body_html.as_str());
166 172 let post = crate::helpers::insert_with_unique_slug(&base, |slug| async move {
167 173 let slug = Slug::from_trusted(slug);
@@ -172,13 +178,13 @@ pub(super) async fn create_blog_post(
172 178 })
173 179 .await?;
174 180
175 - db::projects::bump_cache_generation(&state.db, project_id).await?;
181 + db::projects::bump_cache_generation(&db, project_id).await?;
176 182
177 183 // Create linked MT discussion thread and send announcements if published immediately
178 184 // (skip for sandbox users — no real emails or MT threads)
179 185 if post.published_at.is_some() && !user.is_sandbox {
180 - crate::scheduler::send_blog_post_announcements(&state, &post).await;
181 - crate::scheduler::spawn_mt_thread_for_blog_post(&state, &post, &user);
186 + crate::scheduler::send_blog_post_announcements(&db, &mailer, &config, &post).await;
187 + crate::scheduler::spawn_mt_thread_for_blog_post(&db, &bg, &integrations, &config, &post, &user);
182 188 }
183 189
184 190 Ok(Json(blog_post_response(&post)))
@@ -186,14 +192,19 @@ pub(super) async fn create_blog_post(
186 192
187 193 /// Update an existing blog post.
188 194 #[tracing::instrument(skip_all, name = "blog::update_blog_post")]
195 + #[allow(clippy::too_many_arguments)]
189 196 pub(super) async fn update_blog_post(
190 - State(state): State<AppState>,
197 + State(db): State<PgPool>,
198 + State(mailer): State<crate::email::EmailClient>,
199 + State(config): State<Config>,
200 + State(bg): State<crate::background::BackgroundTx>,
201 + State(integrations): State<Integrations>,
191 202 AuthUser(user): AuthUser,
192 203 Path(id): Path<BlogPostId>,
193 204 ValidatedJson(req): ValidatedJson<UpdateBlogPostRequest>,
194 205 ) -> Result<impl IntoResponse> {
195 206 user.check_not_suspended()?;
196 - let existing = verify_blog_post_ownership(&state.db, id, user.id).await?;
207 + let existing = verify_blog_post_ownership(&db, id, user.id).await?;
197 208
198 209 validation::validate_blog_post_title(&req.title)?;
199 210 // slug is validated by Slug's Deserialize impl
@@ -201,12 +212,12 @@ pub(super) async fn update_blog_post(
201 212
202 213 // Check slug uniqueness if changed
203 214 if req.slug != existing.slug
204 - && db::blog_posts::blog_post_slug_exists(&state.db, existing.project_id, &req.slug).await?
215 + && db::blog_posts::blog_post_slug_exists(&db, existing.project_id, &req.slug).await?
205 216 {
206 217 return Err(AppError::validation("A blog post with this slug already exists".to_string()));
207 218 }
208 219
209 - let cdn_base = state.config.cdn_base_url.as_deref().unwrap_or("https://cdn.makenot.work");
220 + let cdn_base = config.cdn_base_url.as_deref().unwrap_or("https://cdn.makenot.work");
210 221 let body_html = crate::markdown::render_creator_markdown(&req.body_markdown, user.id, cdn_base);
211 222
212 223 // Parse publish_at: None = no change, Some("") = clear, Some(datetime) = set schedule
@@ -229,7 +240,7 @@ pub(super) async fn update_blog_post(
229 240 };
230 241
231 242 let post = db::blog_posts::update_blog_post(
232 - &state.db,
243 + &db,
233 244 id,
234 245 &req.title,
235 246 &req.slug,
@@ -242,14 +253,14 @@ pub(super) async fn update_blog_post(
242 253 )
243 254 .await?;
244 255
245 - db::projects::bump_cache_generation(&state.db, existing.project_id).await?;
256 + db::projects::bump_cache_generation(&db, existing.project_id).await?;
246 257
247 258 // Detect first publish: was unpublished before, now published
248 259 // (skip for sandbox users — no real emails or MT threads)
249 260 if existing.published_at.is_none() && post.published_at.is_some() && !user.is_sandbox {
250 - crate::scheduler::send_blog_post_announcements(&state, &post).await;
261 + crate::scheduler::send_blog_post_announcements(&db, &mailer, &config, &post).await;
251 262 if post.mt_thread_id.is_none() {
252 - crate::scheduler::spawn_mt_thread_for_blog_post(&state, &post, &user);
263 + crate::scheduler::spawn_mt_thread_for_blog_post(&db, &bg, &integrations, &config, &post, &user);
253 264 }
254 265 }
255 266
M server/src/routes/api/items/crud.rs +13 -7
@@ -11,12 +11,13 @@ use sqlx::PgPool;
11 11
12 12 use crate::{
13 13 auth::AuthUser,
14 + config::Config,
14 15 db::{self, AiTier, ContentData, ItemId, ItemType, PriceCents, ProjectId},
15 16 error::{AppError, Result},
16 17 helpers::{is_htmx_request, parse_schedule_datetime},
17 18 templates::SaveStatusTemplate,
18 19 validation,
19 - AppState,
20 + Integrations,
20 21 };
21 22
22 23 use super::super::{verify_item_ownership, verify_project_ownership};
@@ -164,8 +165,13 @@ pub struct UpdateItemRequest {
164 165
165 166 /// Update an existing item owned by the authenticated user.
166 167 #[tracing::instrument(skip_all, name = "items::update_item", fields(item_id))]
168 + #[allow(clippy::too_many_arguments)]
167 169 pub(in crate::routes::api) async fn update_item(
168 - State(state): State<AppState>,
170 + State(db): State<PgPool>,
171 + State(mailer): State<crate::email::EmailClient>,
172 + State(config): State<Config>,
173 + State(bg): State<crate::background::BackgroundTx>,
174 + State(integrations): State<Integrations>,
169 175 headers: HeaderMap,
170 176 AuthUser(user): AuthUser,
171 177 Path(id): Path<ItemId>,
@@ -173,7 +179,7 @@ pub(in crate::routes::api) async fn update_item(
173 179 ) -> Result<Response> {
174 180 tracing::Span::current().record("item_id", tracing::field::display(&id));
175 181 user.check_not_suspended()?;
176 - verify_item_ownership(&state.db, id, user.id).await?;
182 + verify_item_ownership(&db, id, user.id).await?;
177 183
178 184 // Validate input (same rules as create_item, but all fields are optional)
179 185 if let Some(ref title) = req.title {
@@ -226,7 +232,7 @@ pub(in crate::routes::api) async fn update_item(
226 232 };
227 233
228 234 let updated = db::items::update_item(
229 - &state.db,
235 + &db,
230 236 id,
231 237 user.id,
232 238 req.title.as_deref(),
@@ -246,15 +252,15 @@ pub(in crate::routes::api) async fn update_item(
246 252 // Detect first publish: if the request set is_public=true and the item is now public,
247 253 // atomically mark as announced and send release emails to followers.
248 254 if req.is_public == Some(true) && updated.is_public {
249 - crate::scheduler::send_release_announcements(&state, &updated).await;
255 + crate::scheduler::send_release_announcements(&db, &mailer, &config, &updated).await;
250 256
251 257 // Create linked MT discussion thread on first publish
252 258 if updated.mt_thread_id.is_none() {
253 - crate::scheduler::spawn_mt_thread_for_item(&state, &updated, &user);
259 + crate::scheduler::spawn_mt_thread_for_item(&db, &bg, &integrations, &config, &updated, &user);
254 260 }
255 261 }
256 262
257 - db::projects::bump_cache_generation(&state.db, updated.project_id).await?;
263 + db::projects::bump_cache_generation(&db, updated.project_id).await?;
258 264
259 265 if is_htmx_request(&headers) {
260 266 return Ok(axum::response::Html("Saved.".to_string()).into_response());
M server/src/routes/api/users/profile.rs +9 -7
@@ -20,7 +20,7 @@ use crate::{
20 20 helpers::is_htmx_request,
21 21 templates::{AlertTemplate, FormStatusTemplate, SaveStatusTemplate},
22 22 validation,
23 - AppCaches, AppState, Billing, Integrations,
23 + AppCaches, Billing, Integrations,
24 24 };
25 25
26 26 use crate::extractors::ValidatedForm;
@@ -251,18 +251,20 @@ pub(in crate::routes::api) async fn update_password(
251 251 /// so buyers can download their purchased files before removal.
252 252 #[tracing::instrument(skip_all, name = "users::delete_account")]
253 253 pub(in crate::routes::api) async fn delete_account(
254 - State(state): State<AppState>,
254 + State(db): State<PgPool>,
255 + State(mailer): State<crate::email::EmailClient>,
256 + State(caches): State<crate::AppCaches>,
255 257 AuthUser(user): AuthUser,
256 258 ) -> Result<impl IntoResponse> {
257 259 user.check_not_sandbox()?;
258 260
259 - if db::users::has_completed_sales(&state.db, user.id).await? {
260 - db::users::schedule_content_removal(&state.db, user.id).await?;
261 + if db::users::has_completed_sales(&db, user.id).await? {
262 + db::users::schedule_content_removal(&db, user.id).await?;
261 263 tracing::info!(user_id = %user.id, "creator account deletion scheduled with 90-day content grace period");
262 264
263 265 // Notify historical buyers (capped + Postmark-throttled). Fire-and-forget.
264 - let pool = state.db.clone();
265 - let email = state.email.clone();
266 + let pool = db.clone();
267 + let email = mailer.clone();
266 268 let creator_name = user.display_name.clone()
267 269 .unwrap_or_else(|| user.username.to_string());
268 270 let user_id = user.id;
@@ -270,7 +272,7 @@ pub(in crate::routes::api) async fn delete_account(
270 272 crate::email::send_creator_departure_notifications(&pool, &email, user_id, creator_name).await;
271 273 });
272 274 } else {
273 - state.delete_user_account(user.id).await?;
275 + crate::delete_user_account(&db, &caches, user.id).await?;
274 276 }
275 277
276 278 Ok(StatusCode::NO_CONTENT)
M server/src/routes/pages/dashboard/wizards/item/mod.rs +21 -14
@@ -16,11 +16,12 @@ use tower_sessions::Session;
16 16
17 17 use crate::{
18 18 auth::AuthUser,
19 + config::Config,
19 20 db::{self, ItemId, ItemType, PriceCents, ProjectFeature, Slug},
20 21 error::{AppError, Result},
21 22 helpers::get_csrf_token,
22 23 templates::*,
23 - AppState,
24 + Integrations,
24 25 };
25 26 use sqlx::PgPool;
26 27
@@ -253,8 +254,13 @@ pub async fn step_load(
253 254
254 255 /// POST /dashboard/project/{slug}/new-item/{id}/step/{step}
255 256 #[tracing::instrument(skip_all, name = "wizard::item_step_save")]
257 + #[allow(clippy::too_many_arguments)]
256 258 pub async fn step_save(
257 - State(state): State<AppState>,
259 + State(db): State<PgPool>,
260 + State(mailer): State<crate::email::EmailClient>,
261 + State(config): State<Config>,
262 + State(bg): State<crate::background::BackgroundTx>,
263 + State(integrations): State<Integrations>,
258 264 session: Session,
259 265 AuthUser(user): AuthUser,
260 266 Path((slug, id, step)): Path<(String, String, String)>,
@@ -262,19 +268,20 @@ pub async fn step_save(
262 268 Form(form): Form<HashMap<String, String>>,
263 269 ) -> Result<Response> {
264 270 user.check_not_suspended()?;
265 - let (project, item) = verify_item_wizard_access(&state.db, &user, &slug, &id).await?;
271 + let (project, item) = verify_item_wizard_access(&db, &user, &slug, &id).await?;
266 272
267 273 let save_result = match step.as_str() {
268 - "type" => save::save_type(&state.db, &project, &item, &form, user.id).await,
269 - "basics" => save::save_basics(&state.db, &item, &form, user.id).await,
270 - "content" => save::save_content(&state.db, &item, &form, user.id).await,
274 + "type" => save::save_type(&db, &project, &item, &form, user.id).await,
275 + "basics" => save::save_basics(&db, &item, &form, user.id).await,
276 + "content" => save::save_content(&db, &item, &form, user.id).await,
271 277 "sections" => Ok(()), // Sections managed via HTMX API; pass-through
272 - "pricing" => save::save_pricing(&state.db, &item, &form, user.id).await,
273 - // save_preview stays on the full &AppState: it fans out to the
274 - // crate::scheduler helpers (send_release_announcements /
275 - // spawn_mt_thread_for_item), which are shared &AppState entry points
276 - // outside this scope. Mirrors crud::update_item.
277 - "preview" => return save::save_preview(&state, &user, &project, &item, &form).await,
278 + "pricing" => save::save_pricing(&db, &item, &form, user.id).await,
279 + "preview" => {
280 + return save::save_preview(
281 + &db, &mailer, &config, &bg, &integrations, &user, &project, &item, &form,
282 + )
283 + .await;
284 + }
278 285 _ => return Err(AppError::NotFound),
279 286 };
280 287
@@ -299,10 +306,10 @@ pub async fn step_save(
299 306 }
300 307
301 308 // Re-fetch item after update
302 - let item = db::items::get_item_by_id(&state.db, item.id)
309 + let item = db::items::get_item_by_id(&db, item.id)
303 310 .await?
304 311 .ok_or(AppError::NotFound)?;
305 312
306 313 let next = super::next_step(ITEM_STEPS, &step).ok_or(AppError::NotFound)?;
307 - render::render_step(&state.db, &session, &user, &project, &item, next).await
314 + render::render_step(&db, &session, &user, &project, &item, next).await
308 315 }
M server/src/routes/pages/dashboard/wizards/item/save.rs +11 -7
@@ -9,7 +9,6 @@ use crate::{
9 9 error::{AppError, Result},
10 10 pricing::parse_dollars_to_cents,
11 11 validation,
12 - AppState,
13 12 };
14 13 use sqlx::PgPool;
15 14
@@ -240,8 +239,13 @@ pub(super) async fn save_pricing(
240 239 Ok(())
241 240 }
242 241
242 + #[allow(clippy::too_many_arguments)]
243 243 pub(super) async fn save_preview(
244 - state: &AppState,
244 + db: &sqlx::PgPool,
245 + mailer: &crate::email::EmailClient,
246 + config: &crate::config::Config,
247 + bg: &crate::background::BackgroundTx,
248 + integrations: &crate::Integrations,
245 249 user: &crate::auth::SessionUser,
246 250 _project: &db::DbProject,
247 251 item: &db::DbItem,
@@ -252,7 +256,7 @@ pub(super) async fn save_preview(
252 256 match action {
253 257 "publish" => {
254 258 db::items::update_item(
255 - &state.db,
259 + db,
256 260 item.id,
257 261 user.id,
258 262 None,
@@ -269,15 +273,15 @@ pub(super) async fn save_preview(
269 273 .await?;
270 274
271 275 // Re-fetch to get updated is_public state
272 - let updated = db::items::get_item_by_id(&state.db, item.id)
276 + let updated = db::items::get_item_by_id(db, item.id)
273 277 .await?
274 278 .ok_or(AppError::NotFound)?;
275 279
276 280 if updated.is_public {
277 - crate::scheduler::send_release_announcements(state, &updated).await;
281 + crate::scheduler::send_release_announcements(db, mailer, config, &updated).await;
278 282
279 283 if updated.mt_thread_id.is_none() {
280 - crate::scheduler::spawn_mt_thread_for_item(state, &updated, user);
284 + crate::scheduler::spawn_mt_thread_for_item(db, bg, integrations, config, &updated, user);
281 285 }
282 286 }
283 287 }
@@ -293,7 +297,7 @@ pub(super) async fn save_preview(
293 297 .map_err(|_| AppError::validation("Enter a valid publish date and time.".to_string()))?;
294 298 let utc_dt = dt.and_utc();
295 299 db::items::update_item(
296 - &state.db,
300 + db,
297 301 item.id,
298 302 user.id,
299 303 None,
M server/src/routes/pages/email_actions/account.rs +10 -8
@@ -16,7 +16,6 @@ use crate::{
16 16 error::{AppError, Result},
17 17 helpers::{constant_time_compare, get_csrf_token},
18 18 templates::*,
19 - AppState,
20 19 };
21 20
22 21 /// Query parameters for the signed account deletion confirmation link.
@@ -153,13 +152,16 @@ pub(super) async fn confirm_delete_page(
153 152 /// the account and destroys the session.
154 153 #[tracing::instrument(skip_all, name = "email_actions::confirm_delete_handler")]
155 154 pub(super) async fn confirm_delete_handler(
156 - State(state): State<AppState>,
155 + State(db): State<PgPool>,
156 + State(config): State<Config>,
157 + State(mailer): State<crate::email::EmailClient>,
158 + State(caches): State<crate::AppCaches>,
157 159 session: Session,
158 160 Form(form): Form<ConfirmDeleteForm>,
159 161 ) -> Result<Response> {
160 162 // Validate the deletion link parameters
161 163 let user_id =
162 - match validate_deletion_link(&state.db, &state.config, &form.user, &form.expires, &form.sig)
164 + match validate_deletion_link(&db, &config, &form.user, &form.expires, &form.sig)
163 165 .await?
164 166 {
165 167 Err(error_response) => return Ok(error_response),
@@ -167,13 +169,13 @@ pub(super) async fn confirm_delete_handler(
167 169 };
168 170
169 171 // If creator has sales, schedule 90-day content grace period instead of immediate deletion
170 - if db::users::has_completed_sales(&state.db, user_id).await? {
171 - db::users::schedule_content_removal(&state.db, user_id).await?;
172 + if db::users::has_completed_sales(&db, user_id).await? {
173 + db::users::schedule_content_removal(&db, user_id).await?;
172 174 tracing::info!(user_id = %user_id, event = "account_deletion_scheduled", "Creator account scheduled for removal with 90-day content grace period");
173 175
174 176 // Notify all buyers that this creator is leaving (fire-and-forget)
175 - let pool = state.db.clone();
176 - let email_client = state.email.clone();
177 + let pool = db.clone();
178 + let email_client = mailer.clone();
177 179 tokio::spawn(async move {
178 180 let creator_name = match db::users::get_user_by_id(&pool, user_id).await {
179 181 Ok(Some(u)) => u.display_name.unwrap_or_else(|| u.username.to_string()),
@@ -182,7 +184,7 @@ pub(super) async fn confirm_delete_handler(
182 184 crate::email::send_creator_departure_notifications(&pool, &email_client, user_id, creator_name).await;
183 185 });
184 186 } else {
185 - state.delete_user_account(user_id).await?;
187 + crate::delete_user_account(&db, &caches, user_id).await?;
186 188 tracing::info!(user_id = %user_id, event = "account_deleted", "Account permanently deleted via confirmed POST");
187 189 }
188 190
M server/src/scheduler/announcements.rs +37 -34
@@ -1,8 +1,11 @@
1 1 //! Release and blog post announcement emails via project mailing lists.
2 2
3 + use sqlx::PgPool;
4 +
5 + use crate::config::Config;
3 6 use crate::db;
4 7 use crate::db::{DbBlogPost, DbItem, DbUser};
5 - use crate::AppState;
8 + use crate::email::EmailClient;
6 9
7 10 /// Build the mailing-list unsubscribe URL for one subscriber: user-keyed for an
8 11 /// MNW account, email-keyed for an imported email-only subscriber (which has no
@@ -61,8 +64,8 @@ where
61 64 /// scheduler. Safe to call multiple times — `mark_release_announced`
62 65 /// is a no-op if the item was already announced.
63 66 #[tracing::instrument(skip_all, name = "scheduler::send_release_announcements")]
64 - pub async fn send_release_announcements(state: &AppState, item: &DbItem) {
65 - if !db::items::mark_release_announced(&state.db, item.id)
67 + pub async fn send_release_announcements(db: &PgPool, mailer: &EmailClient, config: &Config, item: &DbItem) {
68 + if !db::items::mark_release_announced(db, item.id)
66 69 .await
67 70 .unwrap_or(false)
68 71 {
@@ -74,15 +77,15 @@ pub async fn send_release_announcements(state: &AppState, item: &DbItem) {
74 77 return;
75 78 }
76 79
77 - let Ok(Some(project)) = db::projects::get_project_by_id(&state.db, item.project_id).await
80 + let Ok(Some(project)) = db::projects::get_project_by_id(db, item.project_id).await
78 81 else {
79 82 return;
80 83 };
81 - let Ok(Some(creator)) = db::users::get_user_by_id(&state.db, project.user_id).await else {
84 + let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
82 85 return;
83 86 };
84 87 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
85 - &state.db,
88 + db,
86 89 item.project_id,
87 90 db::MailingListType::Content,
88 91 )
@@ -90,7 +93,7 @@ pub async fn send_release_announcements(state: &AppState, item: &DbItem) {
90 93 else {
91 94 return;
92 95 };
93 - let Ok(subscribers) = db::mailing_lists::get_subscribers(&state.db, list.id).await
96 + let Ok(subscribers) = db::mailing_lists::get_subscribers(db, list.id).await
94 97 else {
95 98 return;
96 99 };
@@ -101,10 +104,10 @@ pub async fn send_release_announcements(state: &AppState, item: &DbItem) {
101 104 .unwrap_or(&creator.username)
102 105 .to_string();
103 106 let item_title = item.title.clone();
104 - let item_url = format!("{}/i/{}", state.config.host_url, item.id);
105 - let email_client = state.email.clone();
106 - let host_url = state.config.host_url.clone();
107 - let signing_secret = state.config.signing_secret.clone();
107 + let item_url = format!("{}/i/{}", config.host_url, item.id);
108 + let email_client = mailer.clone();
109 + let host_url = config.host_url.clone();
110 + let signing_secret = config.signing_secret.clone();
108 111 let list_id_str = list.id.to_string();
109 112
110 113 spawn_bounded_fanout(subscribers, move |subscriber| {
@@ -141,8 +144,8 @@ pub async fn send_release_announcements(state: &AppState, item: &DbItem) {
141 144 /// Safe to call multiple times — `mark_blog_post_announced` is a no-op
142 145 /// if the post was already announced.
143 146 #[tracing::instrument(skip_all, name = "scheduler::send_blog_post_announcements")]
144 - pub async fn send_blog_post_announcements(state: &AppState, post: &DbBlogPost) {
145 - if !db::blog_posts::mark_blog_post_announced(&state.db, post.id)
147 + pub async fn send_blog_post_announcements(db: &PgPool, mailer: &EmailClient, config: &Config, post: &DbBlogPost) {
148 + if !db::blog_posts::mark_blog_post_announced(db, post.id)
146 149 .await
147 150 .unwrap_or(false)
148 151 {
@@ -154,15 +157,15 @@ pub async fn send_blog_post_announcements(state: &AppState, post: &DbBlogPost) {
154 157 return;
155 158 }
156 159
157 - let Ok(Some(project)) = db::projects::get_project_by_id(&state.db, post.project_id).await
160 + let Ok(Some(project)) = db::projects::get_project_by_id(db, post.project_id).await
158 161 else {
159 162 return;
160 163 };
161 - let Ok(Some(creator)) = db::users::get_user_by_id(&state.db, project.user_id).await else {
164 + let Ok(Some(creator)) = db::users::get_user_by_id(db, project.user_id).await else {
162 165 return;
163 166 };
164 167 let Ok(Some(list)) = db::mailing_lists::get_list_by_project_and_type(
165 - &state.db,
168 + db,
166 169 post.project_id,
167 170 db::MailingListType::Content,
168 171 )
@@ -170,7 +173,7 @@ pub async fn send_blog_post_announcements(state: &AppState, post: &DbBlogPost) {
170 173 else {
171 174 return;
172 175 };
173 - let Ok(subscribers) = db::mailing_lists::get_subscribers(&state.db, list.id).await
176 + let Ok(subscribers) = db::mailing_lists::get_subscribers(db, list.id).await
174 177 else {
175 178 return;
176 179 };
@@ -183,11 +186,11 @@ pub async fn send_blog_post_announcements(state: &AppState, post: &DbBlogPost) {
183 186 let post_title = post.title.clone();
184 187 let post_url = format!(
185 188 "{}/{}/blog/{}",
186 - state.config.host_url, project.slug, post.slug
189 + config.host_url, project.slug, post.slug
187 190 );
188 - let email_client = state.email.clone();
189 - let host_url = state.config.host_url.clone();
190 - let signing_secret = state.config.signing_secret.clone();
191 + let email_client = mailer.clone();
192 + let host_url = config.host_url.clone();
193 + let signing_secret = config.signing_secret.clone();
191 194 let list_id_str = list.id.to_string();
192 195
193 196 spawn_bounded_fanout(subscribers, move |subscriber| {
@@ -245,40 +248,40 @@ impl OnboardingStep {
245 248 /// so a backlog of serial email I/O can't extend the advisory-lock hold time
246 249 /// (Run #14 MEDIUM, mirrors the release/blog announcement fan-out).
247 250 #[tracing::instrument(skip_all, name = "scheduler::send_onboarding_emails")]
248 - pub(super) async fn send_onboarding_emails(state: &AppState) {
251 + pub(super) async fn send_onboarding_emails(db: &PgPool, mailer: &EmailClient, config: &Config) {
249 252 // Step 1→2: profile tips (24h after welcome)
250 253 let next = OnboardingStep::ProfileTipsSent;
251 254 if let Ok(users) =
252 - db::users::get_onboarding_candidates(&state.db, OnboardingStep::WelcomeSent.as_i16(), chrono::Duration::hours(24)).await
255 + db::users::get_onboarding_candidates(db, OnboardingStep::WelcomeSent.as_i16(), chrono::Duration::hours(24)).await
253 256 {
254 257 // Batch-advance users who already set a display name (skip email)
255 258 let (skip, send): (Vec<_>, Vec<_>) =
256 259 users.into_iter().partition(|u| u.display_name.is_some());
257 - advance_skipped(state, &skip, next).await;
258 - claim_and_spawn_sends(state, send, next).await;
260 + advance_skipped(db, &skip, next).await;
261 + claim_and_spawn_sends(db, mailer, config, send, next).await;
259 262 }
260 263
261 264 // Step 2→3: Stripe guide (72h after welcome)
262 265 let next = OnboardingStep::StripeGuideSent;
263 266 if let Ok(users) =
264 - db::users::get_onboarding_candidates(&state.db, OnboardingStep::ProfileTipsSent.as_i16(), chrono::Duration::hours(48)).await
267 + db::users::get_onboarding_candidates(db, OnboardingStep::ProfileTipsSent.as_i16(), chrono::Duration::hours(48)).await
265 268 {
266 269 // Batch-advance users who already connected Stripe (skip email)
267 270 let (skip, send): (Vec<_>, Vec<_>) =
268 271 users.into_iter().partition(|u| u.stripe_account_id.is_some());
269 - advance_skipped(state, &skip, next).await;
270 - claim_and_spawn_sends(state, send, next).await;
272 + advance_skipped(db, &skip, next).await;
273 + claim_and_spawn_sends(db, mailer, config, send, next).await;
271 274 }
272 275 }
273 276
274 277 /// Batch-advance users who don't need an email for this step (display name /
275 278 /// Stripe already set). Inline — a single cheap UPDATE.
276 - async fn advance_skipped(state: &AppState, skip: &[DbUser], next: OnboardingStep) {
279 + async fn advance_skipped(db: &PgPool, skip: &[DbUser], next: OnboardingStep) {
277 280 if skip.is_empty() {
278 281 return;
279 282 }
280 283 let skip_ids: Vec<_> = skip.iter().map(|u| u.id).collect();
281 - if let Err(e) = db::users::batch_advance_onboarding_step(&state.db, &skip_ids, next.as_i16()).await {
284 + if let Err(e) = db::users::batch_advance_onboarding_step(db, &skip_ids, next.as_i16()).await {
282 285 tracing::warn!(count = skip_ids.len(), step = ?next, error = ?e, "failed to batch advance onboarding step");
283 286 }
284 287 }
@@ -289,18 +292,18 @@ async fn advance_skipped(state: &AppState, skip: &[DbUser], next: OnboardingStep
289 292 /// claim leaves the rows untouched to retry next tick rather than sending
290 293 /// without a claim. Missing a non-critical onboarding email is better than
291 294 /// sending it twice.
292 - async fn claim_and_spawn_sends(state: &AppState, send: Vec<DbUser>, next: OnboardingStep) {
295 + async fn claim_and_spawn_sends(db: &PgPool, mailer: &EmailClient, config: &Config, send: Vec<DbUser>, next: OnboardingStep) {
293 296 if send.is_empty() {
294 297 return;
295 298 }
296 299 let send_ids: Vec<_> = send.iter().map(|u| u.id).collect();
297 - if let Err(e) = db::users::batch_advance_onboarding_step(&state.db, &send_ids, next.as_i16()).await {
300 + if let Err(e) = db::users::batch_advance_onboarding_step(db, &send_ids, next.as_i16()).await {
298 301 tracing::warn!(count = send_ids.len(), step = ?next, error = ?e, "failed to claim onboarding batch; retrying next tick");
299 302 return;
300 303 }
301 304
302 - let email_client = state.email.clone();
303 - let host_url = state.config.host_url.clone();
305 + let email_client = mailer.clone();
306 + let host_url = config.host_url.clone();
304 307 spawn_bounded_fanout(send, move |user| {
305 308 let email_client = email_client.clone();
306 309 let host_url = host_url.clone();
M server/src/scheduler/cleanup.rs +1 -1
@@ -178,7 +178,7 @@ async fn cleanup_user_s3_and_delete(state: &AppState, user_id: db::UserId, event
178 178 }
179 179
180 180 // CASCADE delete user row (+ purge domain_cache via the coupled entry point)
181 - if let Err(e) = state.delete_user_account(user_id).await {
181 + if let Err(e) = crate::delete_user_account(&state.db, &state.caches, user_id).await {
182 182 tracing::error!(error = ?e, %user_id, "{label}: failed to delete account");
183 183 false
184 184 } else {
M server/src/scheduler/mod.rs +18 -8
@@ -17,9 +17,11 @@ mod webhooks;
17 17 use tokio::sync::watch;
18 18 use tokio::task::JoinHandle;
19 19
20 + use axum::extract::FromRef;
21 +
20 22 use crate::constants;
21 23 use crate::db;
22 - use crate::AppState;
24 + use crate::{AppState, Integrations};
23 25
24 26 // Re-export public API used by route handlers
25 27 pub use announcements::{send_blog_post_announcements, send_release_announcements};
@@ -174,15 +176,19 @@ pub fn spawn_scheduler(
174 176 title = %item.title,
175 177 "scheduler published item"
176 178 );
177 - let state_for_announce = state.clone();
179 + let db_pool = state.db.clone();
180 + let mailer = state.email.clone();
181 + let cfg = state.config.clone();
178 182 let item_for_announce = item.clone();
179 183 state.bg.spawn("release-announcements", async move {
180 184 announcements::send_release_announcements(
181 - &state_for_announce, &item_for_announce,
185 + &db_pool, &mailer, &cfg, &item_for_announce,
182 186 ).await;
183 187 });
184 188 if item.mt_thread_id.is_none() {
185 - mt_threads::spawn_mt_thread_for_item_by_lookup(&state, item);
189 + mt_threads::spawn_mt_thread_for_item_by_lookup(
190 + &state.db, &state.bg, &Integrations::from_ref(&state), &state.config, item,
191 + );
186 192 }
187 193 }
188 194 }
@@ -192,7 +198,7 @@ pub fn spawn_scheduler(
192 198 }
193 199
194 200 // Send onboarding drip emails
195 - announcements::send_onboarding_emails(&state).await;
201 + announcements::send_onboarding_emails(&state.db, &state.email, &state.config).await;
196 202
197 203 // Dispatch pending builds
198 204 crate::build_runner::dispatch_pending_build(&state).await;
@@ -206,15 +212,19 @@ pub fn spawn_scheduler(
206 212 title = %post.title,
207 213 "scheduler published blog post"
208 214 );
209 - let state_for_announce = state.clone();
215 + let db_pool = state.db.clone();
216 + let mailer = state.email.clone();
217 + let cfg = state.config.clone();
210 218 let post_for_announce = post.clone();
211 219 state.bg.spawn("blog-post-announcements", async move {
212 220 announcements::send_blog_post_announcements(
213 - &state_for_announce, &post_for_announce,
221 + &db_pool, &mailer, &cfg, &post_for_announce,
214 222 ).await;
215 223 });
216 224 if post.mt_thread_id.is_none() {
217 - mt_threads::spawn_mt_thread_for_blog_post_by_lookup(&state, post);
225 + mt_threads::spawn_mt_thread_for_blog_post_by_lookup(
226 + &state.db, &state.bg, &Integrations::from_ref(&state), &state.config, post,
227 + );
218 228 }
219 229 }
220 230 }
M server/src/scheduler/mt_threads.rs +43 -21
@@ -1,22 +1,29 @@
1 1 //! Multithreaded forum thread provisioning for published items and blog posts.
2 2
3 + use sqlx::PgPool;
4 +
5 + use crate::background::BackgroundTx;
6 + use crate::config::Config;
3 7 use crate::db;
4 8 use crate::db::{DbBlogPost, DbItem};
5 - use crate::AppState;
9 + use crate::Integrations;
6 10
7 11 /// Fire-and-forget: create an MT discussion thread for a published item.
8 12 /// Called from the item update handler where the user is available.
9 13 pub fn spawn_mt_thread_for_item(
10 - state: &AppState,
14 + db: &PgPool,
15 + bg: &BackgroundTx,
16 + integrations: &Integrations,
17 + config: &Config,
11 18 item: &DbItem,
12 19 user: &crate::auth::SessionUser,
13 20 ) {
14 - let Some(ref mt) = state.mt_client else {
21 + let Some(ref mt) = integrations.mt_client else {
15 22 return;
16 23 };
17 24 let mt = mt.clone();
18 - let db_pool = state.db.clone();
19 - let host_url = state.config.host_url.clone();
25 + let db_pool = db.clone();
26 + let host_url = config.host_url.clone();
20 27 let item_id = item.id;
21 28 let item_title = item.title.clone();
22 29 let project_id = item.project_id;
@@ -24,7 +31,7 @@ pub fn spawn_mt_thread_for_item(
24 31 let username = user.username.to_string();
25 32 let display_name = user.display_name.clone();
26 33
27 - state.bg.spawn("mt-thread-item", async move {
34 + bg.spawn("mt-thread-item", async move {
28 35 create_mt_thread_for_item(
29 36 &mt,
30 37 &db_pool,
@@ -42,18 +49,24 @@ pub fn spawn_mt_thread_for_item(
42 49
43 50 /// Fire-and-forget: create an MT discussion thread for a published item
44 51 /// (scheduler version — looks up the project/user from DB).
45 - pub(super) fn spawn_mt_thread_for_item_by_lookup(state: &AppState, item: &DbItem) {
46 - let Some(ref mt) = state.mt_client else {
52 + pub(super) fn spawn_mt_thread_for_item_by_lookup(
53 + db: &PgPool,
54 + bg: &BackgroundTx,
55 + integrations: &Integrations,
56 + config: &Config,
57 + item: &DbItem,
58 + ) {
59 + let Some(ref mt) = integrations.mt_client else {
47 60 return;
48 61 };
49 62 let mt = mt.clone();
50 - let db_pool = state.db.clone();
51 - let host_url = state.config.host_url.clone();
63 + let db_pool = db.clone();
64 + let host_url = config.host_url.clone();
52 65 let item_id = item.id;
53 66 let item_title = item.title.clone();
54 67 let project_id = item.project_id;
55 68
56 - state.bg.spawn("mt-thread-item-lookup", async move {
69 + bg.spawn("mt-thread-item-lookup", async move {
57 70 let Ok(Some(project)) = db::projects::get_project_by_id(&db_pool, project_id).await
58 71 else {
59 72 return;
@@ -121,16 +134,19 @@ async fn create_mt_thread_for_item(
121 134 /// Fire-and-forget: create an MT discussion thread for a published blog post.
122 135 /// Called from the blog post handler where the user is available.
123 136 pub fn spawn_mt_thread_for_blog_post(
124 - state: &AppState,
137 + db: &PgPool,
138 + bg: &BackgroundTx,
139 + integrations: &Integrations,
140 + config: &Config,
125 141 post: &DbBlogPost,
126 142 user: &crate::auth::SessionUser,
127 143 ) {
128 - let Some(ref mt) = state.mt_client else {
144 + let Some(ref mt) = integrations.mt_client else {
129 145 return;
130 146 };
131 147 let mt = mt.clone();
132 - let db_pool = state.db.clone();
133 - let host_url = state.config.host_url.clone();
148 + let db_pool = db.clone();
149 + let host_url = config.host_url.clone();
134 150 let post_id = post.id;
135 151 let post_title = post.title.clone();
136 152 let post_slug = post.slug.to_string();
@@ -139,7 +155,7 @@ pub fn spawn_mt_thread_for_blog_post(
139 155 let username = user.username.to_string();
140 156 let display_name = user.display_name.clone();
141 157
142 - state.bg.spawn("mt-thread-blog", async move {
158 + bg.spawn("mt-thread-blog", async move {
143 159 create_mt_thread_for_blog_post(
144 160 &mt,
145 161 &db_pool,
@@ -158,19 +174,25 @@ pub fn spawn_mt_thread_for_blog_post(
158 174
159 175 /// Fire-and-forget: create an MT discussion thread for a published blog post
160 176 /// (scheduler version — looks up the project/user from DB).
161 - pub(super) fn spawn_mt_thread_for_blog_post_by_lookup(state: &AppState, post: &DbBlogPost) {
162 - let Some(ref mt) = state.mt_client else {
177 + pub(super) fn spawn_mt_thread_for_blog_post_by_lookup(
178 + db: &PgPool,
179 + bg: &BackgroundTx,
180 + integrations: &Integrations,
181 + config: &Config,
182 + post: &DbBlogPost,
183 + ) {
184 + let Some(ref mt) = integrations.mt_client else {
163 185 return;
164 186 };
165 187 let mt = mt.clone();
166 - let db_pool = state.db.clone();
167 - let host_url = state.config.host_url.clone();
188 + let db_pool = db.clone();
189 + let host_url = config.host_url.clone();
168 190 let post_id = post.id;
169 191 let post_title = post.title.clone();
170 192 let post_slug = post.slug.to_string();
171 193 let project_id = post.project_id;
172 194
173 - state.bg.spawn("mt-thread-blog-lookup", async move {
195 + bg.spawn("mt-thread-blog-lookup", async move {
174 196 let Ok(Some(project)) = db::projects::get_project_by_id(&db_pool, project_id).await
175 197 else {
176 198 return;