rainbow: small refactors, and improve request forwarding (#1023)

A bit of this is moving code around so I could understand which pieces
were doing what. Then improved how the request proxying and requestCrawl
forwarding stuff works, as part of new relay ops.

None of the core configuration or behaviors should change: I tweaked
some argument names, but not the env var names, so nothing about how we
deploy/operate rainbow should change.

No behavior of the core code (ringbuf, upstream websocket subscription,
or serving to clients) has changed.

cmd/rainbow/README.md

···
 - retains upstream firehose "sequence numbers"
 - does not validate events (signatures, repo tree, hashes, etc), just passes through
 - does not archive or mirror individual records or entire repositories (or implement related API endpoints)
-- disk I/O intensive: fast NVMe disks are recommended, and RAM is helpful for caching
+- somewhat disk I/O intensive: fast NVMe disks are recommended, and RAM is helpful for caching
 - single golang binary for easy deployment
-- observability: logging, prometheus metrics, OTEL traces
+- observability: logging, prometheus metrics, and OTEL traces
 
 ## Running 
 
···
 From the top level of this repo, you can build:
 
 ```shell
-go build ./cmd/rainbow -o rainbow-bin
+go build ./cmd/rainbow -o rainbow
 ```
 
 or just run it, and see configuration options:

cmd/rainbow/main.go

···
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
-	_ "net/http/pprof"
 	"os"
 	"os/signal"
 	"syscall"
 	"time"
 
+	_ "github.com/joho/godotenv/autoload"
+	_ "go.uber.org/automaxprocs"
+	_ "net/http/pprof"
+
 	"github.com/bluesky-social/indigo/events/pebblepersist"
 	"github.com/bluesky-social/indigo/splitter"
+	"github.com/bluesky-social/indigo/util/svcutil"
 
 	"github.com/carlmjohnson/versioninfo"
-	_ "github.com/joho/godotenv/autoload"
 	"github.com/urfave/cli/v2"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
···
 	"go.opentelemetry.io/otel/sdk/resource"
 	tracesdk "go.opentelemetry.io/otel/sdk/trace"
 	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
-	_ "go.uber.org/automaxprocs"
 )
 
-var log = slog.Default().With("system", "rainbow")
-
-func init() {
-	// control log level using, eg, GOLOG_LOG_LEVEL=debug
-	//logging.SetAllLoggers(logging.LevelDebug)
-}
-
 func main() {
-	run(os.Args)
+	if err := run(os.Args); err != nil {
+		slog.Error("exiting", "err", err)
+		os.Exit(1)
+	}
 }
 
