jetstream client
atproto jetstream client
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132package main
import ( "bytes" "encoding/json" "fmt" "net/http" "os" "path/filepath" "strconv" "strings"
"github.com/bluesky-social/jetstream/segment" "github.com/jcalabro/atmos/cbor")
// Generate real sealed JSS through upstream's writer, then serve its block// frames. This keeps the wire fixture independent of the Zig decoder.func archiveFrames() ([][]byte, error) { return archiveFramesFor("") }func archiveFramesFor(mode string) ([][]byte, error) { dir, err := os.MkdirTemp("", "jetstream-conformance-") if err != nil { return nil, err } defer os.RemoveAll(dir) path := filepath.Join(dir, "fixture.jss") writer, err := segment.New(segment.Config{Path: path, MaxEventsPerBlock: 1}) if err != nil { return nil, err } defer writer.Close() for seq := 1; seq <= 3; seq++ { collection := "app.bsky.feed.post" if seq == 2 { collection = "app.bsky.feed.like" } payload, err := cbor.Marshal(map[string]any{"$type": collection, "text": "hi"}) if err != nil { return nil, err } kind := segment.KindCreate if seq == 2 && strings.HasPrefix(mode, "archive-kinds") { if !strings.HasPrefix(mode, "archive-kinds-selected") { kind = segment.KindIdentity } if mode == "archive-kinds-valid" { payload, err = cbor.Marshal(map[string]any{"handle": "fixture.test"}) if err != nil { return nil, err } } else { payload = []byte{0xff} } } _, err = writer.Append(segment.Event{Seq: uint64(seq), WitnessedAt: 1783900801000000, Kind: kind, DID: "did:plc:fixture", Collection: collection, Rkey: "rkey1", Rev: "rev1", Payload: payload}) if err != nil { return nil, err } if seq < 3 { if err = writer.Flush(); err != nil { return nil, err } } } if _, err = writer.Seal(); err != nil { return nil, err } raw, err := os.ReadFile(path) if err != nil { return nil, err } header, err := segment.ReadSealedHeader(bytes.NewReader(raw)) if err != nil { return nil, err } if header.BlockCount != 3 { return nil, fmt.Errorf("expected three archive blocks, got %d", header.BlockCount) } var frames [][]byte for i := 0; i < 3; i++ { frame, err := segment.ReadBlockFrame(bytes.NewReader(raw), header, i) if err != nil { return nil, err } frames = append(frames, frame) } return frames, nil}func (f *fixture) serveArchive(w http.ResponseWriter, r *http.Request) bool { if f.mode != "archive-pages" && f.mode != "archive-handoff" && !strings.HasPrefix(f.mode, "archive-kinds") { return false } if r.URL.Path == "/xrpc/network.bsky.jetstream.subscribeEvents" { return false } tip := 3 if f.mode == "archive-handoff" { tip = 2 } switch r.URL.Path { case "/xrpc/network.bsky.jetstream.planSnapshot": var input struct { After int `json:"afterSeq"` Before *int `json:"beforeSeq"` } if err := json.NewDecoder(r.Body).Decode(&input); err != nil { http.Error(w, "invalid plan body", 400) return true } after := input.After if after < 0 || after >= tip { http.Error(w, "unexpected plan cursor", 400) return true } if after > 0 && (input.Before == nil || *input.Before != tip) { http.Error(w, "snapshot tip must remain pinned", 400) return true } _ = json.NewEncoder(w).Encode(map[string]any{"plannedThroughSeq": after + 1, "sealedTipSeq": tip, "segments": []any{map[string]any{"name": "fixture", "index": 0, "minSeq": after + 1, "maxSeq": after + 1, "mode": "blocks", "blocks": []any{map[string]any{"first": after, "last": after}}}}}) case "/xrpc/network.bsky.jetstream.getBlock": index, err := strconv.Atoi(r.URL.Query().Get("blockIndex")) if err != nil || index < 0 || index >= len(f.frames) { http.Error(w, "invalid block", 400) return true } _, _ = w.Write(f.frames[index]) default: http.Error(w, "unexpected archive request", 400) } return true}