### Update test server setup

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-26-local-simulator.md

Replaces the srvURL helper with a two-step setup to correctly initialize the simulator handler with the server URL.

```go
	srv := httptest.NewServer(nil)
	srv.Config.Handler = simhttp.NewHandler(w, srv.URL)
	defer srv.Close()
```

--------------------------------

### Apply Backfill Sequence Options

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/03-options.md

Examples of using sequence bounds to start from the beginning or resume from a specific cursor.

```go
// Backfill from the beginning
client, err := jetstream.Subscribe("jetstream.host", jetstream.WithAfterSeq(0))

// Resume from last-saved cursor
savedCursor := uint64(12345)
client, err := jetstream.Subscribe("jetstream.host", jetstream.WithAfterSeq(savedCursor))
```

--------------------------------

### Initialize and Run Engine

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-18-backfill-bootstrap-design.md

Configures the backfill engine with required dependencies and starts execution.

```go
eng := backfill.NewEngine(backfill.Options{
    SyncClient: sc,
    Store:      ourStore,
    Handler:    ourHandler,
    Directory:  gt.Some(dir),
    HTTPClient: gt.Some(httpClient),
    OnError:    gt.Some(func(did atmos.DID, err error) { ... }),
    OnProgress: gt.Some(func(s backfill.Stats) { ... }),
})
return eng.Run(ctx)
```

--------------------------------

### Setup integration test boilerplate

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-21-live-firehose-consumer.md

Initial imports and setup for the integration test file internal/livestream/consumer_test.go.

```go
package livestream

import (
	"context"
	"io"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/coder/websocket"
	"github.com/jcalabro/atmos/api/comatproto"
	"github.com/jcalabro/atmos/cbor"
	"github.com/jcalabro/gt"
	"github.com/stretchr/testify/require"
)
```

--------------------------------

### Start Simulator and Jetstream

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-26-local-simulator-design.md

Commands to initialize the simulator and point the Jetstream service to it.

```sh
# Terminal 1: start the simulator
just simulator
# (first run takes a few seconds bootstrapping 10k accounts; subsequent runs are instant)

# Terminal 2: jetstream points at the simulator by default
just run serve
```

--------------------------------

### Test steady-state startup

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-23-backfill-to-live-cutover.md

Replaces the existing steady-state refusal test with a verification that the application starts cleanly in the steady-state phase.

```go
// TestServe_StartsInSteadyStatePhase pins the steady-state startup
// path: a data dir already at PhaseSteadyState skips bootstrap and
// merge, runs the steady-state consumer, and shuts down cleanly on
// ctx cancel.
func TestServe_StartsInSteadyStatePhase(t *testing.T) {
	t.Parallel()

	dataDir := t.TempDir()

	// Pre-populate phase=steady_state.
	{
		s, err := store.Open(dataDir)
		require.NoError(t, err)
		require.NoError(t, lifecycle.WritePhase(s, lifecycle.PhaseSteadyState))
		require.NoError(t, s.Close())
	}

	// subscribeReposHit is closed the first time the steady-state
	// live consumer dials the relay. That's our deterministic
	// "serve made it to phase=steady_state work" signal.
	subscribeReposHit := make(chan struct{})
	var hitOnce sync.Once
	relay := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if strings.HasSuffix(r.URL.Path, "/com.atproto.sync.subscribeRepos") {
			hitOnce.Do(func() { close(subscribeReposHit) })
			conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{InsecureSkipVerify: true})
			if err != nil {
				return
			}
			defer func() { _ = conn.CloseNow() }()
			<-r.Context().Done()
			return
		}
	}))
	t.Cleanup(relay.Close)

	ctx, cancel := context.WithCancel(t.Context())
	t.Cleanup(cancel)

	done := make(chan error, 1)
	go func() {
		done <- newApp().Run(ctx, []string{
			"jetstream",
			"--log-format=text",
			"--log-level=warn",
			"serve",
			"--addr=127.0.0.1:0",
			"--debug-addr=127.0.0.1:0",
			"--shutdown-timeout=5s",
			"--relay-url=" + relay.URL,
			"--data-dir=" + dataDir,
		})
	}()

	select {
	case <-subscribeReposHit:
	case err := <-done:
		t.Fatalf("serve exited before reaching the relay: %v", err)
	case <-time.After(5 * time.Second):
		t.Fatal("steady-state consumer never reached the relay")
	}

	cancel()
	select {
	case err := <-done:
		if err != nil && !errors.Is(err, context.Canceled) {
			t.Fatalf("serve exited with unexpected error: %v", err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("serve did not shut down")
	}

	// Phase should still be steady_state.
	s, err := store.Open(dataDir)
	require.NoError(t, err)
	defer func() { _ = s.Close() }()
	p, err := lifecycle.ReadPhase(s)
	require.NoError(t, err)
	require.Equal(t, lifecycle.PhaseSteadyState, p)
}
```

