diff --git a/src/BotInfra/EVENTSTORE.md b/src/BotInfra/EVENTSTORE.md index 585544db..b0d603bd 100644 --- a/src/BotInfra/EVENTSTORE.md +++ b/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 diff --git a/src/BotInfra/EventStore.fs b/src/BotInfra/EventStore.fs index dded4658..efb518e1 100644 --- a/src/BotInfra/EventStore.fs +++ b/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(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> diff --git a/src/VahterBanBot/DB.fs b/src/VahterBanBot/DB.fs index 4e799ce9..5d113825 100644 --- a/src/VahterBanBot/DB.fs +++ b/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 = + member _.RebuildSnapshots(?batchSize: int, ?onProgress: string -> unit, ?usersOnly: bool) : Task = 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( - "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( + "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 } diff --git a/src/VahterBanBot/Program.fs b/src/VahterBanBot/Program.fs index 82743678..fb976a55 100644 --- a/src/VahterBanBot/Program.fs +++ b/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>(fun ctx -> task { if not (WebhookHost.validateApiKey webhookCfg.SecretToken ctx) then return Results.Text("Access Denied", statusCode = 401) else - let db = ctx.RequestServices.GetRequiredService() - let logger = ctx.RequestServices.GetRequiredService>() - 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() + let logger = ctx.RequestServices.GetRequiredService>() + 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 diff --git a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs index 71fafd36..a83c923b 100644 --- a/tests/BotInfra.Tests/EventStoreSnapshotTests.fs +++ b/tests/BotInfra.Tests/EventStoreSnapshotTests.fs @@ -559,6 +559,33 @@ type EventStoreSnapshotTests(db: PostgresFixture) = Assert.Equal(12, expected.Total) } + [] + 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) + } + + [] + 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) + } + [] member _.``snapshot policy requires a snapshot table``() = let plain = EventStore(db.ConnectionString, "event", jsonOpts) diff --git a/tests/VahterBanBot.Tests/SnapshotTests.fs b/tests/VahterBanBot.Tests/SnapshotTests.fs index 92078d05..90439401 100644 --- a/tests/VahterBanBot.Tests/SnapshotTests.fs +++ b/tests/VahterBanBot.Tests/SnapshotTests.fs @@ -107,6 +107,34 @@ type SnapshotTests(fixture: MlDisabledVahterTestContainers) = Assert.Equal("rebuild me one", snap2.Value.text) } + [] + 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(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(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.