Add --skip-dids to Hepa (#1353)

Author: jcalabro

automod/consumer/firehose.go

@@ -35,6 +35,9 @@ type FirehoseConsumer struct {
 	Engine      *automod.Engine
 	Host        string
 
+	// if set, events from these DIDs will be silently skipped
+	SkipDIDs map[string]struct{}
+
 	// TODO: prefilter record collections; or predicate function?
 	// TODO: enable/disable event types; or predicate function?
 
@@ -80,13 +83,19 @@ func (fc *FirehoseConsumer) Run(ctx context.Context) error {
 		},
 		RepoIdentity: func(evt *comatproto.SyncSubscribeRepos_Identity) error {
 			atomic.StoreInt64(&fc.lastSeq, evt.Seq)
+			if _, skip := fc.SkipDIDs[evt.Did]; skip {
+				return nil
+			}
 			if err := fc.Engine.ProcessIdentityEvent(context.Background(), *evt); err != nil {
 				fc.Logger.Error("processing repo identity failed", "did", evt.Did, "seq", evt.Seq, "err", err)
 			}
 			return nil
 		},
 		RepoAccount: func(evt *comatproto.SyncSubscribeRepos_Account) error {
 			atomic.StoreInt64(&fc.lastSeq, evt.Seq)
+			if _, skip := fc.SkipDIDs[evt.Did]; skip {
+				return nil
+			}
 			if err := fc.Engine.ProcessAccountEvent(context.Background(), *evt); err != nil {
 				fc.Logger.Error("processing repo account failed", "did", evt.Did, "seq", evt.Seq, "err", err)
 			}
@@ -130,6 +139,10 @@ func (fc *FirehoseConsumer) HandleRepoCommit(ctx context.Context, evt *comatprot
 		return nil
 	}
 
+	if _, skip := fc.SkipDIDs[evt.Repo]; skip {
+		return nil
+	}
+
 	did, err := syntax.ParseDID(evt.Repo)
 	if err != nil {
 		logger.Error("bad DID syntax in event", "err", err)