Embed zstd dictionary and update readme

Dockerfile

@@ -32,8 +32,5 @@ COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certifica
 # Copy the binary from the first stage.
 COPY --from=builder /app/jetstream .
 
-# Copy the zstd dictionary
-COPY zstd-dictionary /zstd-dictionary
-
 # Set the startup command to run the binary
 CMD ["./jetstream"]

README.md

@@ -40,6 +40,16 @@ The following Query Parameters are supported:
   - An absent cursor or a cursor from the future will result in live-tail operation
   - When reconnecting, use the `time_us` from your most recently processed event and maybe provide a negative buffer (i.e. subtract a few seconds) to ensure gapless playback
 
+### Compression
+
+Jetstream supports `zstd`-based compression of messages. Jetstream uses a custom dictionary for compression that can be found in `pkg/models/zstd_dictionary` and is required to decode compressed messages from the server.
+
+`zstd` compressed Jetstream messages are ~56% smaller on average than the raw JSON version of the Jetstream firehose.
+
+The provided client library uses compression by default, using an embedded copy of the Dictionary from the `models` package.
+
+### Examples
+
 A simple example that hits the public instance looks like:
 
 ```bash

cmd/client/main.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"log"
 	"log/slog"
 	"os"
@@ -28,19 +27,6 @@ func main() {
 	})))
 	logger := slog.Default()
 
-	// Open the zstd dictionary file if it is set
-	f, err := os.Open("zstd-dictionary")
-	if err != nil {
-		log.Fatalf("failed to open zstd dictionary file: %v", err)
-	}
-
-	dictBytes, err := io.ReadAll(f)
-	if err != nil {
-		f.Close()
-		log.Fatalf("failed to read zstd dictionary file: %v", err)
-	}
-	f.Close()
-
 	config := client.DefaultClientConfig()
 	config.WebsocketURL = serverAddr
 	config.Compress = true
@@ -51,7 +37,7 @@ func main() {
 
 	scheduler := sequential.NewScheduler("jetstream_localdev", logger, h.HandleEvent)
 
-	c, err := client.NewClient(config, logger, scheduler, dictBytes)
+	c, err := client.NewClient(config, logger, scheduler)
 	if err != nil {
 		log.Fatalf("failed to create client: %v", err)
 	}

cmd/jetstream/main.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"io"
 	"log"
 	"log/slog"
 	"net/http"
@@ -19,7 +18,6 @@ import (
 	"github.com/bluesky-social/indigo/events/schedulers/parallel"
 	"github.com/bluesky-social/jetstream/pkg/consumer"
 	"github.com/gorilla/websocket"
-	"github.com/klauspost/compress/zstd"
 	"github.com/labstack/echo/v4"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.opentelemetry.io/otel"
@@ -121,35 +119,6 @@ func Jetstream(cctx *cli.Context) error {
 		return fmt.Errorf("failed to parse ws-url: %w", err)
 	}
 
-	// Open the zstd dictionary file if it is set
-	var dictBytes []byte
-	if cctx.String("zstd-dictionary-path") != "" {
-		f, err := os.Open(cctx.String("zstd-dictionary-path"))
-		if err != nil {
-			return fmt.Errorf("failed to open zstd dictionary file: %w", err)
-		}
-
-		dictBytes, err = io.ReadAll(f)
-		if err != nil {
-			f.Close()
-			return fmt.Errorf("failed to read zstd dictionary file: %w", err)
-		}
-		f.Close()
-	}
-
-	var enc *zstd.Encoder
-	if len(dictBytes) > 0 {
-		enc, err = zstd.NewWriter(nil, zstd.WithEncoderDict(dictBytes))
-		if err != nil {
-			return fmt.Errorf("failed to create zstd encoder: %w", err)
-		}
-	} else {
-		enc, err = zstd.NewWriter(nil)
-		if err != nil {
-			return fmt.Errorf("failed to create zstd encoder: %w", err)
-		}
-	}
-
 	s, err := NewServer(cctx.Float64("max-sub-rate"))
 	if err != nil {
 		return fmt.Errorf("failed to create server: %w", err)
@@ -161,7 +130,6 @@ func Jetstream(cctx *cli.Context) error {
 		u.String(),
 		cctx.String("data-dir"),
 		cctx.Duration("event-ttl"),
-		enc,
 		s.Emit,
 	)
 	if err != nil {

docker-compose.yaml

@@ -11,5 +11,4 @@ services:
       - ./data:/data
     environment:
       - JETSTREAM_DATA_DIR=/data
-      - JETSTREAM_ZSTD_DICTIONARY_PATH=/zstd-dictionary
       - JETSTREAM_MAX_SUB_RATE=1_000_000

pkg/client/client.go

@@ -33,7 +33,6 @@ type Client struct {
 	config     *ClientConfig
 	logger     *slog.Logger
 	decoder    *zstd.Decoder
-	dictBytes  []byte
 	BytesRead  atomic.Int64
 	EventsRead atomic.Int64
 	shutdown   chan chan struct{}
@@ -51,7 +50,7 @@ func DefaultClientConfig() *ClientConfig {
 	}
 }
 
-func NewClient(config *ClientConfig, logger *slog.Logger, scheduler Scheduler, dictBytes []byte) (*Client, error) {
+func NewClient(config *ClientConfig, logger *slog.Logger, scheduler Scheduler) (*Client, error) {
 	if config == nil {
 		config = DefaultClientConfig()
 	}
@@ -66,8 +65,7 @@ func NewClient(config *ClientConfig, logger *slog.Logger, scheduler Scheduler, d
 
 	if config.Compress {
 		c.config.ExtraHeaders["Accept-Encoding"] = "zstd"
-		c.dictBytes = dictBytes
-		dec, err := zstd.NewReader(nil, zstd.WithDecoderDicts(dictBytes))
+		dec, err := zstd.NewReader(nil, zstd.WithDecoderDicts(models.ZSTDDictionary))
 		if err != nil {
 			return nil, fmt.Errorf("failed to create zstd decoder: %w", err)
 		}

pkg/consumer/consumer.go

@@ -51,7 +51,6 @@ func NewConsumer(
 	socketURL string,
 	dataDir string,
 	eventTTL time.Duration,
-	encoder *zstd.Encoder,
 	emit func(context.Context, *models.Event, []byte, []byte) error,
 ) (*Consumer, error) {
 	uDBPath := dataDir + "/jetstream.uncompressed.db"
@@ -73,6 +72,11 @@ func NewConsumer(
 		return nil, fmt.Errorf("failed to create clock: %w", err)
 	}
 
+	encoder, err := zstd.NewWriter(nil, zstd.WithEncoderDict(models.ZSTDDictionary))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create zstd encoder: %w", err)
+	}
+
 	c := Consumer{
 		SocketURL: socketURL,
 		Progress: &Progress{

pkg/models/models.go

@@ -3,9 +3,14 @@ package models
 import (
 	"github.com/goccy/go-json"
 
+	_ "embed"
+
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
 )
 
+//go:embed zstd_dictionary
+var ZSTDDictionary []byte
+
 type Event struct {
 	Did       string                                  `json:"did"`
 	TimeUS    int64                                   `json:"time_us"`