jetstream client
atproto jetstream client
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429package main
import ( "context" _ "embed" "encoding/binary" "encoding/json" "fmt" "net/http" "net/http/httptest" "os" "os/exec" "reflect" "strings" "sync" "time"
js "github.com/bluesky-social/jetstream" "github.com/coder/websocket" "github.com/klauspost/compress/zstd")
//go:embed live.dictvar dictionary []byte
type fixture struct { mode string frames [][]byte mu sync.Mutex dicts, connections int didCount int cursors []string}
func (f *fixture) ServeHTTP(w http.ResponseWriter, r *http.Request) { if f.serveArchive(w, r) { return } switch r.URL.Path { case "/xrpc/network.bsky.jetstream.getZstdDictionary": f.mu.Lock() f.dicts++ dicts := f.dicts f.mu.Unlock() if f.mode == "refresh-failed" && dicts > 1 { http.Error(w, "dictionary unavailable", http.StatusBadRequest) return } d := append([]byte(nil), dictionary...) if (f.mode == "rotation" || f.mode == "rotation-stop") && dicts > 1 { binary.LittleEndian.PutUint32(d[4:8], binary.LittleEndian.Uint32(d[4:8])+1) } w.Write(d) case "/xrpc/network.bsky.jetstream.subscribeEvents": f.mu.Lock() f.connections++ connections := f.connections if strings.HasPrefix(f.mode, "terminal-") { f.cursors = append(f.cursors, r.URL.Query().Get("cursor")) } if f.mode == "many-dids" { f.didCount = len(r.URL.Query()["dids"]) } f.mu.Unlock() if f.mode == "many-dids" && !reflect.DeepEqual(r.URL.Query()["dids"], filterDIDs()) { http.Error(w, "DID filters changed in transit", http.StatusBadRequest) return } if strings.HasPrefix(f.mode, "refresh-") && connections > 1 && r.URL.Query().Get("zstdDictionary") != "" { http.Error(w, "expected uncompressed fallback", http.StatusBadRequest) return } if (f.mode == "reconnect" || strings.HasPrefix(f.mode, "terminal-")) && connections > 1 && r.URL.Query().Get("cursor") != "1" { http.Error(w, "expected resume cursor 1", http.StatusBadRequest) return } if (f.mode == "rotation" || f.mode == "rotation-stop" || f.mode == "refresh-failed" || f.mode == "refresh-same") && connections == 1 { w.Header().Set("Content-Type", "application/json") w.WriteHeader(400) fmt.Fprint(w, `{"error":"UnknownZstdDictionary"}`) return } c, e := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"xrpc.v1.json"}}) if e != nil { return } defer c.CloseNow() ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second) defer cancel() d := append([]byte(nil), dictionary...) if f.mode == "rotation" || f.mode == "rotation-stop" { binary.LittleEndian.PutUint32(d[4:8], binary.LittleEndian.Uint32(d[4:8])+1) } enc, e := zstd.NewWriter(nil, zstd.WithEncoderDict(d), zstd.WithEncoderConcurrency(1)) if e != nil { http.Error(w, e.Error(), http.StatusInternalServerError) return } defer enc.Close() compressed := r.URL.Query().Get("zstdDictionary") != "" if f.mode == "malformed" || f.mode == "malformed-stop" { if e = c.Write(ctx, websocket.MessageBinary, []byte("bad zstd")); e != nil { return } } seqs := []int{1, 2, 3} if f.mode == "archive-handoff" { if r.URL.Query().Get("cursor") != "2" { return } seqs = []int{2, 3} } if (f.mode == "reconnect" || strings.HasPrefix(f.mode, "terminal-")) && connections == 1 { seqs = []int{1} } if f.mode == "duplicate" { seqs = []int{1, 1, 2, 3} } for _, seq := range seqs { collection := "app.bsky.feed.post" if seq == 2 { collection = "app.bsky.feed.like" } did := "did:plc:fixture" if f.mode == "many-dids" { did = filterDIDs()[0] } b := []byte(fmt.Sprintf(`{"$type":"message","payload":{"$type":"network.bsky.jetstream.subscribeEvents#commit","collection":%q,"did":%q,"operation":"create","record":{"$type":"app.bsky.feed.post","text":"hi"},"rev":"rev1","rkey":"rkey1","seq":%d,"time":"2026-07-13T00:00:01.000000Z"}}`, collection, did, seq)) typ := websocket.MessageText if compressed { typ = websocket.MessageBinary b = enc.EncodeAll(b, nil) } if e = c.Write(ctx, typ, b); e != nil { return } } if strings.HasPrefix(f.mode, "terminal-") && connections == 1 { _ = c.Write(ctx, websocket.MessageText, []byte(`{"$type":"error","error":"FutureCursor","message":"fixture terminal error"}`)) return } if f.mode == "reconnect" && connections == 1 { return } // Wait for consumer stop rather than inducing reconnect/error differences. _, _, _ = c.Read(ctx) default: http.NotFound(w, r) }}
type result struct { Scenario string `json:"scenario"` Transcript []string `json:"transcript"` DictionaryRequests int `json:"dictionary_requests"` Connections int `json:"connections"` DIDCount int `json:"did_count,omitempty"` Cursors []string `json:"cursors,omitempty"` Details []string `json:"error_details,omitempty"` Failure string `json:"failure,omitempty"`}
func run(mode string, zig bool) (outcome result, runErr error) { outcome.Scenario = mode outcome.Transcript = []string{} f := &fixture{mode: mode} defer func() { if runErr != nil { outcome.Failure = runErr.Error() } }() if strings.HasPrefix(mode, "archive-") { frames, err := archiveFramesFor(mode) if err != nil { return outcome, err } f.frames = frames } server := httptest.NewUnstartedServer(f) server.Config.ReadHeaderTimeout = 2 * time.Second server.Config.ReadTimeout = 5 * time.Second server.Config.WriteTimeout = 5 * time.Second server.Start() defer server.Close() defer func() { f.mu.Lock() defer f.mu.Unlock() outcome.Scenario = mode outcome.DictionaryRequests, outcome.Connections = f.dicts, f.connections outcome.DIDCount, outcome.Cursors = f.didCount, f.cursors }() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() lines := []string{} defer func() { outcome.Transcript = lines }() target := 3 if mode == "exact" || mode == "resume" { target = 2 } if strings.HasPrefix(mode, "archive-kinds") { target = 2 } if zig { cmd := exec.CommandContext(ctx, "zig-out/bin/zig-client", server.URL, mode) out, e := cmd.CombinedOutput() for _, l := range strings.Split(string(out), "\n") { if strings.HasPrefix(l, "DETAIL ") { outcome.Details = append(outcome.Details, strings.TrimPrefix(l, "DETAIL ")) } if strings.HasPrefix(l, "EVENT\t") || l == "ERROR" { lines = append(lines, l) } } if e != nil { return outcome, fmt.Errorf("Zig %s: %w (deadline: %v): %s", mode, e, ctx.Err(), out) } } else { opts := []js.Option{js.WithBatchSize(1), js.WithZstdCompression(mode != "plain" && !isRegression(mode) && !strings.HasPrefix(mode, "archive-"))} if strings.HasPrefix(mode, "archive-") { opts = append(opts, js.WithAfterSeq(0)) if mode == "archive-pages" || strings.HasPrefix(mode, "archive-kinds") { opts = append(opts, js.WithSnapshotOnly()) } } if strings.HasPrefix(mode, "archive-kinds") { opts = append(opts, js.WithKinds([]js.Kind{js.KindCommit})) } if mode == "many-dids" { opts = append(opts, js.WithDIDs(filterDIDs())) } if mode == "exact" { opts = append(opts, js.WithCollection("app.bsky.feed.post")) } if mode == "wildcard" { opts = append(opts, js.WithCollection("app.bsky.feed.*")) } if mode == "resume" { opts = append(opts, js.WithLiveCursor(1)) } client, e := js.Subscribe(server.URL, opts...) if e != nil { return outcome, e } defer client.Close() count := 0 for batch, e := range client.Events(ctx) { if e != nil { if ctx.Err() != nil { return outcome, ctx.Err() } lines = append(lines, "ERROR") outcome.Details = append(outcome.Details, e.Error()) if strings.HasSuffix(mode, "-stop") { break } continue } for _, ev := range batch.Events() { c := ev.Commit if c == nil { return outcome, fmt.Errorf("unexpected non-commit") } lines = append(lines, fmt.Sprintf("EVENT\t%d\t%s\t%s\t%s\t%s", ev.Seq, ev.DID, c.Collection, c.Rkey, c.Operation)) count++ } if count >= target { break } } } if ctx.Err() != nil { return outcome, ctx.Err() } return outcome, nil}func expected(mode string) result { lines := []string{} if mode == "malformed" || mode == "rotation" || strings.HasSuffix(mode, "-stop") || strings.HasPrefix(mode, "refresh-") { lines = append(lines, "ERROR") } for _, seq := range []int{1, 2, 3} { if strings.HasSuffix(mode, "-stop") { break } if mode == "exact" && seq == 2 || mode == "resume" && seq == 1 { continue } if mode == "reconnect" && seq == 2 { lines = append(lines, "ERROR") } coll := "app.bsky.feed.post" if seq == 2 { coll = "app.bsky.feed.like" } lines = append(lines, fmt.Sprintf("EVENT\t%d\tdid:plc:fixture\t%s\trkey1\tcreate", seq, coll)) } dicts, conns := 1, 1 if strings.HasPrefix(mode, "archive-") { dicts = 0 if mode == "archive-pages" || strings.HasPrefix(mode, "archive-kinds") { conns = 0 } } if mode == "plain" { dicts = 0 } if mode == "rotation" || strings.HasPrefix(mode, "refresh-") { dicts = 2 conns = 2 } if mode == "reconnect" { conns = 2 } if mode == "rotation-stop" { dicts = 2 } return result{Scenario: mode, Transcript: lines, DictionaryRequests: dicts, Connections: conns}}func main() { if len(os.Args) > 2 && os.Args[1] == "regressions" { fmt.Fprintln(os.Stderr, "usage: conformance regressions") os.Exit(2) } if len(os.Args) > 1 && os.Args[1] != "regressions" { if err := loadMain(os.Args[1:]); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } return } type comparison struct { Scenario string `json:"scenario"` Match bool `json:"match"` Expected result `json:"expected"` Go result `json:"go"` Zig result `json:"zig"` } results := []comparison{} failed := false modes := []string{"plain", "compressed", "rotation", "exact", "wildcard", "duplicate", "resume", "malformed", "rotation-stop", "malformed-stop", "refresh-failed", "refresh-same", "reconnect", "archive-pages", "archive-handoff"} if len(os.Args) > 1 { modes = regressionModes } for _, mode := range modes { goResult, goErr := run(mode, false) zigResult, zigErr := run(mode, true) want := expected(mode) if isRegression(mode) { want = regressionExpected(mode) } match := goErr == nil && zigErr == nil && matches(goResult, want) && matches(zigResult, want) verdict := "PASS" if !match { verdict = "FAIL" failed = true } fmt.Fprintln(os.Stderr, verdict, mode) results = append(results, comparison{mode, match, want, goResult, zigResult}) } if e := json.NewEncoder(os.Stdout).Encode(results); e != nil { fmt.Fprintln(os.Stderr, e) os.Exit(1) } if failed { os.Exit(1) }}
var regressionModes = []string{"terminal-continue", "terminal-stop", "many-dids", "archive-kinds-excluded", "archive-kinds-excluded-stop", "archive-kinds-valid", "archive-kinds-selected", "archive-kinds-selected-stop"}
func isRegression(mode string) bool { for _, m := range regressionModes { if mode == m { return true } } return false}func filterDIDs() []string { const alphabet = "abcdefghijklmnopqrstuvwxyz234567" dids := make([]string, 120) for i := range dids { dids[i] = "did:plc:" + strings.Repeat("a", 22) + string(alphabet[i/32]) + string(alphabet[i%32]) } return dids}func matches(got, want result) bool { got.Details = nil; return reflect.DeepEqual(got, want) }func regressionExpected(mode string) result { r := result{Scenario: mode, Transcript: []string{}} did := "did:plc:fixture" if mode == "many-dids" { did = filterDIDs()[0] r.DIDCount = 120 } if !strings.HasPrefix(mode, "archive-") { r.Connections = 1 } if strings.HasPrefix(mode, "terminal-") { r.Cursors = []string{""} if mode == "terminal-continue" { r.Connections = 2 r.Cursors = append(r.Cursors, "1") } } for seq := 1; seq <= 3; seq++ { if seq == 2 && strings.HasPrefix(mode, "terminal-") { r.Transcript = append(r.Transcript, "ERROR") if mode == "terminal-stop" { break } } if seq == 2 && strings.HasPrefix(mode, "archive-kinds") { if strings.HasPrefix(mode, "archive-kinds-selected") { r.Transcript = append(r.Transcript, "ERROR") if strings.HasSuffix(mode, "-stop") { break } } continue } collection := "app.bsky.feed.post" if seq == 2 { collection = "app.bsky.feed.like" } r.Transcript = append(r.Transcript, fmt.Sprintf("EVENT\t%d\t%s\t%s\trkey1\tcreate", seq, did, collection)) } return r}