feat(botinfra): aggregate snapshots for EventStore; vahter user streams use them

.github/workflows/vahter-build.yml

@@ -10,7 +10,8 @@ jobs:
     uses: ./.github/workflows/_bot-build.yml
     with:
       test-command: >-
+        dotnet test tests/BotInfra.Tests/BotInfra.Tests.fsproj -c Release --logger "trx;LogFileName=botinfra-test-results.trx" --logger "console;verbosity=detailed" &&
         dotnet test tests/VahterBanBot.Unit.Tests/VahterBanBot.Unit.Tests.fsproj -c Release --logger "trx;LogFileName=unit-test-results.trx" --logger "console;verbosity=detailed" &&
         dotnet test tests/SerializationCompat.Tests/SerializationCompat.Tests.fsproj -c Release --logger "trx;LogFileName=compat-test-results.trx" --logger "console;verbosity=detailed" &&
         dotnet test tests/VahterBanBot.Tests/VahterBanBot.Tests.fsproj -c Release --logger "trx;LogFileName=test-results.trx" --logger "console;verbosity=detailed"
-      bot-paths-regex: '^(src/(BotInfra|VahterBanBot)/|src/Dockerfile\.bot$|tests/(VahterBanBot\.Tests|VahterBanBot\.Unit\.Tests|SerializationCompat\.Tests|BotTestInfra|FakeTgApi|FakeAzureOcrApi)/|Directory\.Build\.props$|Directory\.Packages\.props$|global\.json$)'
+      bot-paths-regex: '^(src/(BotInfra|VahterBanBot)/|src/Dockerfile\.bot$|tests/(BotInfra\.Tests|VahterBanBot\.Tests|VahterBanBot\.Unit\.Tests|SerializationCompat\.Tests|BotTestInfra|FakeTgApi|FakeAzureOcrApi)/|Directory\.Build\.props$|Directory\.Packages\.props$|global\.json$)'

bots.slnx

@@ -9,6 +9,7 @@
   <Folder Name="/tests/">
     <Project Path="tests/AlitaBot.RealTests/AlitaBot.RealTests.fsproj" />
     <Project Path="tests/AlitaBot.Tests/AlitaBot.Tests.fsproj" />
+    <Project Path="tests/BotInfra.Tests/BotInfra.Tests.fsproj" />
     <Project Path="tests/BotTestInfra/BotTestInfra.fsproj" />
     <Project Path="tests/CouponHubBot.Ocr.Tests/CouponHubBot.Ocr.Tests.fsproj" />
     <Project Path="tests/CouponHubBot.RealTests/CouponHubBot.RealTests.fsproj" />

src/BotInfra/BotInfra.fsproj

@@ -8,6 +8,7 @@
     <Compile Include="Utils.fs" />
     <Compile Include="Time.fs" />
     <Compile Include="JsonSetup.fs" />
+    <Compile Include="EventSnapshot.fs" />
     <Compile Include="EventStore.fs" />
     <Compile Include="LiveOptions.fs" />
     <Compile Include="OcrService.fs" />

src/BotInfra/EVENTSTORE.md

@@ -146,10 +146,113 @@ the projection callback always reflects the actual append version.
 If the decider returns `[]`, no events are written and the projection is
 not invoked — events and projection cannot drift.
 
