fix(bans): backfill tracked members + harden SQLite concurrency

crates/rustmail/src/db/operations/banned_users.rs

@@ -81,6 +81,79 @@ pub async fn upsert_tracked_member(member: &TrackedMember, pool: &SqlitePool) ->
     Ok(())
 }
 
+pub async fn bulk_upsert_tracked_members(
+    members: &[TrackedMember],
+    pool: &SqlitePool,
+) -> ModmailResult<()> {
+    if members.is_empty() {
+        return Ok(());
+    }
+
+    // Pre-serialize all roles to JSON up front so errors surface before the transaction opens.
+    let roles_jsons: Vec<String> = members
+        .iter()
+        .map(|m| {
+            serde_json::to_string(&m.roles)
+                .map_err(|_| validation_failed("Failed to serialize member roles"))
+        })
+        .collect::<ModmailResult<Vec<_>>>()?;
+
+    // Pair each member with its serialized roles, then chunk. SQLite's default
+    // SQLITE_MAX_VARIABLE_NUMBER is 999; with 10 bind parameters per row we stay
+    // safely under that limit by capping each statement at 99 rows.
+    let pairs: Vec<(&TrackedMember, &String)> = members.iter().zip(roles_jsons.iter()).collect();
+    const CHUNK_SIZE: usize = 99;
+
+    let mut tx = pool.begin().await.map_err(|e| {
+        eprintln!("Failed to begin tracked members transaction: {e:?}");
+        validation_failed("Failed to begin tracked members transaction")
+    })?;
+
+    for chunk in pairs.chunks(CHUNK_SIZE) {
+        let mut builder = sqlx::QueryBuilder::new(
+            "INSERT INTO tracked_members \
+             (guild_id, user_id, username, global_name, nickname, avatar_url, roles, \
+              joined_at, first_seen_at, last_seen_at) ",
+        );
+
+        builder.push_values(chunk.iter(), |mut b, (member, roles_json)| {
+            b.push_bind(member.guild_id.clone())
+                .push_bind(member.user_id.clone())
+                .push_bind(member.username.clone())
+                .push_bind(member.global_name.clone())
+                .push_bind(member.nickname.clone())
+                .push_bind(member.avatar_url.clone())
+                .push_bind((*roles_json).clone())
+                .push_bind(member.joined_at)
+                .push_bind(member.first_seen_at)
+                .push_bind(member.last_seen_at);
+        });
+
+        builder.push(
+            " ON CONFLICT(guild_id, user_id) DO UPDATE SET \
+             username = excluded.username, \
+             global_name = excluded.global_name, \
+             nickname = excluded.nickname, \
+             avatar_url = excluded.avatar_url, \
+             roles = excluded.roles, \
+             joined_at = COALESCE(excluded.joined_at, tracked_members.joined_at), \
+             last_seen_at = excluded.last_seen_at",
+        );
+
+        builder.build().execute(&mut *tx).await.map_err(|e| {
+            eprintln!("Failed to bulk upsert tracked members chunk: {e:?}");
+            validation_failed("Failed to bulk upsert tracked members chunk")
+        })?;
+    }
+
+    tx.commit().await.map_err(|e| {
+        eprintln!("Failed to commit tracked members transaction: {e:?}");
+        validation_failed("Failed to commit tracked members transaction")
+    })?;
+
+    Ok(())
+}
+
 pub async fn get_tracked_member(
     guild_id: &str,
     user_id: &str,

crates/rustmail/src/db/operations/init.rs

@@ -1,6 +1,11 @@
-use sqlx::{SqlitePool, sqlite::SqlitePoolOptions};
+use sqlx::{
+    SqlitePool,
+    sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous},
+};
 use std::fs;
 use std::path::Path;
