feat(vahter): /rebuild-snapshots backfills aggregate user snapshots (?scope=users)

src/BotInfra/EVENTSTORE.md

@@ -218,7 +218,10 @@ Rules the implementation guarantees:
   concurrency is unchanged: expected version = snapshot version + tail length.
 - **Write cadence.** A snapshot is written when a load or a committed append
   leaves at least `SnapshotEvery` events past the stored one — frozen streams
-  get snapshotted lazily on their first slow load, no backfill needed.
+  get snapshotted lazily on their first slow load. To backfill or repair
+  eagerly, `store.RebuildSnapshot` replays the full log (ignoring the stored
+  snapshot) and force-writes it; vahter exposes this as
+  `POST /rebuild-snapshots?scope=users`.
 - **Request scope.** `BeginRequestScope` caches snapshot-loaded states next to
   raw streams; every committed append (through any API) drops both cached views
   of that stream, then the appending path re-populates its own. A load only

src/BotInfra/EventStore.fs

@@ -681,6 +681,27 @@ ORDER BY stream_id, stream_version
             return unbox<'State> loaded.State, loaded.Version
         }
 
+    /// Replays the whole log (ignoring any stored snapshot) and force-writes the snapshot when the
+    /// stream has at least `policy.SnapshotEvery` events — backfill and repair for admin rebuilds.
+    member _.RebuildSnapshot<'TEvent, 'State>
+            (fold: 'State -> 'TEvent -> 'State, zero: 'State, policy: SnapshotPolicy, streamId: string)
+            : Task<'State * int> =
+        task {
+            validatePolicy policy
+            use activity = EventStoreTelemetry.activitySource.StartActivity("eventStore.snapshotRebuild")
+            use conn = new NpgsqlConnection(connString)
+            let! raws = conn.QueryAsync<RawEvent>(selectAllSql, {| streamId = streamId |})
+            let raws = List.ofSeq raws
+            let state =
+                raws |> List.fold (fun s r -> fold s (JsonSerializer.Deserialize<'TEvent>(r.data, jsonOptions))) zero
+            let version = raws |> List.tryLast |> Option.map _.stream_version |> Option.defaultValue 0
+            if version >= policy.SnapshotEvery then
+                let! _ = tryWriteSnapshot conn activity policy true streamId version state
+                ()
+            cacheEvict streamId
+            return state, version
+        }
+
     /// Snapshot-aware `TransactWithProjection`. The snapshot refresh runs after the commit, so a
     /// failing snapshot write can never roll back the appended events.
     member this.TransactWithProjection<'TEvent, 'State>

src/VahterBanBot/DB.fs

@@ -1383,9 +1383,10 @@ RETURNING job_name;
     /// recomputed from both its streams, so it serializes safely with concurrent live writes.
     /// Paged by stream-id prefix (keyset over idx_event_stream). `onProgress` (optional) is called at
     /// phase boundaries and every ~5000 streams. Returns the number of streams processed.
-    member _.RebuildSnapshots(?batchSize: int, ?onProgress: string -> unit) : Task<int> =
+    member _.RebuildSnapshots(?batchSize: int, ?onProgress: string -> unit, ?usersOnly: bool) : Task<int> =
         let batch = defaultArg batchSize 500
         let report = defaultArg onProgress ignore
+        let usersOnly = defaultArg usersOnly false
         let reportEvery = 5000
         task {
             let mutable grandTotal = 0
@@ -1439,31 +1440,32 @@ RETURNING job_name;
             do! rebuildPrefix "user:%" "users" countDistinctSql (fun sid conn tx ->
                 task {
                     let userId = sid.Substring("user:".Length) |> int64
-                    let! state = store.FoldEvents((fun s e -> User.Fold(s, e)), User.Zero, sid)
+                    let! (state, _) = store.RebuildSnapshot((fun s e -> User.Fold(s, e)), User.Zero, User.SnapshotPolicy, sid)
                     let json = JsonSerializer.Serialize(userSnapshot { state with Id = userId }, snapshotJsonOpts)
                     do! upsertUserSnapshot userId json conn tx
                 } :> Task)
 
-            // Each message row is recomputed from BOTH streams by upsertMessageSnapshot.
-            // one MessageReceived per message stream — cheaper than count(DISTINCT) via the type index.
-            do! rebuildPrefix "message:%" "messages" "SELECT count(*) FROM event WHERE event_type = 'MessageReceived'" (fun sid conn tx ->
-                task {
-                    let chatId, messageId = parseChatMsg sid
-                    do! upsertMessageSnapshot chatId messageId conn tx
-                } :> Task)
-
-            // Moderation streams whose message sibling exists were already covered above; only handle
-            // orphan moderation (a bot/vahter action on a message that was never recorded).
-            do! rebuildPrefix "moderation:%" "moderation (orphans)" countDistinctSql (fun sid conn tx ->
-                task {
-                    let chatId, messageId = parseChatMsg sid
-                    let! hasMsg =
-                        conn.ExecuteScalarAsync<bool>(
-                            "SELECT EXISTS(SELECT 1 FROM event WHERE stream_id = @sid)",
-                            {| sid = $"message:{chatId}:{messageId}" |}, tx)
-                    if not hasMsg then
+            if not usersOnly then
+                // Each message row is recomputed from BOTH streams by upsertMessageSnapshot.
+                // one MessageReceived per message stream — cheaper than count(DISTINCT) via the type index.
+                do! rebuildPrefix "message:%" "messages" "SELECT count(*) FROM event WHERE event_type = 'MessageReceived'" (fun sid conn tx ->
+                    task {
+                        let chatId, messageId = parseChatMsg sid
                         do! upsertMessageSnapshot chatId messageId conn tx
-                } :> Task)
+                    } :> Task)
+
+                // Moderation streams whose message sibling exists were already covered above; only handle
+                // orphan moderation (a bot/vahter action on a message that was never recorded).
+                do! rebuildPrefix "moderation:%" "moderation (orphans)" countDistinctSql (fun sid conn tx ->
+                    task {
+                        let chatId, messageId = parseChatMsg sid
+                        let! hasMsg =
+                            conn.ExecuteScalarAsync<bool>(
+                                "SELECT EXISTS(SELECT 1 FROM event WHERE stream_id = @sid)",
+                                {| sid = $"message:{chatId}:{messageId}" |}, tx)
+                        if not hasMsg then
+                            do! upsertMessageSnapshot chatId messageId conn tx
+                    } :> Task)
 
             return grandTotal
         }

src/VahterBanBot/Program.fs

@@ -318,20 +318,25 @@ SettingsDump.mapConfigDumpEndpoint
         SettingsDump.toJson eventJsonOpts live)
     app
 
-// One-off backfill of the snapshot_* read models from the event log. Idempotent; run manually
-// after deploy. Not auto-run on boot — the event table is too large to rescan every start.
+// Backfill/repair of the snapshot_* read models and aggregate snapshots from the event log.
+// Idempotent; run manually. `?scope=users` limits it to user streams (skips the message pass).
 %app.MapPost("/rebuild-snapshots", Func<HttpContext, Task<IResult>>(fun ctx ->
     task {
         if not (WebhookHost.validateApiKey webhookCfg.SecretToken ctx) then
             return Results.Text("Access Denied", statusCode = 401)
         else
-            let db = ctx.RequestServices.GetRequiredService<DbService>()
-            let logger = ctx.RequestServices.GetRequiredService<ILogger<Root>>()
-            logger.LogInformation "Snapshot rebuild starting"
-            let onProgress (msg: string) = logger.LogInformation("Snapshot rebuild progress: {Progress}", msg)
-            let! count = db.RebuildSnapshots(onProgress = onProgress)
-            logger.LogInformation("Snapshot rebuild done: {Count} streams", count)
-            return Results.Ok $"Rebuilt {count} snapshots"
+            match string ctx.Request.Query["scope"] with
+            | "" | "users" as scope ->
+                let db = ctx.RequestServices.GetRequiredService<DbService>()
+                let logger = ctx.RequestServices.GetRequiredService<ILogger<Root>>()
+                let usersOnly = scope = "users"
+                logger.LogInformation("Snapshot rebuild starting (usersOnly={UsersOnly})", usersOnly)
+                let onProgress (msg: string) = logger.LogInformation("Snapshot rebuild progress: {Progress}", msg)
+                let! count = db.RebuildSnapshots(onProgress = onProgress, usersOnly = usersOnly)
+                logger.LogInformation("Snapshot rebuild done: {Count} streams", count)
+                return Results.Ok $"Rebuilt {count} snapshots"
+            | other ->
+                return Results.BadRequest $"unknown scope '{other}' (expected: users)"
     }))
 
 // Main webhook endpoint with bot-specific update handling

tests/BotInfra.Tests/EventStoreSnapshotTests.fs

@@ -559,6 +559,33 @@ type EventStoreSnapshotTests(db: PostgresFixture) =
         Assert.Equal(12, expected.Total)
     }
 