+## Snapshots (opt-in)
+
+Without a policy every load folds the entire stream. For long-lived aggregates
+(vahter's `user:*` streams run into thousands of events), pass a snapshot table
+and use the `SnapshotPolicy` overloads: state is loaded as *stored snapshot +
+the events after it*, in one statement.
+
+```sql
+CREATE TABLE event_snapshot (
+    stream_id       TEXT        NOT NULL,
+    state_type      TEXT        NOT NULL,
+    schema_version  INT         NOT NULL,
+    stream_version  INT         NOT NULL,
+    state           JSONB       NOT NULL,
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (stream_id, state_type)
+);
+GRANT SELECT, INSERT, UPDATE, DELETE ON event_snapshot TO <bot_service_role>;
+
+-- Snapshots are only valid for an append-only log: any in-place rewrite of events drops them.
+CREATE OR REPLACE FUNCTION event_snapshot_invalidate() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF TG_OP = 'TRUNCATE' THEN
+        TRUNCATE event_snapshot;
+    ELSE
+        DELETE FROM event_snapshot WHERE stream_id = OLD.stream_id OR stream_id = NEW.stream_id;
+    END IF;
+    RETURN NULL;
+END $$;
+CREATE OR REPLACE TRIGGER event_snapshot_invalidate AFTER UPDATE OR DELETE ON event
+    FOR EACH ROW EXECUTE FUNCTION event_snapshot_invalidate();
+CREATE OR REPLACE TRIGGER event_snapshot_invalidate_truncate AFTER TRUNCATE ON event
+    FOR EACH STATEMENT EXECUTE FUNCTION event_snapshot_invalidate();
+```
+
+```fsharp
+let store = EventStore(connString, "event", eventJsonOpts, "event_snapshot")
+
+type User =
+    ...
+    static member SnapshotPolicy = { StateType = "User"; SchemaVersion = 1; SnapshotEvery = 20 }
+
+let! (state, version) = EventStore.loadSnapshottedState<UserEvent, User> store User.SnapshotPolicy streamId
+let! (events, state)  = EventStore.appendSnapshottedEventWithProjection store User.SnapshotPolicy streamId decider
+// explicit fold/zero: store.LoadState / store.Transact / store.TransactWithProjection overloads
+```
+
+Rules the implementation guarantees:
+
+- **Snapshots are a cache, never the source of truth.** Truncating the table is
+  always safe; loads replay and re-create rows. A snapshot write failure is
+  recorded (`eventstore_snapshot_ops_total{op="write_failed"}` + span error) and
+  swallowed — it never fails a load, and on the append path it runs *after* the
+  commit so it can never roll back events.
+- **Stale snapshots are detected and discarded.** A row whose `schema_version`
+  differs from the policy is ignored (and overwritten by the next write). A row
+  that is ahead of its log or doesn't deserialize is ignored, the stream is
+  replayed and the row force-overwritten (`op="discarded_ahead" |
+  "discarded_unreadable"`); a failure there is recorded, never thrown.
+- **Rewriting history drops snapshots.** The triggers above delete a stream's
+  snapshots on any `UPDATE`/`DELETE` of its events (and `TRUNCATE` clears the
+  table), so a data-fix migration can never leave a snapshot that disagrees
+  with the rewritten log. The bot role never updates events, so the triggers
+  only fire during migrations.
+- **Snapshots only move forward** within a schema version (upsert guard), so a
+  slow writer can't regress a newer snapshot. A different schema version (new
+  code or a rollback) always overwrites.
+- **Consistent reads.** Snapshot and tail come from one SQL statement, so a
+  concurrent snapshot rewrite can't make the tail skip events. Optimistic
+  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.
+- **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
+  publishes into the cache if no append invalidated the stream since it read
+  the DB, so a slow parallel load can't put pre-append state back. A disposed
+  scope stops caching, so fire-and-forget work that captured it reads the DB.
+- **Known costs.** While pods with two different `SchemaVersion`s run side by
+  side (rolling deploy of a bump), each treats the other's row as a miss, so
+  long streams replay until the rollout finishes. `TryAppend` with an
+  `expectedVersion` above the head can create a version gap; the Transact
+  loops never do — prefer them.
+
+### Versioning — the one thing you must not forget
+
+A snapshot bakes in two things: the state type's *shape* and the *result of
+its fold*. Bump `SchemaVersion` on any change to either — a new field silently
+deserializes to its default, and a fixed fold bug stays baked into old
+snapshots. Enforce it with two pin tests next to the aggregate (see
+`tests/VahterBanBot.Unit.Tests/UserSnapshotTests.fs`):
+
+1. `SnapshotShape.describe typeof<State>` pinned per schema version.
+2. A fold *transcript* pinned per schema version: the snapshot JSON of every
+   intermediate state of a canonical sequence (every event case plus edge
+   cases), and of each event applied to `Zero` and to a fully-populated state.
+   Pinning only a final state misses changes whose effect is overwritten later;
+   pinning JSON also catches serializer-option changes.
+
+Both assert the *latest* pin's version equals the policy's, so changing the
+type or the fold fails until the version is bumped and a new pin appended.
+Also keep a JSON round-trip test and a "snapshot + tail = full replay at every
+split point" test.
+
 ## What this wrapper does NOT do
 