--------------------------------

### Build and run Jetstream with observability enabled

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-23-observability-sweep.md

Starts the Jetstream binary with environment variables configured for logging and OTLP trace export.

```bash
just build
JETSTREAM_LOG_FORMAT=text JETSTREAM_LOG_LEVEL=info \
    OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 \
    ./bin/jetstream serve --data-dir /tmp/jss-smoke
```

--------------------------------

### Initialize Inspection Package

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-19-inspect-segment.md

Initial setup for the segment inspection package, including necessary imports for file handling and binary parsing.

```go
// Package segment — Inspection surface used by the inspect-segment
// CLI. Active-file support lives in this same file; both paths
// produce the same Inspection value so the renderer is one code
// path.
package segment

import (
	"encoding/binary"
	"errors"
	"fmt"
	"os"
)
```

--------------------------------

### Subscribe to Jetstream Events

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/01-client-api.md

Examples demonstrating live tailing from the current tip and backfilling events from a specific sequence with collection filtering.

```go
// Live tail from current tip
client, err := jetstream.Subscribe("jetstream.us-west.bsky.network")
if err != nil {
    log.Fatal(err)
}
defer client.Close()

for batch, err := range client.Events(context.Background()) {
    if err != nil {
        log.Printf("stream error: %v", err)
        continue
    }
    for _, ev := range batch.Events() {
        // Process event
    }
}

// Backfill from specific cursor
client, err := jetstream.Subscribe("jetstream.us-west.bsky.network",
    jetstream.WithAfterSeq(12345),
    jetstream.WithCollections([]string{"app.bsky.feed.post"}),
)
if err != nil {
    log.Fatal(err)
}
defer client.Close()

for batch, err := range client.Events(context.Background()) {
    // Backfill and then live tail
}
```

--------------------------------

### Start Jetstream server

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-25-public-status-page.md

Starts the Jetstream service with specified debug and server addresses.

```bash
just clean
just run serve --addr :8080 --debug-addr :6060
```

--------------------------------

### Initialize Identity Directory

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-18-backfill-bootstrap-design.md

Sets up the identity directory with a resolver and LRU cache.

```go
dir := &identity.Directory{
    Resolver: &identity.DefaultResolver{
        HTTPClient: gt.Some(httpClient),
    },
    Cache: identity.NewLRUCache(100_000, 24 * time.Hour),
}
```

--------------------------------

### Initialize simulator main package

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-26-local-simulator.md

The main entry point for the simulator, including necessary imports for the local development network.

```go
// Command simulator is a development-only fake atproto network: PLC,
// a single PDS, and a relay (firehose) under one HTTP listener. It
// exists so jetstream can iterate locally without depending on
// bsky.network or plc.directory. Not shipped to users; not in the
// Dockerfile.
package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"math/rand/v2"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/bluesky-social/jetstream/internal/obs"
	"github.com/bluesky-social/jetstream/internal/simulator/fanout"
	simhttp "github.com/bluesky-social/jetstream/internal/simulator/http"
	"github.com/bluesky-social/jetstream/internal/simulator/world"
	"github.com/urfave/cli/v3"
	"golang.org/x/sync/errgroup"
)
```

--------------------------------

### Client.Events(ctx)

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/client.md

Starts the event stream iteration for the client.

```APIDOC
## Client.Events(ctx)

### Description
Returns an `iter.Seq2[*Batch, error]` range-over-func iterator. Recoverable errors are yielded while iteration continues; terminal failures satisfy `errors.Is(err, ErrFatal)` and terminate the stream.
```

--------------------------------

### Enter Development Shell

Source: https://github.com/bluesky-social/jetstream/blob/main/README.md

Commands to initialize the development environment using Nix.

```sh
./dev.sh
# or
just dev
```

--------------------------------