+use std::str::FromStr;
+use std::time::Duration;
 
 pub async fn init_database() -> Result<SqlitePool, sqlx::Error> {
     let db_path = "db/db.sqlite";
@@ -12,10 +17,14 @@ pub async fn init_database() -> Result<SqlitePool, sqlx::Error> {
         println!("Database file created at: {}", db_path);
     }
 
-    let db_url = format!("sqlite://{}", db_path);
+    let connect_options = SqliteConnectOptions::from_str(&format!("sqlite://{}", db_path))?
+        .journal_mode(SqliteJournalMode::Wal)
+        .synchronous(SqliteSynchronous::Normal)
+        .busy_timeout(Duration::from_secs(5));
+
     let pool = SqlitePoolOptions::new()
         .max_connections(30)
-        .connect(&db_url)
+        .connect_with(connect_options)
         .await?;
 
     sqlx::migrate!("../../migrations").run(&pool).await?;

crates/rustmail/src/handlers/guild_ban_handler.rs

@@ -3,7 +3,7 @@ use crate::prelude::config::*;
 use crate::prelude::db::*;
 use chrono::Utc;
 use serenity::all::audit_log::Action;
-use serenity::all::{Context, EventHandler, GuildId, Member, MemberAction, User};
+use serenity::all::{Context, EventHandler, GuildId, Member, MemberAction, User, UserId};
 use serenity::async_trait;
 use sqlx::SqlitePool;
 
@@ -174,3 +174,67 @@ impl EventHandler for GuildBanHandler {
         }
     }
 }
+
+pub async fn backfill_tracked_members(
+    ctx: &Context,
+    config: &Config,
+    shutdown: &mut tokio::sync::watch::Receiver<bool>,
+) {
+    const PAGE_LIMIT: u64 = 1000;
+
+    let Some(pool) = config.db_pool.as_ref() else {
+        return;
+    };
+
+    let guild_id = GuildId::new(config.bot.get_community_guild_id());
+    let now = Utc::now().timestamp();
+
+    let mut after: Option<UserId> = None;
+    let mut total: usize = 0;
+
+    loop {
+        if *shutdown.borrow() {
+            println!("Tracked member backfill cancelled due to shutdown.");
+            return;
+        }
+
+        let page = match guild_id.members(&ctx.http, Some(PAGE_LIMIT), after).await {
+            Ok(p) => p,
+            Err(e) => {
+                eprintln!("Failed to fetch guild members for backfill: {:?}", e);
+                return;
+            }
+        };
+
+        if page.is_empty() {
+            break;
+        }
+
+        let last_id = page.last().map(|m| m.user.id);
+        let page_len = page.len();
+
+        let tracked_batch: Vec<TrackedMember> =
+            page.iter().map(|m| member_to_tracked(m, now)).collect();
+
+        match bulk_upsert_tracked_members(&tracked_batch, pool).await {
+            Ok(()) => total += page_len,
+            Err(e) => {
+                eprintln!(
+                    "Failed to backfill tracked members page in guild {}: {:?}",
+                    guild_id, e
+                );
+            }
+        }
+
+        if page_len < PAGE_LIMIT as usize {
+            break;
+        }
+
+        after = last_id;
+    }
+
+    println!(
+        "Backfilled {} tracked members for community guild {}",
+        total, guild_id
+    );
+}

crates/rustmail/src/handlers/ready_handler.rs

@@ -1,4 +1,5 @@
 use crate::db::get_all_thread_status;
+use crate::handlers::guild_ban_handler::backfill_tracked_members;
 use crate::prelude::commands::*;
 use crate::prelude::config::*;
 use crate::prelude::features::*;
@@ -11,6 +12,7 @@ use serenity::{
 };
 use sqlx::SqlitePool;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 use tokio::sync::{Mutex, watch::Receiver};
 use tokio::time::interval;
@@ -21,6 +23,7 @@ pub struct ReadyHandler {
     pub registry: Arc<CommandRegistry>,
     pub shutdown: Arc<Receiver<bool>>,
     pub bot_state: Arc<Mutex<BotState>>,
+    backfill_started: Arc<AtomicBool>,
 }
 
 impl ReadyHandler {
@@ -35,6 +38,7 @@ impl ReadyHandler {
             registry,
             shutdown: Arc::new(shutdown),
             bot_state,
+            backfill_started: Arc::new(AtomicBool::new(false)),
         }
     }
 }
@@ -75,6 +79,22 @@ impl EventHandler for ReadyHandler {
             }
         });
 
+        if self
+            .backfill_started
+            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+            .is_ok()
+        {
+            tokio::spawn({
+                let ctx = ctx.clone();
+                let config = config.clone();
+                let mut shutdown = (*self.shutdown).clone();
+
+                async move {
+                    backfill_tracked_members(&ctx, &config, &mut shutdown).await;
+                }
+            });
+        }
+
         load_reminders(&ctx, &self.config, &pool.clone(), self.shutdown.clone()).await;
 
         update_threads_status(&ctx, &pool.clone());