-- **Snapshots / state caching.** Every read folds the entire stream. If you
-  have very long streams, project to a separate read model.
 - **Cross-stream transactions.** One stream per call. If you need to write
   to two streams atomically, model them as one stream.
 - **Schema migrations** of payload shapes. If you change a DU case, write a

src/BotInfra/EventSnapshot.fs

@@ -0,0 +1,65 @@
+namespace BotInfra
+
+open System
+open System.Collections.Generic
+open Microsoft.FSharp.Reflection
+
+/// Opt-in `EventStore` snapshotting: state = stored snapshot + events after it. Snapshots are a
+/// disposable cache — deleting them is always safe, loads fall back to a full replay.
+type SnapshotPolicy =
+    { /// Aggregate discriminator, part of the snapshot key (one stream may carry several folds).
+      StateType: string
+      /// Must be bumped on ANY change to the state type's shape OR to its fold semantics;
+      /// snapshots written under another version are ignored and rebuilt by replay.
+      SchemaVersion: int
+      /// A snapshot is (re)written once at least this many events sit past the stored one.
+      SnapshotEvery: int }
+
+/// Deterministic structure of an F# type, for pinning a snapshotted state's shape to its
+/// `SnapshotPolicy.SchemaVersion` in a test. Anonymous records render their fields only.
+module SnapshotShape =
+
+    let private isAnonymousRecord (t: Type) = t.Name.Contains "AnonymousType"
+
+    let rec private render (visiting: HashSet<Type>) (t: Type) : string =
+        if t.IsGenericType && t.GetGenericTypeDefinition() = typedefof<option<_>> then
+            $"option<{render visiting (t.GetGenericArguments()[0])}>"
+        elif t.IsGenericType && t.GetGenericTypeDefinition() = typedefof<list<_>> then
+            $"list<{render visiting (t.GetGenericArguments()[0])}>"
+        elif t.IsGenericType && t.GetGenericTypeDefinition() = typedefof<Nullable<_>> then
+            $"nullable<{render visiting (t.GetGenericArguments()[0])}>"
+        elif t.IsArray then
+            $"array<{render visiting (t.GetElementType())}>"
+        elif FSharpType.IsTuple t then
+            FSharpType.GetTupleElements t |> Array.map (render visiting) |> String.concat " * " |> sprintf "(%s)"
+        elif FSharpType.IsRecord(t, true) || FSharpType.IsUnion(t, true) then
+            let name = if isAnonymousRecord t then "" else t.Name
+            if not (visiting.Add t) then $"@{name}"
+            else
+                let fields (ps: Reflection.PropertyInfo[]) =
+                    ps |> Array.map (fun p -> $"{p.Name}: {render visiting p.PropertyType}") |> String.concat "; "
+                let body =
+                    if FSharpType.IsRecord(t, true) then
+                        $"{{{fields (FSharpType.GetRecordFields(t, true))}}}"
+                    else
+                        FSharpType.GetUnionCases(t, true)
+                        |> Array.map (fun c ->
+                            match c.GetFields() with
+                            | [||] -> c.Name
+                            | fs -> $"{c.Name} of ({fields fs})")
+                        |> String.concat " | "
+                        |> sprintf "[%s]"
+                %visiting.Remove t
+                name + body
+        elif t.IsGenericType then
+            let name = t.Name.Substring(0, t.Name.IndexOf '`')
+            let args = t.GetGenericArguments() |> Array.map (render visiting) |> String.concat ", "
+            $"{name}<{args}>"
+        elif t.IsEnum then
+            let names = Enum.GetNames t |> String.concat ","
+            $"{t.Name}<{names}>"
+        else
+            t.FullName
+
+    /// Renders `t`'s structure; nested records/unions are expanded, recursion is cut with `@Name`.
+    let describe (t: Type) : string = render (HashSet<Type>()) t