### GET /xrpc/network.bsky.jetstream.getSegment

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/08-backfill-protocol.md

Retrieves a raw segment file by its path.

```APIDOC
## GET /xrpc/network.bsky.jetstream.getSegment

### Description
Downloads the raw bytes of a specific segment file.

### Method
GET

### Endpoint
/xrpc/network.bsky.jetstream.getSegment

### Parameters
#### Query Parameters
- **path** (string) - Required - The path of the segment file to retrieve

### Response
#### Success Response (200)
- **body** (binary) - Raw segment file bytes
```

--------------------------------

### Wrap backfill.Run in Observe

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-23-observability-sweep.md

Instruments the backfill execution entry point.

```go
func Run(ctx context.Context, cfg Config) (retErr error) {
	if err := cfg.validate(); err != nil {
		return err
	}
	ctx, _, done := obs.Observe(ctx)
	defer func() { done(retErr) }()

	// ... rest of body unchanged ...
}
```

--------------------------------

### client.Events

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/README.md

Starts a stream of events from the client, returning a channel of event batches.

```APIDOC
## client.Events(ctx context.Context) <-chan Batch

### Description
Consumes events from the established Jetstream connection. Returns batches of events that can be processed sequentially.
```

--------------------------------

### Build Project

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-25-public-status-page.md

Command to build the project and verify wiring.

```bash
just build
```

--------------------------------

### Get segment header

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/05-segment-api.md

Method to retrieve the parsed segment header metadata.

```go
func (r *Reader) Header() Header
```

--------------------------------

### Initialize and Run Backfill Engine

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-18-backfill-bootstrap.md

The Run function initializes the engine with a directory resolver, while runWithDirectory handles the core execution logic, allowing for dependency injection during testing.

```go
// Run drives the atmos backfill engine to completion. It blocks until
// the engine drains or ctx is cancelled. Safe to call multiple times
// across process restarts: each call constructs a fresh Engine
// (atmos engines are single-shot) and resumes by skipping rows
// already at StatusComplete via Store.Lookup.
func Run(ctx context.Context, cfg Config) error {
	if err := cfg.validate(); err != nil {
		return err
	}

	httpClient := cfg.HTTPClient
	if httpClient == nil {
		httpClient = &http.Client{Timeout: 30 * time.Second}
	}

	dir := &identity.Directory{
		Resolver: &identity.DefaultResolver{
			HTTPClient: gt.Some(httpClient),
		},
		Cache: identity.NewLRUCache(directoryCacheCapacity, directoryCacheTTL),
	}

	return runWithDirectory(ctx, cfg, httpClient, dir)
}

// runWithDirectory is the production entry point's internal worker.
// Tests inject a stub resolver via the Directory parameter so we can
// avoid spinning up a real PLC.
func runWithDirectory(ctx context.Context, cfg Config, httpClient *http.Client, dir *identity.Directory) error {
	xc := &xrpc.Client{
		Host:       cfg.RelayURL,
		HTTPClient: gt.Some(httpClient),
		Retry:      gt.Some(xrpc.RetryPolicy{MaxAttempts: gt.Some(1)}),
	}
	sc := atmossync.NewClient(atmossync.Options{
		Client:    xc,
		Directory: gt.Some(dir),
	})

	st := NewStore(cfg.Store, cfg.Metrics)
	handler := NewLogHandler(cfg.Logger)
	logger := cfg.Logger

	engine := atmosbackfill.NewEngine(atmosbackfill.Options{
		SyncClient: sc,
		Store:      st,
		Handler:    handler,
		Directory:  gt.Some(dir),
		HTTPClient: gt.Some(httpClient),
		OnError: gt.Some(func(did atmos.DID, err error) {
			logger.Warn("backfill: repo failed", "did", string(did), "err", err)
		}),
		OnProgress: gt.Some(func(stats atmosbackfill.Stats) {
			if stats.Completed%progressLogInterval == 0 {
				logger.Info("backfill: progress", "completed", stats.Completed)
			}
		}),
	})

	logger.Info("backfill: starting", "relay", cfg.RelayURL)
	err := engine.Run(ctx)
	if err != nil {
		logger.Error("backfill: engine returned error", "err", err)
		return fmt.Errorf("backfill: %w", err)
	}
	logger.Info("backfill: engine drained")
	return nil
}
```

--------------------------------

### Get block count

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/05-segment-api.md