-func run(args []string) {
+func run(args []string) error {
+
 	app := cli.App{
 		Name:    "rainbow",
 		Usage:   "atproto firehose fan-out daemon",
 		Version: versioninfo.Short(),
+		Action:  runSplitter,
 	}
 
 	app.Flags = []cli.Flag{
-		// TODO: unimplemented, always assumes https:// and wss://
-		//&cli.BoolFlag{
-		//	Name:    "crawl-insecure-ws",
-		//	Usage:   "when connecting to PDS instances, use ws:// instead of wss://",
-		//	EnvVars: []string{"RAINBOW_INSECURE_CRAWL"},
-		//},
+		&cli.StringFlag{
+			Name:    "log-level",
+			Usage:   "log verbosity level (eg: warn, info, debug)",
+			EnvVars: []string{"RAINBOW_LOG_LEVEL", "GO_LOG_LEVEL", "LOG_LEVEL"},
+		},
 		&cli.StringFlag{
-			Name:    "splitter-host",
-			Value:   "bsky.network",
+			Name:    "upstream-host",
+			Value:   "http://localhost:2470",
+			Usage:   "URL (schema and hostname, no path) of the upstream host (eg, relay)",
 			EnvVars: []string{"ATP_RELAY_HOST", "RAINBOW_RELAY_HOST"},
 		},
 		&cli.StringFlag{
···
 			EnvVars: []string{"RAINBOW_PERSIST_BYTES", "SPLITTER_PERSIST_BYTES"},
 		},
 		&cli.StringSliceFlag{
+			// TODO: better name for this argument
 			Name:    "next-crawler",
-			Usage:   "forward POST requestCrawl to this url, should be machine root url and not xrpc/requestCrawl, comma separated list",
-			EnvVars: []string{"RELAY_NEXT_CRAWLER"},
+			Usage:   "forward POST requestCrawl to these hosts (schema and host, no path) in addition to upstream-host. Comma-separated or multiple flags",
+			EnvVars: []string{"RAINBOW_NEXT_CRAWLER", "RELAY_NEXT_CRAWLER"},
+		},
+		&cli.StringFlag{
+			Name:    "collectiondir-host",
+			Value:   "http://localhost:2510",
+			Usage:   "host (schema and hostname, no path) of upstream collectiondir instance, for com.atproto.sync.listReposByCollection",
+			EnvVars: []string{"RAINBOW_COLLECTIONDIR_HOST"},
+		},
+		&cli.StringFlag{
+			Name:    "env",
+			Usage:   "operating environment (eg, 'prod', 'test')",
+			Value:   "dev",
+			EnvVars: []string{"ENVIRONMENT"},
+		},
+		&cli.BoolFlag{
+			Name:    "enable-otel-otlp",
+			Usage:   "enables OTEL OTLP exporter endpoint",
+			EnvVars: []string{"RAINBOW_ENABLE_OTEL_OTLP", "ENABLE_OTEL_OTLP"},
+		},
+		&cli.StringFlag{
+			Name:    "otel-otlp-endpoint",
+			Usage:   "OTEL traces export endpoint",
+			Value:   "http://localhost:4318",
+			EnvVars: []string{"OTEL_EXPORTER_OTLP_ENDPOINT"},
 		},
 	}
 
-	// TODO: slog.SetDefault and set module `var log *slog.Logger` based on flags and env
-
-	app.Action = Splitter
-	err := app.Run(os.Args)
-	if err != nil {
-		log.Error(err.Error())
-		os.Exit(1)
-	}
+	return app.Run(args)
 }
 
-func Splitter(cctx *cli.Context) error {
+func runSplitter(cctx *cli.Context) error {
 	// Trap SIGINT to trigger a shutdown.
 	signals := make(chan os.Signal, 1)
 	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
 
+	logger := svcutil.ConfigLogger(cctx, os.Stdout).With("system", "rainbow")
+
 	// Enable OTLP HTTP exporter
 	// For relevant environment variables:
 	// https://pkg.go.dev/go.opentelemetry.io/otel/exporters/otlp/otlptrace#readme-environment-variables
-	// At a minimum, you need to set
-	// OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318
-	if ep := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); ep != "" {
-		log.Info("setting up trace exporter", "endpoint", ep)
+	if cctx.Bool("enable-otel-otlp") {
+		ep := cctx.String("otel-otlp-endpoint")
+		logger.Info("setting up trace exporter", "endpoint", ep)
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
 		exp, err := otlptracehttp.New(ctx)
 		if err != nil {
-			log.Error("failed to create trace exporter", "error", err)
+			logger.Error("failed to create trace exporter", "error", err)
 			os.Exit(1)
 		}
 		defer func() {
 			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 			defer cancel()
 			if err := exp.Shutdown(ctx); err != nil {
-				log.Error("failed to shutdown trace exporter", "error", err)
+				logger.Error("failed to shutdown trace exporter", "error", err)
 			}
 		}()
 
+		env := cctx.String("env")
 		tp := tracesdk.NewTracerProvider(
 			tracesdk.WithBatcher(exp),
 			tracesdk.WithResource(resource.NewWithAttributes(
 				semconv.SchemaURL,
 				semconv.ServiceNameKey.String("splitter"),
-				attribute.String("env", os.Getenv("ENVIRONMENT")),         // DataDog
-				attribute.String("environment", os.Getenv("ENVIRONMENT")), // Others
+				attribute.String("env", env),         // DataDog
+				attribute.String("environment", env), // Others
 				attribute.Int64("ID", 1),
 			)),
 		)
···
 	}
 
 	persistPath := cctx.String("persist-db")
-	upstreamHost := cctx.String("splitter-host")
+	upstreamHost := cctx.String("upstream-host")
+	collectionDirHost := cctx.String("collectiondir-host")
 	nextCrawlers := cctx.StringSlice("next-crawler")
 
 	var spl *splitter.Splitter
 	var err error
 	if persistPath != "" {
-		log.Info("building splitter with storage at", "path", persistPath)
+		logger.Info("building splitter with storage at", "path", persistPath)
 		ppopts := pebblepersist.PebblePersistOptions{
 			DbPath:          persistPath,
 			PersistDuration: time.Duration(float64(time.Hour) * cctx.Float64("persist-hours")),
···
 			MaxBytes:        uint64(cctx.Int64("persist-bytes")),
 		}
 		conf := splitter.SplitterConfig{
-			UpstreamHost:  upstreamHost,
-			CursorFile:    cctx.String("cursor-file"),
-			PebbleOptions: &ppopts,
+			UpstreamHost:      upstreamHost,
+			CollectionDirHost: collectionDirHost,
+			CursorFile:        cctx.String("cursor-file"),
+			PebbleOptions:     &ppopts,
+			UserAgent:         fmt.Sprintf("rainbow/%s", versioninfo.Short()),
 		}
 		spl, err = splitter.NewSplitter(conf, nextCrawlers)
 	} else {
-		log.Info("building in-memory splitter")
+		logger.Info("building in-memory splitter")
 		conf := splitter.SplitterConfig{
-			UpstreamHost: upstreamHost,
-			CursorFile:   cctx.String("cursor-file"),
+			UpstreamHost:      upstreamHost,
+			CollectionDirHost: collectionDirHost,
+			CursorFile:        cctx.String("cursor-file"),
 		}
 		spl, err = splitter.NewSplitter(conf, nextCrawlers)
 	}
 	if err != nil {
-		log.Error("failed to create splitter", "path", persistPath, "error", err)
+		logger.Error("failed to create splitter", "path", persistPath, "error", err)
 		os.Exit(1)
 		return err
 	}
 
 	// set up metrics endpoint
+	metricsListen := cctx.String("metrics-listen")
 	go func() {
-		if err := spl.StartMetrics(cctx.String("metrics-listen")); err != nil {
-			log.Error("failed to start metrics endpoint", "err", err)
+		if err := spl.StartMetrics(metricsListen); err != nil {
+			logger.Error("failed to start metrics endpoint", "err", err)
 			os.Exit(1)
 		}
 	}()
···
 	runErr := make(chan error, 1)
 
 	go func() {
-		err := spl.Start(cctx.String("api-listen"))
+		err := spl.StartAPI(cctx.String("api-listen"))
 		runErr <- err
 	}()
 
-	log.Info("startup complete")
+	logger.Info("startup complete")
 	select {
 	case <-signals:
-		log.Info("received shutdown signal")
+		logger.Info("received shutdown signal")
 		if err := spl.Shutdown(); err != nil {
-			log.Error("error during Splitter shutdown", "err", err)
+			logger.Error("error during Splitter shutdown", "err", err)
 		}
 	case err := <-runErr:
 		if err != nil {
-			log.Error("error during Splitter startup", "err", err)
+			logger.Error("error during Splitter startup", "err", err)
 		}
-		log.Info("shutting down")
+		logger.Info("shutting down")
 		if err := spl.Shutdown(); err != nil {
-			log.Error("error during Splitter shutdown", "err", err)
+			logger.Error("error during Splitter shutdown", "err", err)
 		}
 	}
 
-	log.Info("shutdown complete")
+	logger.Info("shutdown complete")
 
 	return nil
 }

splitter/firehose.go

···
+package splitter
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/bluesky-social/indigo/events"
+
+	"github.com/gorilla/websocket"
+	"github.com/labstack/echo/v4"
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+)
+
+func (s *Splitter) HandleSubscribeRepos(c echo.Context) error {
+	var since *int64
+	if sinceVal := c.QueryParam("cursor"); sinceVal != "" {
+		sval, err := strconv.ParseInt(sinceVal, 10, 64)
+		if err != nil {
+			return err
+		}
+		since = &sval
+	}
+
+	// NOTE: the request context outlives the HTTP 101 response; it lives as long as the WebSocket is open, and then get cancelled. That is the behavior we want for this ctx, but should be careful if spawning goroutines which should outlive the WebSocket connection.
+	// https://github.com/bluesky-social/indigo/pull/1023#pullrequestreview-2768335762
+	ctx, cancel := context.WithCancel(c.Request().Context())
+	defer cancel()
+
+	// TODO: authhhh
+	conn, err := websocket.Upgrade(c.Response(), c.Request(), c.Response().Header(), 10<<10, 10<<10)
+	if err != nil {
+		return fmt.Errorf("upgrading websocket: %w", err)
+	}
+
+	defer conn.Close()
+
+	lastWriteLk := sync.Mutex{}
+	lastWrite := time.Now()
+
+	// Start a goroutine to ping the client every 30 seconds to check if it's
+	// still alive. If the client doesn't respond to a ping within 5 seconds,
+	// we'll close the connection and teardown the consumer.
+	go func() {
+		ticker := time.NewTicker(30 * time.Second)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				lastWriteLk.Lock()
+				lw := lastWrite
+				lastWriteLk.Unlock()
+
+				if time.Since(lw) < 30*time.Second {
+					continue
+				}
+
+				if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil {
+					s.logger.Error("failed to ping client", "err", err)
+					cancel()
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	conn.SetPingHandler(func(message string) error {
+		err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60))
+		if err == websocket.ErrCloseSent {
+			return nil
+		} else if e, ok := err.(net.Error); ok && e.Temporary() {
+			return nil
+		}
+		return err
+	})
+
+	// Start a goroutine to read messages from the client and discard them.
+	go func() {
+		for {
+			_, _, err := conn.ReadMessage()
+			if err != nil {
+				s.logger.Error("failed to read message from client", "err", err)
+				cancel()
+				return
+			}
+		}
+	}()
+
+	ident := c.RealIP() + "-" + c.Request().UserAgent()
+
+	evts, cleanup, err := s.events.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool { return true }, since)
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	// Keep track of the consumer for metrics and admin endpoints
+	consumer := SocketConsumer{
+		RemoteAddr:  c.RealIP(),
+		UserAgent:   c.Request().UserAgent(),
+		ConnectedAt: time.Now(),
+	}
+	sentCounter := eventsSentCounter.WithLabelValues(consumer.RemoteAddr, consumer.UserAgent)
+	consumer.EventsSent = sentCounter
+
+	consumerID := s.registerConsumer(&consumer)
+	defer s.cleanupConsumer(consumerID)
+
+	s.logger.Info("new consumer",
+		"remote_addr", consumer.RemoteAddr,
+		"user_agent", consumer.UserAgent,
+		"cursor", since,
+		"consumer_id", consumerID,
+	)
+	activeClientGauge.Inc()
+	defer activeClientGauge.Dec()
+
+	for {
+		select {
+		case evt, ok := <-evts:
+			if !ok {
+				s.logger.Error("event stream closed unexpectedly")
+				return nil
+			}
+
+			wc, err := conn.NextWriter(websocket.BinaryMessage)
+			if err != nil {
+				s.logger.Error("failed to get next writer", "err", err)
+				return err
+			}
+
+			if evt.Preserialized != nil {
+				_, err = wc.Write(evt.Preserialized)
+			} else {
+				err = evt.Serialize(wc)
+			}
+			if err != nil {
+				return fmt.Errorf("failed to write event: %w", err)
+			}
+
+			if err := wc.Close(); err != nil {
+				s.logger.Warn("failed to flush-close our event write", "err", err)
+				return nil
+			}
+
+			lastWriteLk.Lock()
+			lastWrite = time.Now()
+			lastWriteLk.Unlock()
+			sentCounter.Inc()
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
+type SocketConsumer struct {
+	UserAgent   string
+	RemoteAddr  string
+	ConnectedAt time.Time
+	EventsSent  prometheus.Counter
+}
+
+func (s *Splitter) registerConsumer(c *SocketConsumer) uint64 {
+	s.consumersLk.Lock()
+	defer s.consumersLk.Unlock()
+
+	id := s.nextConsumerID
+	s.nextConsumerID++
+
+	s.consumers[id] = c
+
+	return id
+}
+
+func (s *Splitter) cleanupConsumer(id uint64) {
+	s.consumersLk.Lock()
+	defer s.consumersLk.Unlock()
+
+	c := s.consumers[id]
+
+	var m = &dto.Metric{}
+	if err := c.EventsSent.Write(m); err != nil {
+		s.logger.Error("failed to get sent counter", "err", err)
+	}
+
+	s.logger.Info("consumer disconnected",
+		"consumer_id", id,
+		"remote_addr", c.RemoteAddr,
+		"user_agent", c.UserAgent,
+		"events_sent", m.Counter.GetValue())
+
+	delete(s.consumers, id)
+}

splitter/handlers.go

···
+package splitter
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+
+	comatproto "github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/xrpc"
+
+	"github.com/labstack/echo/v4"
+)
+
+type HealthStatus struct {
+	Service string `json:"service,const=rainbow"`
+	Status  string `json:"status"`
+	Message string `json:"msg,omitempty"`
+}
+
+func (s *Splitter) HandleHealthCheck(c echo.Context) error {
+	return c.JSON(http.StatusOK, HealthStatus{Status: "ok"})
+}
+
+var homeMessage string = `
+          _      _
+ _ _ __ _(_)_ _ | |__  _____ __ __
+| '_/ _' | | ' \| '_ \/ _ \ V  V /
+|_| \__,_|_|_||_|_.__/\___/\_/\_/
+
+This is an atproto [https://atproto.com] firehose fanout service, running the 'rainbow' codebase [https://github.com/bluesky-social/indigo]
+
+The firehose WebSocket path is at:  /xrpc/com.atproto.sync.subscribeRepos
+`
+
+func (s *Splitter) HandleHomeMessage(c echo.Context) error {
+	return c.String(http.StatusOK, homeMessage)
+}
+
+func (s *Splitter) HandleComAtprotoSyncRequestCrawl(c echo.Context) error {
+	ctx := c.Request().Context()
+	var body comatproto.SyncRequestCrawl_Input
+	if err := c.Bind(&body); err != nil {
+		return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("invalid body: %s", err)})
+	}
+	if body.Hostname == "" {
+		return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "must include a hostname"})
+	}
+
+	// first forward to the upstream
+	xrpcc := xrpc.Client{
+		Client: s.upstreamClient,
+		Host:   s.conf.UpstreamHostHTTP(),
+	}
+
+	err := comatproto.SyncRequestCrawl(ctx, &xrpcc, &body)
+	if err != nil {
+		httpError, ok := err.(*xrpc.Error)
+		if ok {
+			return c.JSON(httpError.StatusCode, xrpc.XRPCError{ErrStr: "UpstreamError", Message: fmt.Sprintf("%s", httpError.Wrapped)})
+		}
+		return c.JSON(http.StatusInternalServerError, xrpc.XRPCError{ErrStr: "ProxyRequestFailed", Message: fmt.Sprintf("failed forwarding request: %s", err)})
+	}
+
+	// if that was successful, then forward on to the other upstreams (in goroutines)
+	for _, c := range s.nextCrawlers {
+		// intentional local copy of loop variable
+		hostname := c.String()
+		go func() {
+			// new context to outlive original HTTP request
+			ctx := context.Background()
+			xrpcc := xrpc.Client{
+				Client: s.upstreamClient,
+				Host:   hostname,
+			}
+			if err := comatproto.SyncRequestCrawl(ctx, &xrpcc, &body); err != nil {
+				s.logger.Warn("failed to forward requestCrawl", "upstream", hostname, "targetHost", body.Hostname, "err", err)
+			}
+			s.logger.Info("successfully forwarded requestCrawl", "upstream", hostname, "targetHost", body.Hostname)
+		}()
+	}
+
+	return c.JSON(http.StatusOK, map[string]any{"success": true})
+}
+
+// Proxies a request to the single upstream (relay)
+func (s *Splitter) ProxyRequestUpstream(c echo.Context) error {
+	u, err := url.Parse(s.conf.UpstreamHostHTTP())
+	if err != nil {
+		return err
+	}
+	return s.ProxyRequest(c, u.Host, u.Scheme)
+}
+
+// Proxies a request to the collectiondir
+func (s *Splitter) ProxyRequestCollectionDir(c echo.Context) error {
+	u, err := url.Parse(s.conf.CollectionDirHost)
+	if err != nil {
+		return err
+	}
+	return s.ProxyRequest(c, u.Host, u.Scheme)
+}
+
+func (s *Splitter) ProxyRequest(c echo.Context, hostname, scheme string) error {
+
+	req := c.Request()
+	respWriter := c.Response()
+
+	u := req.URL
+	u.Scheme = scheme
+	u.Host = hostname
+	upstreamReq, err := http.NewRequest(req.Method, u.String(), req.Body)
+	if err != nil {
+		s.logger.Warn("proxy request failed", "err", err)
+		return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "failed to proxy to upstream relay"})
+	}
+
+	for k, vals := range req.Header {
+		if strings.ToLower(k) == "accept" {
+			upstreamReq.Header.Add(k, vals[0])
+		}
+	}
+
+	upstreamResp, err := s.upstreamClient.Do(upstreamReq)
+	if err != nil {
+		return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: "failed to proxy to upstream relay"})
+	}
+	defer upstreamResp.Body.Close()
+
+	respWriter.Header()["Content-Type"] = []string{upstreamResp.Header.Get("Content-Type")}
+	respWriter.WriteHeader(upstreamResp.StatusCode)
+
+	_, err = io.Copy(respWriter, upstreamResp.Body)
+	if err != nil {
+		s.logger.Error("error copying proxy body", "err", err)
+	}
+
+	return nil
+}

