Enable Jetstream v2

CarpaNet.Samples.slnx

@@ -10,6 +10,7 @@
     <Project Path="samples/AuthTest/AuthTest.csproj" />
     <Project Path="samples/FirehoseTest/FirehoseTest.csproj" />
     <Project Path="samples/JetstreamTest/JetstreamTest.csproj" />
+    <Project Path="samples/JetstreamV2Test/JetstreamV2Test.csproj" />
     <Project Path="samples/RemoteResolution/RemoteResolution.csproj" />
     <Project Path="samples/XrpcServer/XrpcServer.csproj" />
     <Project Path="samples/MebiByteTest/MebiByteTest.csproj" />

docs/docs/jetstream.md

@@ -3,6 +3,69 @@
 Jetstream provides a lightweight, JSON-based WebSocket event stream. Requires the `CarpaNet.Jetstream` package.
 This is useful when you only care for a type of collection that you want to respond to. It also uses far less data than the full Firehose.
 
+The package contains two clients:
+
+- **`JetstreamV2Client`** — the current Jetstream service ([bluesky-social/jetstream](https://github.com/bluesky-social/jetstream)): the `network.bsky.jetstream.subscribeEvents` websocket plus a full-network sealed archive you can replay over HTTP. Use this for new code.
+- **`JetstreamClient`** — the legacy v1 `/subscribe` wire, kept for servers that only expose it.
+
+## Jetstream v2
+
+`JetstreamV2Client.SubscribeAsync` is a managed stream: it reconnects with exponential backoff, resumes at the last delivered seq, deduplicates the at-least-once overlap, fetches and rotates the zstd compression dictionary, and, when you ask for history, downloads the sealed archive over HTTP and cuts over to the live tail with no gap.
+
+```csharp
+using CarpaNet;
+using CarpaNet.Jetstream;
+
+using var client = new JetstreamV2Client(
+    new Uri(BlueskyServices.JetstreamUsEast),
+    new JetstreamV2ClientOptions
+    {
+        EnableCompression = true,   // dict-zstd; the dictionary is fetched automatically
+    });
+
+var options = new JetstreamV2SubscribeOptions
+{
+    Collections = new[] { "app.bsky.feed.post" },  // exact NSIDs or wildcards like app.bsky.feed.*
+    // LiveCursor = lastSeq,     // resume a live tail from a saved seq
+    // AfterSeq = 0,             // or: replay the whole archive first, then go live
+    // SnapshotOnly = true,      // or: archive dump only, no websocket
+};
+
+await foreach (var evt in client.SubscribeAsync(options))
+{
+    switch (evt.Kind)
+    {
+        case JetstreamV2EventKind.Commit when evt.Commit is { } commit:
+            Console.WriteLine($"[{commit.Operation}] seq={evt.Seq} {commit.Collection}/{commit.Rkey}");
+            // Typed records via your generated JSON context:
+            // var post = commit.GetRecord(ATProtoJsonContext.Default.AppBskyFeedPost);
+            break;
+
+        case JetstreamV2EventKind.Identity when evt.Identity is { } identity:
+            Console.WriteLine($"[Identity] {evt.Did} → {identity.Handle}");
+            break;
+
+        case JetstreamV2EventKind.Account when evt.Account is { } account:
+            Console.WriteLine($"[Account] {evt.Did} active={account.Active} status={account.Status}");
+            break;
+
+        case JetstreamV2EventKind.Sync when evt.Sync is { } sync:
+            Console.WriteLine($"[Sync] {evt.Did} rev={sync.Rev}");
+            break;
+    }
+}
+```
+
+Things worth knowing:
+
+- `evt.Seq` is the cursor: Persist the last seen value and pass it back as `LiveCursor` (pure live) or `AfterSeq` (archive replay) to resume. Delivery is at-least-once across restarts — fold idempotently or dedup by seq.
+- Collection filters never suppress `#identity`/`#account`/`#sync`: those DID-level events are your only signal to purge a deleted account's records. Ask for `Kinds = [JetstreamV2EventKind.Commit]` explicitly if you want commits only.
+- Filters are immutable per connection on the v2 wire; there is no `options_update`. Start a new subscription to change them.
+- `CursorTooOld`: on a backfill-enabled stream the client automatically re-enters archive replay; on a pure live tail it throws `JetstreamV2Exception` — resume with `AfterSeq` instead.
+- The archive endpoints (`PlanSnapshotAsync`, `GetSegmentAsync`, `GetBlockAsync`, decoded via `JetstreamSegmentFormat`) are also public for direct use, and `JetstreamV2ClientOptions.ApiKey` sends a bearer key on them (never on the public websocket or dictionary fetch).
+
+## Jetstream v1 (legacy)
+
 ```csharp
 using CarpaNet.Jetstream;
 
@@ -42,11 +105,14 @@ await foreach (var evt in client.SubscribeAsync(options))
 }
 ```
 
-## Dynamic Filter Updates
+### Dynamic Filter Updates (v1 only)
 
 ```csharp
 await client.SendOptionsUpdateAsync(new JetstreamOptionsUpdate
 {
-    WantedCollections = new[] { "app.bsky.graph.follow" },
+    Payload = new JetstreamOptionsPayload
+    {
+        WantedCollections = new List<string> { "app.bsky.graph.follow" },
+    },
 });
 ```

samples/JetstreamV2Test/JetstreamV2Test.csproj

@@ -0,0 +1,20 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>net10.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+    <PublishAot>true</PublishAot>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="../../src/CarpaNet.Jetstream/CarpaNet.Jetstream.csproj" />
+    <ProjectReference Include="..\..\src\CarpaNet\CarpaNet.csproj" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Logging.Console" />
+  </ItemGroup>
+
+</Project>

samples/JetstreamV2Test/Program.cs

@@ -0,0 +1,176 @@
+using CarpaNet;
+using CarpaNet.Jetstream;
+using Microsoft.Extensions.Logging;
+
+using var cts = new CancellationTokenSource();
+Console.CancelKeyPress += (_, e) =>
+{
+    e.Cancel = true;
+    cts.Cancel();
+    Console.WriteLine();
+    Console.WriteLine("Shutting down...");
+};
+
+// Parse arguments
+long? liveCursor = null;
+long? afterSeq = null;
+long? beforeSeq = null;
+bool snapshotOnly = false;
+bool compress = false;
+bool verbose = false;
+var collections = new List<string>();
+var dids = new List<string>();
+var kinds = new List<JetstreamV2EventKind>();
+string endpoint = BlueskyServices.JetstreamUsEast;
+string? apiKey = Environment.GetEnvironmentVariable("JETSTREAM_CLIENT_API_KEY");
+
+for (int i = 0; i < args.Length; i++)
+{
+    switch (args[i])
+    {
+        case "--cursor" when i + 1 < args.Length && long.TryParse(args[i + 1], out var c):
+            liveCursor = c;
+            i++;
+            break;
+        case "--after-seq" when i + 1 < args.Length && long.TryParse(args[i + 1], out var a):
+            afterSeq = a;
+            i++;
+            break;
+        case "--before-seq" when i + 1 < args.Length && long.TryParse(args[i + 1], out var b):
+            beforeSeq = b;
+            i++;
+            break;
+        case "--snapshot-only":
+            snapshotOnly = true;
+            break;
+        case "--collection" when i + 1 < args.Length:
+            collections.Add(args[i + 1]);
+            i++;
+            break;
+        case "--did" when i + 1 < args.Length:
+            dids.Add(args[i + 1]);
+            i++;
+            break;
+        case "--kind" when i + 1 < args.Length && Enum.TryParse<JetstreamV2EventKind>(args[i + 1], ignoreCase: true, out var k):
+            kinds.Add(k);
+            i++;
+            break;
+        case "--endpoint" when i + 1 < args.Length:
+            endpoint = args[i + 1];
+            i++;
+            break;
+        case "--api-key" when i + 1 < args.Length:
+            apiKey = args[i + 1];
+            i++;
+            break;
+        case "--compress":
+            compress = true;
+            break;
+        case "--verbose":
+            verbose = true;
+            break;
+        case "--help":
+            Console.WriteLine("""
+                CarpaNet Jetstream v2 sample.
+
+                Live tail:          JetstreamV2Test [--cursor <seq>]
+                Full replay then live:        JetstreamV2Test --after-seq 0
+                Point-in-time snapshot:       JetstreamV2Test --after-seq 0 [--before-seq <seq>] --snapshot-only
+
+                Options:
+                  --collection <nsid>   Filter to a collection
+                  --did <did>           Filter to a repo DID
+                  --kind <kind>         Filter to commit|identity|account|sync
+                  --endpoint <url>      Jetstream v2 instance (default: jetstream.us-east.bsky.network)
+                  --api-key <key>       Archive API key (or JETSTREAM_CLIENT_API_KEY env var)
+                  --compress            Enable dict-zstd live-tail compression
+                  --verbose             Enable client diagnostics logging
+                """);
+            return;
+    }
+}
+
+Console.WriteLine("=== CarpaNet Jetstream v2 Test ===");
+Console.WriteLine($"Endpoint: {endpoint}");
+if (afterSeq.HasValue)
+{
+    Console.WriteLine(snapshotOnly
+        ? $"Mode: archive snapshot ({afterSeq}, {(beforeSeq.HasValue ? beforeSeq.ToString() : "sealed tip")}]"
+        : $"Mode: backfill from seq {afterSeq}, then live");
+}
+else
+{
+    Console.WriteLine(liveCursor.HasValue ? $"Mode: live from cursor {liveCursor}" : "Mode: live from the current tip");
+}
+
+Console.WriteLine("Press Ctrl+C to stop.");
+Console.WriteLine();
+
+using var loggerFactory = verbose
+    ? LoggerFactory.Create(builder => builder.AddSimpleConsole(o => o.SingleLine = true).SetMinimumLevel(LogLevel.Information))
+    : null;
+
+using var client = new JetstreamV2Client(new Uri(endpoint), new JetstreamV2ClientOptions
+{
+    EnableCompression = compress,
+    ApiKey = apiKey,
+    LoggerFactory = loggerFactory,
+});
+
+var options = new JetstreamV2SubscribeOptions
+{
+    LiveCursor = liveCursor,
+    AfterSeq = afterSeq,
+    BeforeSeq = beforeSeq,
+    SnapshotOnly = snapshotOnly,
+    Collections = collections.Count > 0 ? collections : null,
+    Dids = dids.Count > 0 ? dids : null,
+    Kinds = kinds.Count > 0 ? kinds : null,
+};
+
+long count = 0;
+long lastSeq = 0;
+try
+{
+    await foreach (var evt in client.SubscribeAsync(options, cts.Token))
+    {
+        count++;
+        lastSeq = evt.Seq;
+        switch (evt.Kind)
+        {
+            case JetstreamV2EventKind.Commit when evt.Commit != null:
+                var commit = evt.Commit;
+                Console.WriteLine($"[Commit seq={evt.Seq}] {commit.Operation} {commit.Collection}/{commit.Rkey} from {evt.Did}");
+                if (commit.Record.HasValue &&
+                    commit.Record.Value.TryGetProperty("$type", out var typeEl))
+                {
+                    Console.WriteLine($"  record $type={typeEl.GetString()} cid={commit.Cid}");
+                }
+
+                break;
+
+            case JetstreamV2EventKind.Identity when evt.Identity != null:
+                Console.WriteLine($"[Identity seq={evt.Seq}] did={evt.Did} handle={evt.Identity.Handle}");
+                break;
+
+            case JetstreamV2EventKind.Account when evt.Account != null:
+                Console.WriteLine($"[Account seq={evt.Seq}] did={evt.Did} active={evt.Account.Active} status={evt.Account.Status}");
+                break;
+
+            case JetstreamV2EventKind.Sync when evt.Sync != null:
+                Console.WriteLine($"[Sync seq={evt.Seq}] did={evt.Did} rev={evt.Sync.Rev}");
+                break;
+        }
+    }
+}
+catch (OperationCanceledException)
+{
+    // Expected on Ctrl+C
+}
+catch (JetstreamV2Exception ex)
+{
+    Console.WriteLine($"Stream failed{(ex.ErrorName != null ? $" ({ex.ErrorName})" : string.Empty)}: {ex.Message}");
+}
+
+Console.WriteLine();
+Console.WriteLine($"Stream ended after {count} events; last seq {lastSeq} (pass it back with --cursor or --after-seq to resume).");

samples/JetstreamV2Test/README.md

@@ -0,0 +1,51 @@
+# JetstreamV2Test
+
+Demonstrates `JetstreamV2Client`: the managed Jetstream v2 stream that welds sealed-archive
+replay (planSnapshot → getSegment/getBlock over HTTP) and the live
+`network.bsky.jetstream.subscribeEvents` websocket into one seq-ordered event stream with
+automatic reconnect, resume, and compression handling.
+
+## Usage
+
+```bash
+# Live tail from the current tip
+dotnet run
+
+# Live tail, filtered, with dict-zstd compression (dictionary fetched automatically)
+dotnet run -- --collection app.bsky.feed.post --compress
+
+# Resume a live tail from a saved seq
+dotnet run -- --cursor 24959600000
+
+# Replay sealed history from a seq, then cut over to live with no gap
+dotnet run -- --after-seq 24959600000 --collection app.bsky.feed.post
+
+# Point-in-time archive snapshot (no websocket)
+dotnet run -- --after-seq 1000000 --before-seq 1000400 --snapshot-only
+```
+
+Options:
+
+| Flag | Meaning |
+| --- | --- |
+| `--collection <nsid>` | Filter commits to a collection (repeatable; wildcards like `app.bsky.feed.*`) |
+| `--did <did>` | Filter to a repo DID (repeatable) |
+| `--kind <kind>` | Filter to `commit`, `identity`, `account`, or `sync` (repeatable) |
+| `--cursor <seq>` | Resume a pure live tail from a saved seq |
+| `--after-seq <seq>` | Replay the sealed archive after this seq, then go live (0 = everything) |
+| `--before-seq <seq>` | Snapshot upper bound (requires `--snapshot-only`) |
+| `--snapshot-only` | Archive dump only; the stream ends at the sealed tip |
+| `--compress` | Enable dict-zstd live-tail compression |
+| `--endpoint <url>` | Jetstream v2 instance (default: `jetstream.us-east.bsky.network`) |
+| `--api-key <key>` | Archive API key (also read from `JETSTREAM_CLIENT_API_KEY`) |
+| `--verbose` | Show client diagnostics (reconnects, sweeps, degraded compression) |
+
+The final line prints the last delivered seq — pass it back with `--cursor` (live) or
+`--after-seq` (replay) to resume. Delivery is at-least-once across restarts; consumers fold
+idempotently or dedup by seq.
+
+## Jetstream v2 vs v1
+
+The v2 wire (`JetstreamV2Client`) adds a durable per-event `seq` cursor, an archive you can
+replay from seq 0, a `sync` event kind, and self-describing frames. The v1 `/subscribe` wire
+(`JetstreamClient`, see the `JetstreamTest` sample) remains for legacy servers.