+    [<Fact>]
+    member _.``rebuilding a snapshot replays the whole log and repairs a wrong snapshot at the same version``() = task {
+        let sid = newStream ()
+        do! appendRaw sid (added 25)
+        let! _ = load (policy 1) sid
+        do! exec """UPDATE event_snapshot SET state = jsonb_set(state, '{Total}', '1000') WHERE stream_id = @streamId""" sid
+
+        let! (state, version) = store.RebuildSnapshot(fold, Tally.Zero, policy 20, sid)
+        let! expected = replay sid
+        Assert.Equal(expected, state)
+        Assert.Equal(25, version)
+        do! assertSnapshotMatchesLog sid
+        let! (loaded, _) = load (policy 20) sid
+        Assert.Equal(expected, loaded)
+    }
+
+    [<Fact>]
+    member _.``rebuilding skips streams shorter than SnapshotEvery``() = task {
+        let sid = newStream ()
+        do! appendRaw sid (added 3)
+        let! (state, version) = store.RebuildSnapshot(fold, Tally.Zero, policy 5, sid)
+        Assert.Equal(6, state.Total)
+        Assert.Equal(3, version)
+        let! row = snapshotRow "event_snapshot" sid
+        Assert.Equal(None, row)
+    }
+
     [<Fact>]
     member _.``snapshot policy requires a snapshot table``() =
         let plain = EventStore(db.ConnectionString, "event", jsonOpts)