splitter/splitter.go

···
 package splitter
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
···
 	"sync"
 	"time"
 
-	"github.com/bluesky-social/indigo/api/atproto"
-	comatproto "github.com/bluesky-social/indigo/api/atproto"
 	"github.com/bluesky-social/indigo/events"
 	"github.com/bluesky-social/indigo/events/pebblepersist"
 	"github.com/bluesky-social/indigo/events/schedulers/sequential"
 	"github.com/bluesky-social/indigo/util"
 	"github.com/bluesky-social/indigo/util/svcutil"
-	"github.com/bluesky-social/indigo/xrpc"
 
 	"github.com/gorilla/websocket"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
-	dto "github.com/prometheus/client_model/go"
-	"go.opentelemetry.io/otel"
 )
 
 type Splitter struct {
···
 
 	conf SplitterConfig
 
-	log *slog.Logger
+	logger *slog.Logger
 
-	httpC        *http.Client
-	nextCrawlers []*url.URL
+	upstreamClient *http.Client
+	nextCrawlers   []*url.URL
 }
 
 type SplitterConfig struct {
-	UpstreamHost  string
-	CursorFile    string
-	PebbleOptions *pebblepersist.PebblePersistOptions
+	UpstreamHost      string
+	CollectionDirHost string
+	CursorFile        string
+	UserAgent         string
+	PebbleOptions     *pebblepersist.PebblePersistOptions
+	Logger            *slog.Logger
 }
 