Method to retrieve the total number of blocks in the segment.

```go
func (r *Reader) BlockCount() int
```

--------------------------------

### Implement Phase-Aware Startup Logic

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-21-live-firehose-consumer-design.md

Consults the lifecycle phase during service startup to determine whether to initiate backfill or steady-state processing.

```go
phase, err := lifecycle.ReadPhase(metaStore)
if err != nil { return err }

if phase == "" {
    // Fresh data dir or upgrade from a pre-phase build. Both are
    // bootstrap.
    phase = lifecycle.PhaseBootstrap
    if err := lifecycle.WritePhase(metaStore, phase); err != nil {
        return err
    }
}

switch phase {
case lifecycle.PhaseBootstrap:
    g.Go(...backfill...)
    g.Go(...liveConsumer...)
case lifecycle.PhaseSteadyState:
    return errors.New("serve: steady-state phase not yet supported; the merge step has not been implemented")
default:
    // Unreachable: ReadPhase rejected unknown values.
    return fmt.Errorf("serve: unhandled phase %q", phase)
}
```

--------------------------------

### Inspect-all output format

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-28-inspect-all-design.md

Example of the text-based output generated by the inspect-all command.

```text
inspect-all
data-dir: ./data
generated: 2026-05-28T17:42:31.000000Z

network totals:
  segments:               1234 (1233 sealed, 1 active)
  blocks:                 58,231
  events:                 412,834,991
  collections:            42 distinct NSIDs
  seq range:              [1, 412834991]
  indexed_at range:       2025-01-04T00:00:00.000000Z → 2026-05-28T17:41:58.123456Z
  payload (uncompressed): 87.4 GiB
  payload (compressed):   31.2 GiB
  disk usage:             32.1 GiB
  compression ratio:      2.80x

trees:
  [0] segments/
        files:        1233 sealed + 0 active
        events:       411,200,000
        blocks:       58,112
        seq range:    [1, 411200000]
        indexed_at:   2025-01-04T00:00:00.000000Z → 2026-05-27T22:14:11.000000Z
        oldest mtime: 2025-01-04T00:00:01Z
        newest mtime: 2026-05-27T22:14:18Z
        compressed:   31.0 GiB
        uncompressed: 87.0 GiB
        disk:         31.9 GiB
        latest:       idx=1232 sealed events=320,000 blocks=12 size=26.1 MiB
  [1] backfill/live_segments/
        (empty)

collections (42 distinct NSIDs):
  [  0] app.bsky.feed.post              events: 251,034,118  segments: 1234  blocks: 47,210
  [  1] app.bsky.feed.like              events:  93,442,001  segments: 1180  blocks:  8,012
  ...

warnings (1):
  /home/jcalabro/data/segments/seg_0001033.jss: corrupt segment: bad magic "..."
```

--------------------------------

### Integrate Live Consumer in Main Application

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-21-live-firehose-consumer-design.md

Pseudocode demonstrating the initialization and execution of the live consumer within the main application run loop.

```go
func runServe(ctx context.Context, cmd *cli.Command) error {
    // ... existing logger / tracing / metrics setup ...
    // ... existing metaStore Open ...

    phase, err := lifecycle.ReadPhase(metaStore)
    if err != nil { return fmt.Errorf("serve: read phase: %w", err) }
    if phase == "" {
        phase = lifecycle.PhaseBootstrap
        if err := lifecycle.WritePhase(metaStore, phase); err != nil {
            return fmt.Errorf("serve: write phase: %w", err)
        }
    }

    if phase == lifecycle.PhaseSteadyState {
        return errors.New("serve: steady-state phase not yet supported")
    }

    // ... existing ingest.Open for the backfill writer ...

    liveConsumer, err := livestream.Open(livestream.Config{
        SegmentsDir:       filepath.Join(dataDir, "backfill", "live_segments"),
        Store:             metaStore,
        SeqKey:            "live_segments/seq/next",
        CursorKey:         "relay/cursor",
        RelayURL:          cmd.String("relay-url"),
        Logger:            logger,
        Metrics:           livestream.NewMetrics(metrics.Registry),
    })
    if err != nil { return fmt.Errorf("livestream open: %w", err) }
    defer func() {
        if cerr := liveConsumer.Close(); cerr != nil {
            logger.Error("close live consumer", "err", cerr)
        }
    }()

    // ... existing srv := server.New ...
    // ... existing signal.NotifyContext ...

    g, gctx := errgroup.WithContext(runCtx)
    g.Go(func() error { return srv.Run(gctx) })
    g.Go(func() error { return backfill.Run(gctx, /* ... */) })
    g.Go(func() error { return liveConsumer.Run(gctx) })

    // ... existing graceful shutdown ...
}
```