tests/VahterBanBot.Tests/SnapshotTests.fs

@@ -107,6 +107,34 @@ type SnapshotTests(fixture: MlDisabledVahterTestContainers) =
         Assert.Equal("rebuild me one", snap2.Value.text)
     }
 
+    [<Fact>]
+    let ``rebuild with scope=users backfills aggregate user snapshots and leaves message snapshots alone`` () = task {
+        let userId, chat, mId = 9_700_001L, -97100L, 1
+        let sid = $"user:{userId}"
+        do! fixture.InsertRawEvent(sid, 1, UsernameChanged {| userId = userId; username = Some "heavy" |}, t 0)
+        for v in 2 .. 26 do
+            do! fixture.InsertRawEvent(sid, v,
+                    UserReactionRecorded {| userId = userId; chatId = Some chat; messageId = Some v; emoji = Some "🔥"; delta = 1 |}, t v)
+        do! fixture.InsertRawEvent($"message:{chat}:{mId}", 1, received chat mId userId, t 30)
+
+        let! resp = fixture.BotHttp.PostAsync("/rebuild-snapshots?scope=users", null)
+        Assert.Equal(HttpStatusCode.OK, resp.StatusCode)
+
+        use conn = new Npgsql.NpgsqlConnection(fixture.DbConnectionString)
+        let! row =
+            Dapper.SqlMapper.QuerySingleOrDefaultAsync<struct (int * string)>(conn,
+                "SELECT stream_version, state::TEXT FROM event_snapshot WHERE stream_id = @sid AND state_type = 'User'",
+                {| sid = sid |})
+        let struct (version, state) = row
+        Assert.Equal(26, version)
+        let user = JsonSerializer.Deserialize<User>(state, eventJsonOpts)
+        Assert.Equal(25, user.ReactionCount)
+        Assert.Equal(Some "heavy", user.Username)
+
+        let! msgSnap = fixture.TryGetSnapshotMessage(chat, mId)
+        Assert.True(msgSnap.IsNone, "scope=users must not run the message phase")
+    }
+
     // ---- spam_status folded from BOTH streams (controlled-timeline raw events + rebuild) ----
     // The pure verdict logic is covered by Message.FoldTimeline tests; these verify the snapshot
     // projection/rebuild actually fold both streams into snapshot_message.spam_status.