-func (sc *SplitterConfig) XrpcRootUrl() string {
-	if strings.HasPrefix(sc.UpstreamHost, "http://") {
-		return sc.UpstreamHost
+func (sc *SplitterConfig) UpstreamHostWebsocket() string {
+
+	if !strings.Contains(sc.UpstreamHost, "://") {
+		return "wss://" + sc.UpstreamHost
 	}
-	if strings.HasPrefix(sc.UpstreamHost, "https://") {
-		return sc.UpstreamHost
+	u, err := url.Parse(sc.UpstreamHost)
+	if err != nil {
+		// this will cause an error downstream
+		return ""
 	}
-	if strings.HasPrefix(sc.UpstreamHost, "ws://") {
-		return "http://" + sc.UpstreamHost[5:]
-	}
-	if strings.HasPrefix(sc.UpstreamHost, "wss://") {
-		return "https://" + sc.UpstreamHost[6:]
+
+	switch u.Scheme {
+	case "http", "ws":
+		return "ws://" + u.Host
+	case "https", "wss":
+		return "wss://" + u.Host
+	default:
+		return "wss://" + u.Host
 	}
-	return "https://" + sc.UpstreamHost
 }
 
-func (sc *SplitterConfig) UpstreamUrl() string {
-	if strings.HasPrefix(sc.UpstreamHost, "ws://") {
-		return "http://" + sc.UpstreamHost[7:]
+func (sc *SplitterConfig) UpstreamHostHTTP() string {
+
+	if !strings.Contains(sc.UpstreamHost, "://") {
+		return "https://" + sc.UpstreamHost
 	}
-	if strings.HasPrefix(sc.UpstreamHost, "wss://") {
-		return "https://" + sc.UpstreamHost[8:]
+	u, err := url.Parse(sc.UpstreamHost)
+	if err != nil {
+		// this will cause an error downstream
+		return ""
 	}
-	if strings.HasPrefix(sc.UpstreamHost, "ws://") {
-		return sc.UpstreamHost
+
+	switch u.Scheme {
+	case "http", "ws":
+		return "http://" + u.Host
+	case "https", "wss":
+		return "https://" + u.Host
+	default:
+		return "https://" + u.Host
 	}
-	if strings.HasPrefix(sc.UpstreamHost, "wss://") {
-		return sc.UpstreamHost
-	}
-	return "wss://" + sc.UpstreamHost
 }
 
 func NewSplitter(conf SplitterConfig, nextCrawlers []string) (*Splitter, error) {
+
+	logger := conf.Logger
+	if logger == nil {
+		logger = slog.Default().With("system", "splitter")
+	}
+
 	var nextCrawlerURLs []*url.URL
-	log := slog.Default().With("system", "splitter")
 	if len(nextCrawlers) > 0 {
 		nextCrawlerURLs = make([]*url.URL, len(nextCrawlers))
 		for i, tu := range nextCrawlers {
···
 			if err != nil {
 				return nil, fmt.Errorf("failed to parse next-crawler url: %w", err)
 			}
-			log.Info("configuring relay for requestCrawl", "host", nextCrawlerURLs[i])
+			logger.Info("configuring relay for requestCrawl", "host", nextCrawlerURLs[i])
 		}
 	}
 
-	_, err := url.Parse(conf.UpstreamUrl())
+	_, err := url.Parse(conf.UpstreamHostHTTP())
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse upstream url %#v: %w", conf.UpstreamUrl(), err)
+		return nil, fmt.Errorf("failed to parse upstream url %#v: %w", conf.UpstreamHostHTTP(), err)
 	}
 
 	s := &Splitter{
-		conf:         conf,
-		consumers:    make(map[uint64]*SocketConsumer),
-		log:          log,
-		httpC:        util.RobustHTTPClient(),
-		nextCrawlers: nextCrawlerURLs,
+		conf:           conf,
+		consumers:      make(map[uint64]*SocketConsumer),
+		logger:         logger,
+		upstreamClient: util.RobustHTTPClient(),
+		nextCrawlers:   nextCrawlerURLs,
 	}
 
 	if conf.PebbleOptions == nil {
···
 
 	return s, nil
 }
-func NewDiskSplitter(host, path string, persistHours float64, maxBytes int64) (*Splitter, error) {
-	ppopts := pebblepersist.PebblePersistOptions{
-		DbPath:          path,
-		PersistDuration: time.Duration(float64(time.Hour) * persistHours),
-		GCPeriod:        5 * time.Minute,
-		MaxBytes:        uint64(maxBytes),
-	}
-	conf := SplitterConfig{
-		UpstreamHost:  host,
-		CursorFile:    "cursor-file",
-		PebbleOptions: &ppopts,
-	}
-	pp, err := pebblepersist.NewPebblePersistance(&ppopts)
-	if err != nil {
-		return nil, err
-	}
-
-	go pp.GCThread(context.Background())
-	em := events.NewEventManager(pp)
-	return &Splitter{
-		conf:      conf,
-		pp:        pp,
-		events:    em,
-		consumers: make(map[uint64]*SocketConsumer),
-		log:       slog.Default().With("system", "splitter"),
-	}, nil
-}
 
-func (s *Splitter) Start(addr string) error {
+func (s *Splitter) StartAPI(addr string) error {
 	var lc net.ListenConfig
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
 	defer cancel()
···
 	if err != nil {
 		return err
 	}
-	return s.StartWithListener(li)
+	return s.startWithListener(li)
 }
 
 func (s *Splitter) StartMetrics(listen string) error {
···
 	return nil
 }
 
-func (s *Splitter) StartWithListener(listen net.Listener) error {
+func (s *Splitter) startWithListener(listen net.Listener) error {
 	e := echo.New()
 	e.HideBanner = true
 
···
 	*/
 
 	e.Use(svcutil.MetricsMiddleware)
+	e.HTTPErrorHandler = s.errorHandler
 
-	e.HTTPErrorHandler = func(err error, ctx echo.Context) {
-		switch err := err.(type) {
-		case *echo.HTTPError:
-			if err2 := ctx.JSON(err.Code, map[string]any{
-				"error": err.Message,
-			}); err2 != nil {
-				s.log.Error("Failed to write http error", "err", err2)
-			}
-		default:
-			sendHeader := true
-			if ctx.Path() == "/xrpc/com.atproto.sync.subscribeRepos" {
-				sendHeader = false
-			}
+	e.POST("/xrpc/com.atproto.sync.requestCrawl", s.HandleComAtprotoSyncRequestCrawl)
+	e.GET("/xrpc/com.atproto.sync.subscribeRepos", s.HandleSubscribeRepos)
 
-			s.log.Warn("HANDLER ERROR", "path", ctx.Path(), "err", err)
-
-			if strings.HasPrefix(ctx.Path(), "/admin/") {
-				ctx.JSON(500, map[string]any{
-					"error": err.Error(),
-				})
-				return
-			}
-
-			if sendHeader {
-				ctx.Response().WriteHeader(500)
-			}
-		}
-	}
-
-	// TODO: this API is temporary until we formalize what we want here
+	// proxy endpoints to upstream (relay)
+	e.GET("/xrpc/com.atproto.sync.listRepos", s.ProxyRequestUpstream)
+	e.GET("/xrpc/com.atproto.sync.getRepoStatus", s.ProxyRequestUpstream)
+	e.GET("/xrpc/com.atproto.sync.getLatestCommit", s.ProxyRequestUpstream)
+	e.GET("/xrpc/com.atproto.sync.listHosts", s.ProxyRequestUpstream)
+	e.GET("/xrpc/com.atproto.sync.getHostStatus", s.ProxyRequestUpstream)
+	e.GET("/xrpc/com.atproto.sync.getRepo", s.ProxyRequestUpstream)
 
-	e.POST("/xrpc/com.atproto.sync.requestCrawl", s.RequestCrawlHandler)
-	e.GET("/xrpc/com.atproto.sync.subscribeRepos", s.EventsHandler)
-	e.GET("/xrpc/com.atproto.sync.listRepos", s.HandleComAtprotoSyncListRepos)
+	// proxy endpoint to collectiondir
+	e.GET("/xrpc/com.atproto.sync.listReposByCollection", s.ProxyRequestCollectionDir)
 
 	e.GET("/xrpc/_health", s.HandleHealthCheck)
 	e.GET("/_health", s.HandleHealthCheck)
···
 	return e.StartServer(srv)
 }
 
-type HealthStatus struct {
-	Status  string `json:"status"`
-	Message string `json:"msg,omitempty"`
-}
-
-func (s *Splitter) HandleHealthCheck(c echo.Context) error {
-	return c.JSON(200, HealthStatus{Status: "ok"})
-}
-
-var homeMessage string = `
-          _      _
- _ _ __ _(_)_ _ | |__  _____ __ __
-| '_/ _' | | ' \| '_ \/ _ \ V  V /
-|_| \__,_|_|_||_|_.__/\___/\_/\_/
-
-This is an atproto [https://atproto.com] firehose fanout service, running the 'rainbow' codebase [https://github.com/bluesky-social/indigo]
-
-The firehose WebSocket path is at:  /xrpc/com.atproto.sync.subscribeRepos
-`
-
-func (s *Splitter) HandleHomeMessage(c echo.Context) error {
-	return c.String(http.StatusOK, homeMessage)
-}
-
-type XRPCError struct {
-	Message string `json:"message"`
-}
-
-func (s *Splitter) RequestCrawlHandler(c echo.Context) error {
-	ctx := c.Request().Context()
-	var body comatproto.SyncRequestCrawl_Input
-	if err := c.Bind(&body); err != nil {
-		return c.JSON(http.StatusBadRequest, XRPCError{Message: fmt.Sprintf("invalid body: %s", err)})
-	}
-
-	host := body.Hostname
-	if host == "" {
-		return echo.NewHTTPError(http.StatusBadRequest, "must pass hostname")
-	}
-
-	if !strings.HasPrefix(host, "http://") && !strings.HasPrefix(host, "https://") {
-		host = "https://" + host
-	}
-
-	u, err := url.Parse(host)
-	if err != nil {
-		return echo.NewHTTPError(http.StatusBadRequest, "failed to parse hostname")
-	}
-
-	if u.Scheme == "http" {
-		return echo.NewHTTPError(http.StatusBadRequest, "this server requires https")
-	}
-	if u.Path != "" {
-		return echo.NewHTTPError(http.StatusBadRequest, "must pass hostname without path")
-	}
-
-	if u.Query().Encode() != "" {
-		return echo.NewHTTPError(http.StatusBadRequest, "must pass hostname without query")
-	}
-
-	host = u.Host // potentially hostname:port
-
-	clientHost := fmt.Sprintf("%s://%s", u.Scheme, host)
-
-	xrpcC := &xrpc.Client{
-		Host:   clientHost,
-		Client: http.DefaultClient, // not using the client that auto-retries
-	}
-
-	desc, err := atproto.ServerDescribeServer(ctx, xrpcC)
-	if err != nil {
-		errMsg := fmt.Sprintf("requested host (%s) failed to respond to describe request", clientHost)
-		return echo.NewHTTPError(http.StatusBadRequest, errMsg)
-	}
+func (s *Splitter) errorHandler(err error, ctx echo.Context) {
+	switch err := err.(type) {
+	case *echo.HTTPError:
+		if err2 := ctx.JSON(err.Code, map[string]any{
+			"error": err.Message,
+		}); err2 != nil {
+			s.logger.Error("Failed to write http error", "err", err2)
+		}
+	default:
+		sendHeader := true
+		if ctx.Path() == "/xrpc/com.atproto.sync.subscribeRepos" {
+			sendHeader = false
+		}
 
-	// Maybe we could do something with this response later
-	_ = desc
+		s.logger.Warn("HANDLER ERROR", "path", ctx.Path(), "err", err)
 
-	if len(s.nextCrawlers) != 0 {
-		blob, err := json.Marshal(body)
-		if err != nil {
-			s.log.Warn("could not forward requestCrawl, json err", "err", err)
-		} else {
-			go func(bodyBlob []byte) {
-				for _, remote := range s.nextCrawlers {
-					if remote == nil {
-						continue
-					}
+		if strings.HasPrefix(ctx.Path(), "/admin/") {
+			ctx.JSON(500, map[string]any{
+				"error": err.Error(),
+			})
+			return
+		}
 
-					pu := remote.JoinPath("/xrpc/com.atproto.sync.requestCrawl")
-					response, err := s.httpC.Post(pu.String(), "application/json", bytes.NewReader(bodyBlob))
-					if response != nil && response.Body != nil {
-						response.Body.Close()
-					}
-					if err != nil || response == nil {
-						s.log.Warn("requestCrawl forward failed", "host", remote, "err", err)
-					} else if response.StatusCode != http.StatusOK {
-						s.log.Warn("requestCrawl forward failed", "host", remote, "status", response.Status)
-					} else {
-						s.log.Info("requestCrawl forward successful", "host", remote)
-					}
-				}
-			}(blob)
+		if sendHeader {
+			ctx.Response().WriteHeader(500)
 		}
 	}
-
-	return c.JSON(200, HealthStatus{Status: "ok"})
 }
 
-func (s *Splitter) HandleComAtprotoSyncListRepos(c echo.Context) error {
-	ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleComAtprotoSyncListRepos")
-	defer span.End()
-
-	cursorQuery := c.QueryParam("cursor")
-	limitQuery := c.QueryParam("limit")
-
-	var err error
-
-	limit := int64(500)
-	if limitQuery != "" {
-		limit, err = strconv.ParseInt(limitQuery, 10, 64)
-		if err != nil || limit < 1 || limit > 1000 {
-			return c.JSON(http.StatusBadRequest, XRPCError{Message: fmt.Sprintf("invalid limit: %s", limitQuery)})
+func (s *Splitter) getLastCursor() (int64, error) {
+	if s.pp != nil {
+		seq, millis, _, err := s.pp.GetLast(context.Background())
+		if err == nil {
+			s.logger.Debug("got last cursor from pebble", "seq", seq, "millis", millis)
+			return seq, nil
+		} else if errors.Is(err, pebblepersist.ErrNoLast) {
+			s.logger.Info("pebble no last")
+		} else {
+			s.logger.Error("pebble seq fail", "err", err)
 		}
 	}
 
-	client := xrpc.Client{
-		Client: s.httpC,
-		Host:   s.conf.XrpcRootUrl(),
-	}
-
-	out, handleErr := atproto.SyncListRepos(ctx, &client, cursorQuery, limit)
-	if handleErr != nil {
-		return handleErr
-	}
-	return c.JSON(200, out)
-}
-
-func (s *Splitter) EventsHandler(c echo.Context) error {
-	var since *int64
-	if sinceVal := c.QueryParam("cursor"); sinceVal != "" {
-		sval, err := strconv.ParseInt(sinceVal, 10, 64)
-		if err != nil {
-			return err
+	fi, err := os.Open(s.conf.CursorFile)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return -1, nil
 		}
-		since = &sval
+		return -1, err
 	}
 
-	ctx, cancel := context.WithCancel(c.Request().Context())
-	defer cancel()
-
-	// TODO: authhhh
-	conn, err := websocket.Upgrade(c.Response(), c.Request(), c.Response().Header(), 10<<10, 10<<10)
+	b, err := io.ReadAll(fi)
 	if err != nil {
-		return fmt.Errorf("upgrading websocket: %w", err)
+		return -1, err
 	}
 
-	defer conn.Close()
-
-	lastWriteLk := sync.Mutex{}
-	lastWrite := time.Now()
-
-	// Start a goroutine to ping the client every 30 seconds to check if it's
-	// still alive. If the client doesn't respond to a ping within 5 seconds,
-	// we'll close the connection and teardown the consumer.
-	go func() {
-		ticker := time.NewTicker(30 * time.Second)
-		defer ticker.Stop()
-
-		for {
-			select {
-			case <-ticker.C:
-				lastWriteLk.Lock()
-				lw := lastWrite
-				lastWriteLk.Unlock()
-
-				if time.Since(lw) < 30*time.Second {
-					continue
-				}
-
-				if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil {
-					s.log.Error("failed to ping client", "err", err)
-					cancel()
-					return
-				}
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
-
-	conn.SetPingHandler(func(message string) error {
-		err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60))
-		if err == websocket.ErrCloseSent {
-			return nil
-		} else if e, ok := err.(net.Error); ok && e.Temporary() {
-			return nil
-		}
-		return err
-	})
-
-	// Start a goroutine to read messages from the client and discard them.
-	go func() {
-		for {
-			_, _, err := conn.ReadMessage()
-			if err != nil {
-				s.log.Error("failed to read message from client", "err", err)
-				cancel()
-				return
-			}
-		}
-	}()
-
-	ident := c.RealIP() + "-" + c.Request().UserAgent()
-
-	evts, cleanup, err := s.events.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool { return true }, since)
+	v, err := strconv.ParseInt(string(b), 10, 64)
 	if err != nil {
-		return err
-	}
-	defer cleanup()
-
-	// Keep track of the consumer for metrics and admin endpoints
-	consumer := SocketConsumer{
-		RemoteAddr:  c.RealIP(),
-		UserAgent:   c.Request().UserAgent(),
-		ConnectedAt: time.Now(),
-	}
-	sentCounter := eventsSentCounter.WithLabelValues(consumer.RemoteAddr, consumer.UserAgent)
-	consumer.EventsSent = sentCounter
-
-	consumerID := s.registerConsumer(&consumer)
-	defer s.cleanupConsumer(consumerID)
-
-	s.log.Info("new consumer",
-		"remote_addr", consumer.RemoteAddr,
-		"user_agent", consumer.UserAgent,
-		"cursor", since,
-		"consumer_id", consumerID,
-	)
-	activeClientGauge.Inc()
-	defer activeClientGauge.Dec()
-
-	for {
-		select {
-		case evt, ok := <-evts:
-			if !ok {
-				s.log.Error("event stream closed unexpectedly")
-				return nil
-			}
-
-			wc, err := conn.NextWriter(websocket.BinaryMessage)
-			if err != nil {
-				s.log.Error("failed to get next writer", "err", err)
-				return err
-			}
-
-			if evt.Preserialized != nil {
-				_, err = wc.Write(evt.Preserialized)
-			} else {
-				err = evt.Serialize(wc)
-			}
-			if err != nil {
-				return fmt.Errorf("failed to write event: %w", err)
-			}
-
-			if err := wc.Close(); err != nil {
-				s.log.Warn("failed to flush-close our event write", "err", err)
-				return nil
-			}
-
-			lastWriteLk.Lock()
-			lastWrite = time.Now()
-			lastWriteLk.Unlock()
-			sentCounter.Inc()
-		case <-ctx.Done():
-			return nil
-		}
+		return -1, err
 	}
-}
 
-type SocketConsumer struct {
-	UserAgent   string
-	RemoteAddr  string
-	ConnectedAt time.Time
-	EventsSent  prometheus.Counter
-}
-
-func (s *Splitter) registerConsumer(c *SocketConsumer) uint64 {
-	s.consumersLk.Lock()
-	defer s.consumersLk.Unlock()
-
-	id := s.nextConsumerID
-	s.nextConsumerID++
-
-	s.consumers[id] = c
-
-	return id
+	return v, nil
 }
 
-func (s *Splitter) cleanupConsumer(id uint64) {
-	s.consumersLk.Lock()
-	defer s.consumersLk.Unlock()
-
-	c := s.consumers[id]
-
-	var m = &dto.Metric{}
-	if err := c.EventsSent.Write(m); err != nil {
-		s.log.Error("failed to get sent counter", "err", err)
-	}
-
-	s.log.Info("consumer disconnected",
-		"consumer_id", id,
-		"remote_addr", c.RemoteAddr,
-		"user_agent", c.UserAgent,
-		"events_sent", m.Counter.GetValue())
-
-	delete(s.consumers, id)
+func (s *Splitter) writeCursor(curs int64) error {
+	return os.WriteFile(s.conf.CursorFile, []byte(fmt.Sprint(curs)), 0664)
 }
 
 func sleepForBackoff(b int) time.Duration {
···
 func (s *Splitter) subscribeWithRedialer(ctx context.Context, cursor int64) {
 	d := websocket.Dialer{}
 
-	upstreamUrl, err := url.Parse(s.conf.UpstreamUrl())
+	upstreamUrl, err := url.Parse(s.conf.UpstreamHostWebsocket())
 	if err != nil {
 		panic(err) // this should have been checked in NewSplitter
 	}
 	upstreamUrl = upstreamUrl.JoinPath("/xrpc/com.atproto.sync.subscribeRepos")
+
+	header := http.Header{
+		"User-Agent": []string{s.conf.UserAgent},
+	}
 
 	var backoff int
 	for {
···
 		default:
 		}
 
-		header := http.Header{
-			"User-Agent": []string{"bgs-rainbow-v0"},
-		}
-
 		var uurl string
 		if cursor < 0 {
 			upstreamUrl.RawQuery = ""
···
 		}
 		con, res, err := d.DialContext(ctx, uurl, header)
 		if err != nil {
-			s.log.Warn("dialing failed", "url", uurl, "err", err, "backoff", backoff)
+			s.logger.Warn("dialing failed", "url", uurl, "err", err, "backoff", backoff)
 			time.Sleep(sleepForBackoff(backoff))
 			backoff++
 
 			continue
 		}
 
-		s.log.Info("event subscription response", "code", res.StatusCode)
+		s.logger.Info("event subscription response", "code", res.StatusCode)
 
-		if err := s.handleConnection(ctx, con, &cursor); err != nil {
-			s.log.Warn("connection failed", "url", uurl, "err", err)
+		if err := s.handleUpstreamConnection(ctx, con, &cursor); err != nil {
+			s.logger.Warn("upstream connection failed", "url", uurl, "err", err)
 		}
 	}
 }
 
-func (s *Splitter) handleConnection(ctx context.Context, con *websocket.Conn, lastCursor *int64) error {
+func (s *Splitter) handleUpstreamConnection(ctx context.Context, con *websocket.Conn, lastCursor *int64) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
···
 		if seq%5000 == 0 {
 			// TODO: don't need this after we move to getting seq from pebble
 			if err := s.writeCursor(seq); err != nil {
-				s.log.Error("write cursor failed", "err", err)
+				s.logger.Error("write cursor failed", "err", err)
 			}
 		}
 
···
 
 	return events.HandleRepoStream(ctx, con, sched, nil)
 }
-
-func (s *Splitter) getLastCursor() (int64, error) {
-	if s.pp != nil {
-		seq, millis, _, err := s.pp.GetLast(context.Background())
-		if err == nil {
-			s.log.Debug("got last cursor from pebble", "seq", seq, "millis", millis)
-			return seq, nil
-		} else if errors.Is(err, pebblepersist.ErrNoLast) {
-			s.log.Info("pebble no last")
-		} else {
-			s.log.Error("pebble seq fail", "err", err)
-		}
-	}
-
-	fi, err := os.Open(s.conf.CursorFile)
-	if err != nil {
-		if os.IsNotExist(err) {
-			return -1, nil
-		}
-		return -1, err
-	}
-
-	b, err := io.ReadAll(fi)
-	if err != nil {
-		return -1, err
-	}
-
-	v, err := strconv.ParseInt(string(b), 10, 64)
-	if err != nil {
-		return -1, err
-	}
-
-	return v, nil
-}
-
-func (s *Splitter) writeCursor(curs int64) error {
-	return os.WriteFile(s.conf.CursorFile, []byte(fmt.Sprint(curs)), 0664)
-}

util/svcutil/logger.go

···
+package svcutil
+
+import (
+	"io"
+	"log/slog"
+	"strings"
+
+	"github.com/urfave/cli/v2"
+)
+
+func ConfigLogger(cctx *cli.Context, writer io.Writer) *slog.Logger {
+	var level slog.Level
+	switch strings.ToLower(cctx.String("log-level")) {
+	case "error":
+		level = slog.LevelError
+	case "warn":
+		level = slog.LevelWarn
+	case "info":
+		level = slog.LevelInfo
+	case "debug":
+		level = slog.LevelDebug
+	default:
+		level = slog.LevelInfo
+	}
+	logger := slog.New(slog.NewJSONHandler(writer, &slog.HandlerOptions{
+		Level: level,
+	}))
+	slog.SetDefault(logger)
+	return logger
+}