add ignored/watched collections and services (#1)

* add ignored/watched collections/services

* start ignoring/watching collections and services

* fixes

* fixes

* cleanup

atkafka/atkafka.go

···
 	bootstrapServers []string
 	outputTopic      string
 	ospreyCompat     bool
-	logger           *slog.Logger
 
-	producer *Producer
+	watchedServices []string
+	ignoredServices []string
 
+	watchedCollections []string
+	ignoredCollections []string
+
+	producer  *Producer
 	plcClient *PlcClient
+	logger    *slog.Logger
 }
 
 type ServerArgs struct {
-	RelayHost        string
-	PlcHost          string
+	// network params
+	RelayHost string
+	PlcHost   string
+
+	// for watched and ignoed services or collections, only one list may be supplied
+	// for both services and collections, wildcards are acceptable. for example:
+	// app.bsky.* will watch/ignore any collection that falls under the app.bsky namespace.
+	// *.bsky.network will watch/ignore any event that falls under the bsky.network list of PDSes
+
+	// list of services that are events will be emitted for
+	WatchedServices []string
+	// list of services that events are ignored for
+	IgnoredServices []string
+
+	// list of collections that events are emitted for
+	WatchedCollections []string
+	// list of collections that events are ignored for
+	IgnoredCollections []string
+
+	// kafka params
 	BootstrapServers []string
 	OutputTopic      string
-	OspreyCompat     bool
-	Logger           *slog.Logger
+
+	// osprey-specific params
+	OspreyCompat bool
+
+	// other
+	Logger *slog.Logger
 }
 