--------------------------------

### getBlock Request Format

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-06-16-getblock-endpoint-design.md

The endpoint uses a GET request with segment and blockIndex parameters.

```http
GET /xrpc/network.bsky.jetstream.getBlock?segment=seg_000000002a.jss&blockIndex=7
```

--------------------------------

### Implement runBootstrap in bootstrap.go

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-23-backfill-to-live-cutover.md

This method initializes the backfill writer and bootstrap live consumer, then runs them within an errgroup to manage the cutover lifecycle.

```go
package orchestrator

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"path/filepath"
	"time"

	"github.com/bluesky-social/jetstream/internal/ingest"
	"github.com/bluesky-social/jetstream/internal/ingest/backfill"
	"github.com/bluesky-social/jetstream/internal/ingest/live"
	"golang.org/x/sync/errgroup"
)

// runBootstrap is the orchestrator's State 0. It builds:
//
//   - a shared ingest.Writer pointed at <DataDir>/segments (used by
//     the backfill engine; closed in State 4 of the cutover),
//   - the backfill engine itself,
//   - a live.Consumer pointed at <DataDir>/backfill/live_segments
//     with the throwaway "live_segments/seq/next" seq counter and
//     the shared "relay/cursor" upstream cursor.
//
// It runs the backfill engine and the live consumer as siblings
// under an internal errgroup, with the live consumer attached to a
// derived context the orchestrator can cancel independently. When
// backfill drains (returns nil), the orchestrator:
//
//   1. Writes phase=merging (commit point #1).
//   2. Cancels the live consumer's derived context.
//   3. Awaits the live consumer's Run return, then SealAndClose its
//      writer (state 3).
//   4. Closes the backfill writer (state 4).
//
// On success, runBootstrap returns nil and the caller falls through
// to the merge case. On any subsystem error before backfill drains,
// the errgroup cancels both and the error is returned without
// touching the phase.
func (o *Orchestrator) runBootstrap(ctx context.Context) error {
	segmentsDir := filepath.Join(o.cfg.DataDir, "segments")
	liveSegmentsDir := filepath.Join(o.cfg.DataDir, "backfill", "live_segments")

	// Backfill writer (shared with the backfill engine).
	bw, err := ingest.Open(ingest.Config{
		SegmentsDir: segmentsDir,
		Store:       o.cfg.Store,
		Logger:      o.cfg.Logger.With(slog.String("component", "orchestrator/backfill-ingest")),
		Metrics:     o.cfg.IngestMetrics,
	})
	if err != nil {
		return fmt.Errorf("orchestrator: open backfill ingest writer: %w", err)
	}

	// Bootstrap-time live consumer.
	bootstrapLive, err := live.Open(live.Config{
		SegmentsDir: liveSegmentsDir,
		Store:       o.cfg.Store,
		SeqKey:      "live_segments/seq/next",
		CursorKey:   "relay/cursor",
		RelayURL:    o.cfg.RelayURL,
		Logger:      o.cfg.Logger.With(slog.String("component", "orchestrator/bootstrap-live")),
		Metrics:     o.cfg.LiveMetrics,
		Verifier:    o.cfg.Verifier,
	})
	if err != nil {
		_ = bw.Close()
		return fmt.Errorf("orchestrator: open bootstrap-live consumer: %w", err)
	}

	g, gctx := errgroup.WithContext(ctx)

	// Derived context the orchestrator cancels at cutover-time.
	// Wrapping gctx means errgroup-driven cancellation (e.g. from a
	// backfill error) also propagates to the live consumer, while
	// still letting us call cancelLive() to stop ONLY the live
	// consumer when backfill drains successfully — gctx remains
	// uncancelled so the backfill goroutine can return nil normally.
	liveCtx, cancelLive := context.WithCancel(gctx)
	defer cancelLive()
```

--------------------------------

### Commit changes with git

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-21-live-firehose-consumer.md

Example command to commit the OnAfterFlush feature with a descriptive message.

```bash
git add internal/ingest/
git commit -m "$(cat <<'EOF'
feat(ingest): add OnAfterFlush callback for downstream durability

After each block flush, the writer now invokes Config.OnAfterFlush
with the caller's context, after segment.Flush has fsynced and
SeqKey has been pebble.Sync'd. The live_segments consumer will use
this to advance relay/cursor with the same per-block cadence the
DESIGN.md §3.1.1 invariant requires.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

--------------------------------

### Allocate helloCh and sync.Once

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-27-subscribe-require-hello.md

Initialize the signaling channel and sync.Once primitive before the reader goroutine starts.

```go
	// helloCh is closed by the reader goroutine on the first valid
	// options_update IFF requireHello is set. The signal is idempotent
	// via sync.Once so a chatty client sending multiple updates doesn't
	// panic on a closed channel. The pre-Subscribe wait below selects
	// on this channel and ctx.Done().
	helloCh := make(chan struct{})
	var helloOnce sync.Once
	signalHello := func() {
		helloOnce.Do(func() { close(helloCh) })
	}
```

--------------------------------

### Run local simulator and server

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-06-30-tombstone-pagination-review-guide.md

Commands to launch the local simulator, server, and bundled client for manual testing.

```sh
just simulator       # the fake network, in one terminal
just run             # the server, in another
just run-client ...  # the bundled client
```

--------------------------------

### Run Jetstream Against Local Simulator

Source: https://github.com/bluesky-social/jetstream/blob/main/README.md

Commands to start the local simulator and connect the Jetstream service to it.

```sh
# terminal 1: starts the simulator on :7777 with 10,000 mock accounts
# (this takes a minute to start up)
just simulator serve

# terminal 2: jetstream points at the simulator
just run serve
```

--------------------------------

### Run Full Test Suite

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-25-public-status-page.md

Execute linting and tests for the entire project.

```bash
just lint test
```

--------------------------------

### Visibility Assertion Failure Log

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/oracle/2026-07-06-rebackfill-erased-by-stale-account-tombstone.md

Example of the non-deterministic failure log observed during oracle-sweep testing.

```text
segmentfault-write-shortwrite-first-flush: recreated record
  app.bsky.feed.post/… must be visible (no permanent tombstone)
```

--------------------------------

### Consuming Typed Events

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/04-typed-events.md

Example usage of TypedEvents to subscribe to a collection and process decoded records.

```go
// Assuming bsky is a lexicon package that defines FeedPost
import "github.com/jcalabro/atmos/bsky"

client, err := jetstream.Subscribe("jetstream.us-west.bsky.network",
    jetstream.WithCollections([]string{"app.bsky.feed.post"}),
    jetstream.WithRawRecords(), // Zero-copy decode on workers
)
if err != nil {
    log.Fatal(err)
}
defer client.Close()

for tb, err := range jetstream.TypedEvents[bsky.FeedPost](ctx, client, "app.bsky.feed.post") {
    if err != nil {
        if errors.Is(err, jetstream.ErrFatal) {
            log.Fatal(err)
        }
        log.Printf("recoverable error: %v", err)
        continue
    }
    
    for _, te := range tb.Events() {
        if te.Record != nil {
            // te.Record is *bsky.FeedPost
            log.Printf("post by %s: %s", te.Event.DID, te.Record.Text)
        } else if te.DecodeErr != nil {
            log.Printf("decode error on %s: %v", te.Event.DID, te.DecodeErr)
        } else {
            // Delete, non-commit, or different collection
            log.Printf("non-decodable event: kind=%s", te.Event.Kind)
        }
    }
    
    // Cursor for resumption
    cursor := tb.LastCursor()
    _ = db.SaveCursor(cursor)
}
```

--------------------------------

### Create .env file

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-26-local-simulator.md

Define default environment variables for local simulator connectivity.

```text
# Dev-only defaults that point jetstream at the local simulator.
# `just run` and any other recipe with `set dotenv-load` picks these
# up automatically. `just run-prod` overrides these inline.
JETSTREAM_RELAY_URL=http://localhost:7777
JETSTREAM_PLC_URL=http://localhost:7777
```

--------------------------------

### Define package documentation

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-14-segment-file-format.md

Initializes the segment package with documentation describing the file format and concurrency constraints.

```go
// Package segment implements the jetstream segment file format: a
// columnar, zstd-compressed, length-prefixed binary log of atproto
// firehose events.
//
// This slice covers writing only. A future slice will add a public
// Reader, segment sealing (the 256-byte fixed header and footer
// described in DESIGN.md §3.1.2), and crash recovery.
//
// Concurrency: Writer is not safe for concurrent use. Callers
// serialize access. The package contains no goroutines, timers,
// or context plumbing; lifecycle (time-based flushes, graceful
// shutdown, pebble metadata coupling) is the responsibility of the
// ingestion orchestrator that composes Writer with the rest of the
// system.
package segment
```

--------------------------------

### Execute Backfill-Only Stream

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/03-options.md

Example of performing a bounded archive dump and processing the resulting events.

```go
// Export app.bsky.feed.like records from seq 0 to 10000
client, err := jetstream.Subscribe("jetstream.host",
    jetstream.WithAfterSeq(0),
    jetstream.WithBeforeSeq(10000),
    jetstream.WithCollections([]string{"app.bsky.feed.like"}),
    jetstream.WithBackfillOnly(),
)
if err != nil {
    log.Fatal(err)
}

for batch, err := range client.Events(ctx) {
    if err != nil {
        continue
    }
    // Process and stream exits when beforeSeq is reached
}
```

--------------------------------

### World Package Documentation

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-26-local-simulator.md

Package-level documentation for the simulator's on-disk state management.

```go
// Package world owns the simulator's on-disk state: the pebble db, the
// global RNG, the deterministic account roster, and the live commit
// generator. The HTTP layer reads through *World; only the traffic
// goroutine writes to pebble after bootstrap.
package world
```

--------------------------------

### Consuming Events with Range-over-func

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/01-client-api.md

Example implementation for processing event batches and handling errors during iteration.

```go
for batch, err := range client.Events(ctx) {
    if err != nil {
        if errors.Is(err, jetstream.ErrFatal) {
            // Terminal error: stop and surface to caller
            return err
        }
        // Recoverable error: log and continue
        log.Printf("recoverable error: %v", err)
        continue
    }
    // Process this batch's events
    events := batch.Events()
    cursor := batch.LastCursor()
}
```

--------------------------------

### Initialize Sync Verifier

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-22-sync-1-1-upgrade-design.md

Configures the sync verifier with the directory, state store, and sync client.

```go
verifier, err := sync.NewVerifier(sync.VerifierOptions{
    Directory:  directory,
    StateStore: stateStore,
    SyncClient: gt.Some(syncClient),
})
```

--------------------------------

### Git commit command

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-21-live-firehose-consumer.md

Example command to commit the new URL helper files with a descriptive message.

```bash
git add internal/livestream/url.go internal/livestream/url_test.go
git commit -m "$(cat <<'EOF'
feat(livestream): URL helper that converts --relay-url to wss firehose

deriveSubscribeReposURL maps https→wss, http→ws, and rejects empty,
missing-host, or unsupported-scheme inputs. The Consumer reuses
this so cmd/jetstream does not need a new flag for the WebSocket
URL.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

--------------------------------

### Initialize New Writer

Source: https://github.com/bluesky-social/jetstream/blob/main/_autodocs/05-segment-api.md

Constructor for creating a new Writer instance with the specified configuration.

```go
func New(cfg Config) (*Writer, error)
```

```go
w, err := segment.New(segment.Config{
    Path:               "seg_0000000001.jss",
    MaxEventsPerBlock:  4096,
})
if err != nil {
    log.Fatal(err)
}
defer w.Close()
```

--------------------------------

### Run Jetstream Service

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-26-local-simulator.md

Starts the Jetstream service configured to point at the local simulator via environment variables.

```sh
just run serve
```

--------------------------------

### Run Local Simulator

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-26-local-simulator.md

Starts the local simulator on port 7777 to emulate PLC, PDS, and relay services.

```sh
just simulator
```

--------------------------------

### Initialize Identity Directory

Source: https://github.com/bluesky-social/jetstream/blob/main/specs/notes/2026-05-22-sync-1-1-upgrade-design.md

Sets up the identity directory using a persistent cache instead of an in-memory one.

```go
directory := &identity.Directory{
    Resolver:               &identity.DefaultResolver{},
    Cache:                  identitycache.New(metaStore, identitycache.DefaultTTL),
    SkipHandleVerification: true,  // signing-key-only on the firehose hot path
}
```