-func NewServer(args *ServerArgs) *Server {
+func NewServer(args *ServerArgs) (*Server, error) {
 	if args.Logger == nil {
 		args.Logger = slog.Default()
 	}
 
+	if len(args.WatchedServices) > 0 && len(args.IgnoredServices) > 0 {
+		return nil, fmt.Errorf("you may only specify a list of watched services _or_ ignored services, not both")
+	}
+
+	if (len(args.WatchedServices) > 0 || len(args.IgnoredServices) > 0) && args.PlcHost == "" {
+		return nil, fmt.Errorf("unable to support watched/ignored services without specifying a PLC host")
+	}
+
+	if len(args.WatchedCollections) > 0 && len(args.IgnoredCollections) > 0 {
+		return nil, fmt.Errorf("you may only specify a list of watched collections _or_ ignored collections, not both")
+	}
+
 	var plcClient *PlcClient
 	if args.PlcHost != "" {
 		plcClient = NewPlcClient(&PlcClientArgs{
···
 		})
 	}
 
-	return &Server{
+	s := &Server{
 		relayHost:        args.RelayHost,
 		plcClient:        plcClient,
 		bootstrapServers: args.BootstrapServers,
···
 		ospreyCompat:     args.OspreyCompat,
 		logger:           args.Logger,
 	}
+
+	if len(args.WatchedServices) > 0 {
+		watchedServices := make([]string, 0, len(args.WatchedServices))
+		for _, service := range args.WatchedServices {
+			watchedServices = append(watchedServices, strings.TrimPrefix(strings.TrimPrefix(service, "*."), "."))
+		}
+		s.watchedServices = watchedServices
+	} else if len(args.IgnoredServices) > 0 {
+		ignoredServices := make([]string, 0, len(args.IgnoredServices))
+		for _, service := range args.IgnoredServices {
+			ignoredServices = append(ignoredServices, strings.TrimPrefix(strings.TrimPrefix(service, "*."), "."))
+		}
+		s.ignoredServices = ignoredServices
+	}
+
+	if len(args.WatchedCollections) > 0 {
+		watchedCollections := make([]string, 0, len(args.WatchedCollections))
+		for _, collection := range args.WatchedCollections {
+			watchedCollections = append(watchedCollections, strings.TrimSuffix(strings.TrimSuffix(collection, ".*"), "."))
+		}
+		s.watchedCollections = watchedCollections
+	} else if len(args.IgnoredCollections) > 0 {
+		ignoredCollections := make([]string, 0, len(args.IgnoredCollections))
+		for _, collection := range args.IgnoredCollections {
+			ignoredCollections = append(ignoredCollections, strings.TrimSuffix(strings.TrimSuffix(collection, ".*"), "."))
+		}
+		s.ignoredCollections = ignoredCollections
+	}
+
+	return s, nil
 }
 
 func (s *Server) Run(ctx context.Context) error {
···
 	return nil
 }
 
-type EventMetadata struct {
-	DidDocument  *identity.DIDDocument `json:"didDocument,omitempty"`
-	PdsHost      string                `json:"pdsHost,omitempty"`
-	Handle       string                `json:"handle,omitempty"`
-	DidCreatedAt string                `json:"didCreatedAt,omitempty"`
-	AccountAge   int64                 `json:"accountAge"`
-}
-
-func (s *Server) FetchEventMetadata(ctx context.Context, did string) (*EventMetadata, error) {
-	var didDocument *identity.DIDDocument
+func (s *Server) FetchEventMetadata(ctx context.Context, did string) (*EventMetadata, *identity.Identity, error) {
+	var ident *identity.Identity
+	var didDocument identity.DIDDocument
 	var pdsHost string
 	var handle string
 	var didCreatedAt string
···
 	if s.plcClient != nil {
 		wg.Go(func() {
 			logger := s.logger.With("component", "didDoc")
-			doc, err := s.plcClient.GetDIDDoc(ctx, did)
+			var err error
+			ident, err = s.plcClient.GetIdentity(ctx, did)
 			if err != nil {
 				logger.Error("error fetching did doc", "did", did, "err", err)
 				return
 			}
-			didDocument = doc
-
-			for _, svc := range doc.Service {
-				if svc.ID == "#atproto_pds" {
-					pdsHost = svc.ServiceEndpoint
-					break
-				}
-			}
-
-			for _, aka := range doc.AlsoKnownAs {
-				if strings.HasPrefix(aka, "at://") {
-					handle = strings.TrimPrefix(aka, "at://")
-					break
-				}
-			}
+			didDocument = ident.DIDDocument()
+			pdsHost = ident.PDSEndpoint()
+			handle = ident.Handle.String()
 		})
 
 		wg.Go(func() {
···
 		Handle:       handle,
 		DidCreatedAt: didCreatedAt,
 		AccountAge:   accountAge,
-	}, nil
+	}, ident, nil
 }
 
 func (s *Server) handleEvent(ctx context.Context, evt *events.XRPCStreamEvent) error {
···
 	var collection string
 	var actionName string
 
+	var evtKey string
+	var evtsToProduce [][]byte
+
 	if evt.RepoCommit != nil {
+		// key events by DID
+		evtKey = evt.RepoCommit.Repo
+
 		// read the repo
 		rr, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.RepoCommit.Blocks))
 		if err != nil {
···
 			return nil
 		}
 
+		eventMetadata, ident, err := s.FetchEventMetadata(dispatchCtx, evt.RepoCommit.Repo)
+		if err != nil {
+			logger.Error("error fetching event metadata", "err", err)
+		} else if ident != nil {
+			skip := false
+			pdsEndpoint := ident.PDSEndpoint()
+			u, err := url.Parse(pdsEndpoint)
+			if err != nil {
+				return fmt.Errorf("failed to parse pds host: %w", err)
+			}
+			pdsHost := u.Hostname()
+
+			if pdsHost != "" {
+				if len(s.watchedServices) > 0 {
+					skip = true
+					for _, watchedService := range s.watchedServices {
+						if watchedService == pdsHost || strings.HasSuffix(pdsHost, "."+watchedService) {
+							skip = false
+							break
+						}
+					}
+				} else if len(s.ignoredServices) > 0 {
+					for _, ignoredService := range s.ignoredServices {
+						if ignoredService == pdsHost || strings.HasSuffix(pdsHost, "."+ignoredService) {
+							skip = true
+							break
+						}
+					}
+				}
+			}
+
+			if skip {
+				logger.Debug("skipping event based on pds host", "pdsHost", pdsHost)
+				return nil
+			}
+		}
+
 		for _, op := range evt.RepoCommit.Ops {
 			kind := repomgr.EventKind(op.Action)
 			collection = strings.Split(op.Path, "/")[0]
 			rkey := strings.Split(op.Path, "/")[1]
 			atUri := fmt.Sprintf("at://%s/%s/%s", evt.RepoCommit.Repo, collection, rkey)
 
+			skip := false
+			if len(s.watchedCollections) > 0 {
+				skip = true
+				for _, watchedCollection := range s.watchedCollections {
+					if watchedCollection == collection || strings.HasPrefix(collection, watchedCollection+".") {
+						skip = false
+						break
+					}
+				}
+			} else if len(s.ignoredCollections) > 0 {
+				for _, ignoredCollection := range s.ignoredCollections {
+					if ignoredCollection == collection || strings.HasPrefix(collection, ignoredCollection+".") {
+						skip = true
+						break
+					}
+				}
+			}
+
+			if skip {
+				logger.Debug("skipping event based on collection", "collection", collection)
+				continue
+			}
+
 			kindStr := "create"
 			switch kind {
 			case repomgr.EvtKindUpdateRecord:
···
 				Operation: &atkOp,
 			}
 
-			eventMetadata, err := s.FetchEventMetadata(dispatchCtx, evt.RepoCommit.Repo)
-			if err != nil {
-				logger.Error("error fetching event metadata", "err", err)
-			} else {
+			if eventMetadata != nil {
 				kafkaEvt.Metadata = eventMetadata
 			}
 
-			var kafkaEvtBytes []byte
+			var evtBytes []byte
 			if s.ospreyCompat {
 				// create the wrapper event for osprey
 				ospreyKafkaEvent := OspreyAtKafkaEvent{
···
 					SendTime: time.Now().Format(time.RFC3339),
 				}
 
-				kafkaEvtBytes, err = json.Marshal(&ospreyKafkaEvent)
+				evtBytes, err = json.Marshal(&ospreyKafkaEvent)
 			} else {
-				kafkaEvtBytes, err = json.Marshal(&kafkaEvt)
+				evtBytes, err = json.Marshal(&kafkaEvt)
 			}
 			if err != nil {
 				return fmt.Errorf("failed to marshal kafka event: %w", err)
 			}
 
-			if err := s.produceAsync(ctx, evt.RepoCommit.Repo, kafkaEvtBytes); err != nil {
-				return err
-			}
+			evtsToProduce = append(evtsToProduce, evtBytes)
 		}
 	} else {
 		defer func() {
···
 		}
 
 		if did != "" {
-			eventMetadata, err := s.FetchEventMetadata(dispatchCtx, did)
+			// key events by DID
+			evtKey = did
+			eventMetadata, ident, err := s.FetchEventMetadata(dispatchCtx, did)
 			if err != nil {
 				logger.Error("error fetching event metadata", "err", err)
-			} else {
+			} else if ident != nil {
+				skip := false
+				pdsEndpoint := ident.PDSEndpoint()
+				u, err := url.Parse(pdsEndpoint)
+				if err != nil {
+					return fmt.Errorf("failed to parse pds host: %w", err)
+				}
+				pdsHost := u.Hostname()
+
+				if pdsHost != "" {
+					if len(s.watchedServices) > 0 {
+						skip = true
+						for _, watchedService := range s.watchedServices {
+							if watchedService == pdsHost || strings.HasSuffix(pdsHost, "."+watchedService) {
+								skip = false
+								break
+							}
+						}
+					} else if len(s.ignoredServices) > 0 {
+						for _, ignoredService := range s.ignoredServices {
+							if ignoredService == pdsHost || strings.HasSuffix(pdsHost, "."+ignoredService) {
+								skip = true
+								break
+							}
+						}
+					}
+				}
+
+				if skip {
+					logger.Debug("skipping event based on pds host", "pdsHost", pdsHost)
+					return nil
+				}
+
 				kafkaEvt.Metadata = eventMetadata
 			}
+		} else {
+			// key events without a DID by "unknown"
+			evtKey = "<unknown>"
 		}
 
 		// create the kafka event bytes
-		var kafkaEvtBytes []byte
+		var evtBytes []byte
 		var err error
 
 		if s.ospreyCompat {
···
 				SendTime: time.Now().Format(time.RFC3339),
 			}
 
-			kafkaEvtBytes, err = json.Marshal(&ospreyKafkaEvent)
+			evtBytes, err = json.Marshal(&ospreyKafkaEvent)
 		} else {
-			kafkaEvtBytes, err = json.Marshal(&kafkaEvt)
+			evtBytes, err = json.Marshal(&kafkaEvt)
 		}
 		if err != nil {
 			return fmt.Errorf("failed to marshal kafka event: %w", err)
 		}
 
-		if err := s.produceAsync(ctx, did, kafkaEvtBytes); err != nil {
+		evtsToProduce = append(evtsToProduce, evtBytes)
+	}
+
+	for _, evtBytes := range evtsToProduce {
+		if err := s.produceAsync(ctx, evtKey, evtBytes); err != nil {
 			return err
 		}
 	}
···
 		producedEvents.WithLabelValues(status).Inc()
 	}
 
-	if !s.ospreyCompat {
-		if err := s.producer.ProduceAsync(ctx, key, msg, callback); err != nil {
-			return fmt.Errorf("failed to produce message: %w", err)
-		}
-	} else if s.ospreyCompat {
-		if err := s.producer.ProduceAsync(ctx, key, msg, callback); err != nil {
-			return fmt.Errorf("failed to produce message: %w", err)
-		}
+	if err := s.producer.ProduceAsync(ctx, key, msg, callback); err != nil {
+		return fmt.Errorf("failed to produce message: %w", err)
 	}
 
 	return nil
 }
-
-type AtKafkaOp struct {
-	Action     string         `json:"action"`
-	Collection string         `json:"collection"`
-	Rkey       string         `json:"rkey"`
-	Uri        string         `json:"uri"`
-	Cid        string         `json:"cid"`
-	Path       string         `json:"path"`
-	Record     map[string]any `json:"record"`
-}
-
-type AtKafkaIdentity struct {
-	Seq    int64  `json:"seq"`
-	Handle string `json:"handle"`
-}
-
-type AtKafkaInfo struct {
-	Name    string  `json:"name"`
-	Message *string `json:"message,omitempty"`
-}
-
-type AtKafkaAccount struct {
-	Active bool    `json:"active"`
-	Seq    int64   `json:"seq"`
-	Status *string `json:"status,omitempty"`
-}
-
-type AtKafkaEvent struct {
-	Did       string         `json:"did"`
-	Timestamp string         `json:"timestamp"`
-	Metadata  *EventMetadata `json:"eventMetadata"`
-
-	Operation *AtKafkaOp       `json:"operation,omitempty"`
-	Account   *AtKafkaAccount  `json:"account,omitempty"`
-	Identity  *AtKafkaIdentity `json:"identity,omitempty"`
-	Info      *AtKafkaInfo     `json:"info,omitempty"`
-}
-
-// Intentionally using snake case since that is what Osprey expects
-type OspreyEventData struct {
-	ActionName string            `json:"action_name"`
-	ActionId   int64             `json:"action_id"`
-	Data       AtKafkaEvent      `json:"data"`
-	Timestamp  string            `json:"timestamp"`
-	SecretData map[string]string `json:"secret_data"`
-	Encoding   string            `json:"encoding"`
-}
-
-type OspreyAtKafkaEvent struct {
-	Data     OspreyEventData `json:"data"`
-	SendTime string          `json:"send_time"`
-}

atkafka/plc.go

···
 	"errors"
 	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"time"
 
···
 
 type PlcClient struct {
 	client     *http.Client
-	dir        *identity.BaseDirectory
-	plcHost    string
-	docCache   *lru.LRU[string, *identity.DIDDocument]
+	dir        *identity.CacheDirectory
 	auditCache *lru.LRU[string, *DidAuditEntry]
+	plcHost    string
 }
 
 type PlcClientArgs struct {
···
 	client := robusthttp.NewClient(robusthttp.WithMaxRetries(2))
 	client.Timeout = 3 * time.Second
 
-	baseDir := identity.BaseDirectory{
-		PLCURL:     args.PlcHost,
-		PLCLimiter: rate.NewLimiter(rate.Limit(200), 100),
-		HTTPClient: *client,
-		Resolver: net.Resolver{
-			PreferGo: true,
-			Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
-				dialer := net.Dialer{Timeout: time.Second * 5}
-				nameserver := address
-				return dialer.DialContext(ctx, network, nameserver)
-			},
+	baseDirectory := identity.BaseDirectory{
+		PLCURL: "https://plc.directory",
+		HTTPClient: http.Client{
+			Timeout: time.Second * 5,
 		},
-		TryAuthoritativeDNS: true,
-		// primary Bluesky PDS instance only supports HTTP resolution method
-		SkipDNSDomainSuffixes: []string{".bsky.social"},
+		PLCLimiter:            rate.NewLimiter(rate.Limit(200), 100),
+		TryAuthoritativeDNS:   true,
+		SkipDNSDomainSuffixes: []string{".bsky.social", ".staging.bsky.dev"},
 	}
-
-	docCache := lru.NewLRU(100_000, func(_ string, _ *identity.DIDDocument) {
-		cacheSize.WithLabelValues("did_doc").Dec()
-	}, 5*time.Minute)
+	directory := identity.NewCacheDirectory(&baseDirectory, 100_000, time.Hour*48, time.Minute*15, time.Minute*15)
 
 	auditCache := lru.NewLRU(100_000, func(_ string, _ *DidAuditEntry) {
 		cacheSize.WithLabelValues("audit_log").Dec()
···
 
 	return &PlcClient{
 		client:     client,
-		dir:        &baseDir,
+		dir:        &directory,
+		auditCache: auditCache,
 		plcHost:    args.PlcHost,
-		docCache:   docCache,
-		auditCache: auditCache,
 	}
 }
 
···
 
 type DidAuditLog []DidAuditEntry
 
-func (c *PlcClient) GetDIDDoc(ctx context.Context, did string) (*identity.DIDDocument, error) {
+func (c *PlcClient) GetIdentity(ctx context.Context, did string) (*identity.Identity, error) {
 	status := "error"
-	cached := false
 
 	defer func() {
-		plcRequests.WithLabelValues("did_doc", status, fmt.Sprintf("%t", cached)).Inc()
+		plcRequests.WithLabelValues("did_doc", status, "unknown").Inc()
 	}()
 
-	if val, ok := c.docCache.Get(did); ok {
-		status = "ok"
-		cached = true
-		return val, nil
-	}
-
-	didDoc, err := c.dir.ResolveDID(ctx, syntax.DID(did))
+	identity, err := c.dir.LookupDID(ctx, syntax.DID(did))
 	if err != nil {
 		return nil, fmt.Errorf("failed to lookup DID: %w", err)
 	}
 
-	if didDoc == nil {
-		return nil, fmt.Errorf("DID Document not found")
-	}
-
-	if c.docCache != nil {
-		c.docCache.Add(did, didDoc)
-	}
-
 	cacheSize.WithLabelValues("did_doc").Inc()
 	status = "ok"
 
-	return didDoc, nil
+	return identity, nil
 }
 
 var ErrAuditLogNotFound = errors.New("audit log not found for DID")

atkafka/types.go

···
+package atkafka
+
+import "github.com/bluesky-social/indigo/atproto/identity"
+
+type AtKafkaOp struct {
+	Action     string         `json:"action"`
+	Collection string         `json:"collection"`
+	Rkey       string         `json:"rkey"`
+	Uri        string         `json:"uri"`
+	Cid        string         `json:"cid"`
+	Path       string         `json:"path"`
+	Record     map[string]any `json:"record"`
+}
+
+type AtKafkaIdentity struct {
+	Seq    int64  `json:"seq"`
+	Handle string `json:"handle"`
+}
+
+type AtKafkaInfo struct {
+	Name    string  `json:"name"`
+	Message *string `json:"message,omitempty"`
+}
+
+type AtKafkaAccount struct {
+	Active bool    `json:"active"`
+	Seq    int64   `json:"seq"`
+	Status *string `json:"status,omitempty"`
+}
+
+type AtKafkaEvent struct {
+	Did       string         `json:"did"`
+	Timestamp string         `json:"timestamp"`
+	Metadata  *EventMetadata `json:"eventMetadata"`
+
+	Operation *AtKafkaOp       `json:"operation,omitempty"`
+	Account   *AtKafkaAccount  `json:"account,omitempty"`
+	Identity  *AtKafkaIdentity `json:"identity,omitempty"`
+	Info      *AtKafkaInfo     `json:"info,omitempty"`
+}
+
+// Intentionally using snake case since that is what Osprey expects
+type OspreyEventData struct {
+	ActionName string            `json:"action_name"`
+	ActionId   int64             `json:"action_id"`
+	Data       AtKafkaEvent      `json:"data"`
+	Timestamp  string            `json:"timestamp"`
+	SecretData map[string]string `json:"secret_data"`
+	Encoding   string            `json:"encoding"`
+}
+
+type OspreyAtKafkaEvent struct {
+	Data     OspreyEventData `json:"data"`
+	SendTime string          `json:"send_time"`
+}
+
+type EventMetadata struct {
+	DidDocument  identity.DIDDocument `json:"didDocument,omitempty"`
+	PdsHost      string               `json:"pdsHost,omitempty"`
+	Handle       string               `json:"handle,omitempty"`
+	DidCreatedAt string               `json:"didCreatedAt,omitempty"`
+	AccountAge   int64                `json:"accountAge"`
+}

cmd/atkafka/main.go

···
 			telemetry.CLIFlagMetricsListenAddress,
 			&cli.StringFlag{
 				Name:    "relay-host",
+				Usage:   "Websocket host to subscribe to for events",
 				Value:   "wss://bsky.network",
 				EnvVars: []string{"ATKAFKA_RELAY_HOST"},
 			},
 			&cli.StringSliceFlag{
 				Name:     "bootstrap-servers",
+				Usage:    "List of Kafka bootstrap servers",
 				EnvVars:  []string{"ATKAFKA_BOOTSTRAP_SERVERS"},
 				Required: true,
 			},
 			&cli.StringFlag{
 				Name:     "output-topic",
+				Usage:    "The Kafka topic to produce events to",
 				EnvVars:  []string{"ATKAFKA_OUTPUT_TOPIC"},
 				Required: true,
 			},
 			&cli.BoolFlag{
 				Name:    "osprey-compatible",
+				Usage:   "Whether or not events should be formulated in an Osprey-compatible format",
 				EnvVars: []string{"ATKAFKA_OSPREY_COMPATIBLE"},
 				Value:   false,
 			},
 			&cli.StringFlag{
 				Name:    "plc-host",
+				Usage:   "The host of the PLC directory you want to use for event metadata",
 				EnvVars: []string{"ATKAFKA_PLC_HOST"},
 			},
+			&cli.StringSliceFlag{
+				Name:    "watched-services",
+				Usage:   "A list of ATProto services inside a user's DID document that you want to watch. Wildcards like *.bsky.network are allowed.",
+				EnvVars: []string{"ATKAFKA_WATCHED_SERVICES"},
+			},
+			&cli.StringSliceFlag{
+				Name:    "ignored-services",
+				Usage:   "A list of ATProto services inside a user's DID document that you want to ignore. Wildcards like *.bsky.network are allowed.",
+				EnvVars: []string{"ATKAFKA_IGNORED_SERVICES"},
+			},
+			&cli.StringSliceFlag{
+				Name:    "watched-collections",
+				Usage:   "A list of collections that you want to watch. Wildcards like *.bsky.app are allowed.",
+				EnvVars: []string{"ATKAFKA_WATCHED_COLLECTIONS"},
+			},
+			&cli.StringSliceFlag{
+				Name:    "ignored-collections",
+				Usage:   "A list of collections that you want to ignore. Wildcards like *.bsky.app are allowed.",
+				EnvVars: []string{"ATKAFKA_IGNORED_COLLECTIONS"},
+			},
 		},
 		Action: func(cmd *cli.Context) error {
 			ctx := context.Background()
···
 			telemetry.StartMetrics(cmd)
 			logger := telemetry.StartLogger(cmd)
 
-			s := atkafka.NewServer(&atkafka.ServerArgs{
-				RelayHost:        cmd.String("relay-host"),
-				BootstrapServers: cmd.StringSlice("bootstrap-servers"),
-				OutputTopic:      cmd.String("output-topic"),
-				OspreyCompat:     cmd.Bool("osprey-compatible"),
-				PlcHost:          cmd.String("plc-host"),
-				Logger:           logger,
+			s, err := atkafka.NewServer(&atkafka.ServerArgs{
+				RelayHost:          cmd.String("relay-host"),
+				BootstrapServers:   cmd.StringSlice("bootstrap-servers"),
+				OutputTopic:        cmd.String("output-topic"),
+				OspreyCompat:       cmd.Bool("osprey-compatible"),
+				PlcHost:            cmd.String("plc-host"),
+				WatchedServices:    cmd.StringSlice("watched-services"),
+				IgnoredServices:    cmd.StringSlice("ignored-services"),
+				WatchedCollections: cmd.StringSlice("watched-collections"),
+				IgnoredCollections: cmd.StringSlice("ignored-collections"),
+				Logger:             logger,
 			})
+			if err != nil {
+				return fmt.Errorf("failed to create new server: %w", err)
+			}
 
 			if err := s.Run(ctx); err != nil {
 				return fmt.Errorf("error running server: %w", err)

docker-compose.yml

···
       ATKAFKA_OUTPUT_TOPIC: "atproto-events"
       ATKAFKA_OSPREY_COMPATIBLE: "false"
       ATKAFKA_PLC_HOST: "https://plc.directory"
+
+      ATKAFKA_WATCHED_SERVICES: "*.bsky.network"
+
+      ATKAFKA_IGNORED_COLLECTIONS: "app.bsky.*"
     restart: unless-stopped
 
 volumes: