automod: significantly simplify firehose consumption, for now

automod/countstore.go

···
 
 type CountStore interface {
 	GetCount(ctx context.Context, key, period string) (int, error)
-	Increment(ctx context.Context, key string) (int, error)
+	Increment(ctx context.Context, key string) error
 }
 
 // TODO: this implementation isn't race-safe (yet)!
···
 	}
 }
 
-func (s *MemCountStore) GetCount(ctx context.Context, key, period string) (int, error) {
+func (s MemCountStore) GetCount(ctx context.Context, key, period string) (int, error) {
 	v, ok := s.Counts[PeriodKey(key, period)]
 	if !ok {
 		return 0, nil
···
 	return v, nil
 }
 
-func (s *MemCountStore) Increment(ctx context.Context, key string) error {
+func (s MemCountStore) Increment(ctx context.Context, key string) error {
 	for _, p := range []string{PeriodTotal, PeriodDay, PeriodHour} {
 		k := PeriodKey(key, p)
 		v, ok := s.Counts[k]

automod/engine.go

···
 package automod
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"log/slog"
 	"strings"
 	"sync"
 
-	comatproto "github.com/bluesky-social/indigo/api/atproto"
+	appbsky "github.com/bluesky-social/indigo/api/bsky"
 	"github.com/bluesky-social/indigo/atproto/identity"
 	"github.com/bluesky-social/indigo/atproto/syntax"
-	lexutil "github.com/bluesky-social/indigo/lex/util"
-	"github.com/bluesky-social/indigo/repo"
-	"github.com/bluesky-social/indigo/repomgr"
 	"github.com/bluesky-social/indigo/xrpc"
 )
 
-// runtime for executing rules, managing state, and recording moderation actions
+// runtime for executing rules, managing state, and recording moderation actions.
+//
+// TODO: careful when initializing: several fields should not be null or zero, even though they are pointer type.
 type Engine struct {
+	Logger  *slog.Logger
+	Directory identity.Directory
 	// current rule sets. will eventually be possible to swap these out at runtime
 	RulesMap  sync.Map
-	Directory identity.Directory
 	// used to persist moderation actions in mod service (optional)
 	AdminClient *xrpc.Client
 	CountStore  CountStore
 }
 
-func (e *Engine) ProcessIdentityEvent(t string, did syntax.DID) error {
-	ctx := context.Background()
-
+func (e *Engine) ProcessIdentityEvent(ctx context.Context, t string, did syntax.DID) error {
 	// similar to an HTTP server, we want to recover any panics from rule execution
 	defer func() {
 		if r := recover(); r != nil {
-			slog.Error("automod event execution exception", "err", r)
-			// TODO: mark repo as dirty?
-			// TODO: circuit-break on repeated panics?
+			e.Logger.Error("automod event execution exception", "err", r)
 		}
 	}()
 
···
 			Account: AccountMeta{Identity: ident},
 		},
 	}
-	e.CallIdentityRules(&evt)
-
-	_ = ctx
+	// TODO: call rules
+	_ = evt
 	return nil
 }
 
-// this method takes a full firehose commit event. it must not be a tooBig
-func (e *Engine) ProcessCommit(ctx context.Context, commit *comatproto.SyncSubscribeRepos_Commit) error {
-
+func (e *Engine) ProcessRecord(ctx context.Context, did syntax.DID, path string, rec any) error {
 	// similar to an HTTP server, we want to recover any panics from rule execution
-	/*
 	defer func() {
 		if r := recover(); r != nil {
-			slog.Error("automod event execution exception", "err", r)
-			// TODO: mark repo as dirty?
-			// TODO: circuit-break on repeated panics?
+			e.Logger.Error("automod event execution exception", "err", r)
 		}
 	}()
-	*/
-
-	r, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(commit.Blocks))
-	if err != nil {
-		// TODO: handle this case (instead of return nil)
-		slog.Error("reading repo from car", "size_bytes", len(commit.Blocks), "err", err)
-		return nil
-	}
-
-	did, err := syntax.ParseDID(commit.Repo)
-	if err != nil {
-		return fmt.Errorf("bad DID syntax in event: %w", err)
-	}
 
 	ident, err := e.Directory.LookupDID(ctx, did)
 	if err != nil {
···
 	if ident == nil {
 		return fmt.Errorf("identity not found for did: %s", did.String())
 	}
+	collection := strings.SplitN(path, "/", 2)[0]
 
-	for _, op := range commit.Ops {
-		ek := repomgr.EventKind(op.Action)
-		logOp := slog.With("op_path", op.Path, "op_cid", op.Cid)
-		switch ek {
-		case repomgr.EvtKindCreateRecord:
-			rc, rec, err := r.GetRecord(ctx, op.Path)
-			if err != nil {
-				// TODO: handle this case (instead of return nil)
-				logOp.Error("fetching record from event CAR slice", "err", err)
-				return nil
-			}
-			if lexutil.LexLink(rc) != *op.Cid {
-				// TODO: handle this case (instead of return nil)
-				logOp.Error("mismatch in record and op cid", "record_cid", rc)
-				return nil
-			}
-
-			if strings.HasPrefix(op.Path, "app.bsky.feed.post/") {
-				// TODO: handle as a PostEvent specially
-			} else {
-				// XXX: pass record in to event
-				_ = rec
-				evt := RecordEvent{
-					Event{
-						Engine:  e,
-						Account: AccountMeta{Identity: ident},
-					},
-					[]string{},
-					false,
-					[]ModReport{},
-					[]string{},
-				}
-				e.CallRecordRules(&evt)
-				// TODO persist
-			}
-		case repomgr.EvtKindUpdateRecord:
-			slog.Info("ignoring record update", "did", commit.Repo, "seq", commit.Seq, "path", op.Path)
-			return nil
-		case repomgr.EvtKindDeleteRecord:
-			slog.Info("ignoring record deletion", "did", commit.Repo, "seq", commit.Seq, "path", op.Path)
-			return nil
+	switch collection {
+	case "app.bsky.feed.post":
+		post, ok := rec.(*appbsky.FeedPost)
+		if !ok {
+			return fmt.Errorf("mismatch between collection (%s) and type", collection)
 		}
+		evt := e.NewPostEvent(ident, path, post)
+		e.Logger.Info("processing post", "did", ident.DID, "path", path)
+		_ = evt
+		// TODO: call rules
+	default:
+		evt := e.NewRecordEvent(ident, path, rec)
+		e.Logger.Info("processing record", "did", ident.DID, "path", path)
+		_ = evt
+		// TODO: call rules
 	}
 
-	_ = ctx
 	return nil
 }
 
-func (e *Engine) CallIdentityRules(evt *IdentityEvent) error {
-	slog.Info("calling rules on identity event")
-	return nil
+func (e *Engine) NewPostEvent(ident *identity.Identity, path string, post *appbsky.FeedPost) PostEvent {
+	return PostEvent{
+		RecordEvent {
+			Event{
+				Engine:  e,
+				Account: AccountMeta{Identity: ident},
+			},
+			[]string{},
+			false,
+			[]ModReport{},
+			[]string{},
+		},
+	}
 }
 
-func (e *Engine) CallRecordRules(evt *RecordEvent) error {
-	slog.Info("calling rules on record event")
-	return nil
-}
-
-func (e *Engine) PersistModActions() error {
-	// XXX
-	return nil
+func (e *Engine) NewRecordEvent(ident *identity.Identity, path string, rec any) RecordEvent {
+	return RecordEvent{
+		Event{
+			Engine:  e,
+			Account: AccountMeta{Identity: ident},
+		},
+		[]string{},
+		false,
+		[]ModReport{},
+		[]string{},
+	}
 }
 
 func (e *Engine) GetCount(key, period string) (int, error) {
 	return e.CountStore.GetCount(context.TODO(), key, period)
 }
 
-func (e *Engine) InSet(name, val string) (bool, error) {
+func (e *Engine) InSet(setName, val string) (bool, error) {
 	// XXX: implement
 	return false, nil
 }

cmd/hepa/consumer.go

···
+package main
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"net/http"
+	"net/url"
+
+	comatproto "github.com/bluesky-social/indigo/api/atproto"
+	"github.com/bluesky-social/indigo/atproto/syntax"
+	"github.com/bluesky-social/indigo/events/schedulers/autoscaling"
+	lexutil "github.com/bluesky-social/indigo/lex/util"
+
+	"github.com/bluesky-social/indigo/events"
+	"github.com/bluesky-social/indigo/repo"
+	"github.com/bluesky-social/indigo/repomgr"
+	"github.com/carlmjohnson/versioninfo"
+	"github.com/gorilla/websocket"
+)
+
+func (s *Server) RunConsumer(ctx context.Context) error {
+
+	// TODO: persist cursor in a database or local disk
+	cur := 0
+
+	dialer := websocket.DefaultDialer
+	u, err := url.Parse(s.bgshost)
+	if err != nil {
+		return fmt.Errorf("invalid bgshost URI: %w", err)
+	}
+	u.Path = "xrpc/com.atproto.sync.subscribeRepos"
+	if cur != 0 {
+		u.RawQuery = fmt.Sprintf("cursor=%d", cur)
+	}
+	s.logger.Info("subscribing to repo event stream", "upstream", s.bgshost, "cursor", cur)
+	con, _, err := dialer.Dial(u.String(), http.Header{
+		"User-Agent": []string{fmt.Sprintf("hepa/%s", versioninfo.Short())},
+	})
+	if err != nil {
+		return fmt.Errorf("subscribing to firehose failed (dialing): %w", err)
+	}
+
+	rsc := &events.RepoStreamCallbacks{
+		RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {
+			return s.HandleRepoCommit(ctx, evt)
+		},
+		RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error {
+			did, err := syntax.ParseDID(evt.Did)
+			if err != nil {
+				s.logger.Error("bad DID in RepoHandle event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
+				return nil
+			}
+			if err := s.engine.ProcessIdentityEvent(ctx, "handle", did); err != nil {
+				s.logger.Error("processing handle update failed", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
+			}
+			return nil
+		},
+		// TODO: other event callbacks as needed
+	}
+
+	return events.HandleRepoStream(
+		ctx, con, autoscaling.NewScheduler(
+			autoscaling.DefaultAutoscaleSettings(),
+			s.bgshost,
+			rsc.EventHandler,
+		),
+	)
+}
+
+// NOTE: for now, this function basically never errors, just logs and returns nil. Should think through error processing better.
+func (s *Server) HandleRepoCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error {
+
+	logger := s.logger.With("event", "commit", "did", evt.Repo, "rev", evt.Rev, "seq", evt.Seq)
+	// XXX: debug, not info
+	logger.Info("received commit event")
+
+	if evt.TooBig {
+		logger.Warn("skipping tooBig events for now")
+		return nil
+	}
+
+	did, err := syntax.ParseDID(evt.Repo)
+	if err != nil {
+		logger.Error("bad DID syntax in event", "err", err)
+		return nil
+	}
+
+	rr, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.Blocks))
+	if err != nil {
+		logger.Error("failed to read repo from car", "err", err)
+		return nil
+	}
+
+	for _, op := range evt.Ops {
+		logger = logger.With("eventKind", op.Action, "path", op.Path)
+
+		ek := repomgr.EventKind(op.Action)
+		switch ek {
+		case repomgr.EvtKindCreateRecord:
+			// read the record from blocks, and verify CID
+			rc, rec, err := rr.GetRecord(ctx, op.Path)
+			if err != nil {
+				logger.Error("reading record from event blocks (CAR)", "err", err)
+				break
+			}
+			if op.Cid == nil || lexutil.LexLink(rc) != *op.Cid {
+				logger.Error("mismatch between commit op CID and record block", "recordCID", rc, "opCID", op.Cid)
+				break
+			}
+
+			err = s.engine.ProcessRecord(ctx, did, op.Path, rec)
+			if err != nil {
+				logger.Error("engine failed to process record", "err", err)
+				continue
+			}
+		default:
+			// TODO: other event types: update, delete
+		}
+	}
+
+	return nil
+}

cmd/hepa/firehose.go

···
-package main
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"net/http"
-	"net/url"
-	"strings"
-	"time"
-
-	comatproto "github.com/bluesky-social/indigo/api/atproto"
-	//bsky "github.com/bluesky-social/indigo/api/bsky"
-	"github.com/bluesky-social/indigo/atproto/syntax"
-	"github.com/bluesky-social/indigo/events"
-	"github.com/bluesky-social/indigo/events/schedulers/autoscaling"
-	"github.com/bluesky-social/indigo/repo"
-
-	"github.com/carlmjohnson/versioninfo"
-	"github.com/gorilla/websocket"
-	"github.com/ipfs/go-cid"
-	typegen "github.com/whyrusleeping/cbor-gen"
-)
-
-func (s *Server) getLastCursor() (int64, error) {
-	var lastSeq LastSeq
-	if err := s.db.Find(&lastSeq).Error; err != nil {
-		return 0, err
-	}
-
-	if lastSeq.ID == 0 {
-		return 0, s.db.Create(&lastSeq).Error
-	}
-
-	return lastSeq.Seq, nil
-}
-
-func (s *Server) updateLastCursor(curs int64) error {
-	return s.db.Model(LastSeq{}).Where("id = 1").Update("seq", curs).Error
-}
-
-func (s *Server) Run(ctx context.Context) error {
-	cur, err := s.getLastCursor()
-	if err != nil {
-		return fmt.Errorf("get last cursor: %w", err)
-	}
-
-	err = s.bfs.LoadJobs(ctx)
-	if err != nil {
-		return fmt.Errorf("loading backfill jobs: %w", err)
-	}
-	go s.bf.Start()
-	go s.discoverRepos()
-
-	d := websocket.DefaultDialer
-	u, err := url.Parse(s.bgshost)
-	if err != nil {
-		return fmt.Errorf("invalid bgshost URI: %w", err)
-	}
-	u.Path = "xrpc/com.atproto.sync.subscribeRepos"
-	if cur != 0 {
-		u.RawQuery = fmt.Sprintf("cursor=%d", cur)
-	}
-	con, _, err := d.Dial(u.String(), http.Header{
-		"User-Agent": []string{fmt.Sprintf("palomar/%s", versioninfo.Short())},
-	})
-	if err != nil {
-		return fmt.Errorf("events dial failed: %w", err)
-	}
-
-	rsc := &events.RepoStreamCallbacks{
-		RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {
-			ctx := context.Background()
-			ctx, span := tracer.Start(ctx, "RepoCommit")
-			defer span.End()
-
-			defer func() {
-				if evt.Seq%50 == 0 {
-					if err := s.updateLastCursor(evt.Seq); err != nil {
-						s.logger.Error("failed to persist cursor", "err", err)
-					}
-				}
-			}()
-			logEvt := s.logger.With("repo", evt.Repo, "rev", evt.Rev, "seq", evt.Seq)
-			if evt.TooBig && evt.Prev != nil {
-				// TODO: handle this case (instead of return nil)
-				logEvt.Error("skipping non-genesis tooBig events for now")
-				return nil
-			}
-
-			if evt.TooBig {
-				if err := s.processTooBigCommit(ctx, evt); err != nil {
-					// TODO: handle this case (instead of return nil)
-					logEvt.Error("failed to process tooBig event", "err", err)
-					return nil
-				}
-
-				return nil
-			}
-
-			if !s.skipBackfill {
-				// Check if we've backfilled this repo, if not, we should enqueue it
-				job, err := s.bfs.GetJob(ctx, evt.Repo)
-				if job == nil && err == nil {
-					logEvt.Info("enqueueing backfill job for new repo")
-					if err := s.bfs.EnqueueJob(evt.Repo); err != nil {
-						logEvt.Warn("failed to enqueue backfill job", "err", err)
-					}
-				}
-			}
-
-			if err = s.engine.ProcessCommit(ctx, evt); err != nil {
-				// TODO: handle this, instead of return nul
-				logEvt.Error("failed to process commit", "err", err)
-				return nil
-			}
-
-			return nil
-
-		},
-		RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error {
-			ctx := context.Background()
-			ctx, span := tracer.Start(ctx, "RepoHandle")
-			defer span.End()
-
-			did, err := syntax.ParseDID(evt.Did)
-			if err != nil {
-				s.logger.Error("bad DID in RepoHandle event", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
-				return nil
-			}
-			if err := s.engine.ProcessIdentityEvent("handle", did); err != nil {
-				s.logger.Error("processing handle update failed", "did", evt.Did, "handle", evt.Handle, "seq", evt.Seq, "err", err)
-			}
-			return nil
-		},
-	}
-
-	return events.HandleRepoStream(
-		ctx, con, autoscaling.NewScheduler(
-			autoscaling.DefaultAutoscaleSettings(),
-			s.bgshost,
-			rsc.EventHandler,
-		),
-	)
-}
-
-func (s *Server) discoverRepos() {
-	if s.skipBackfill {
-		s.logger.Info("skipping repo discovery")
-		return
-	}
-
-	ctx := context.Background()
-	log := s.logger.With("func", "discoverRepos")
-	log.Info("starting repo discovery")
-
-	cursor := ""
-	limit := int64(500)
-
-	totalEnqueued := 0
-	totalSkipped := 0
-	totalErrored := 0
-
-	for {
-		resp, err := comatproto.SyncListRepos(ctx, s.bgsxrpc, cursor, limit)
-		if err != nil {
-			log.Error("failed to list repos", "err", err)
-			time.Sleep(5 * time.Second)
-			continue
-		}
-		log.Info("got repo page", "count", len(resp.Repos), "cursor", resp.Cursor)
-		enqueued := 0
-		skipped := 0
-		errored := 0
-		for _, repo := range resp.Repos {
-			job, err := s.bfs.GetJob(ctx, repo.Did)
-			if job == nil && err == nil {
-				log.Info("enqueuing backfill job for new repo", "did", repo.Did)
-				if err := s.bfs.EnqueueJob(repo.Did); err != nil {
-					log.Warn("failed to enqueue backfill job", "err", err)
-					errored++
-					continue
-				}
-				enqueued++
-			} else if err != nil {
-				log.Warn("failed to get backfill job", "did", repo.Did, "err", err)
-				errored++
-			} else {
-				skipped++
-			}
-		}
-		log.Info("enqueued repos", "enqueued", enqueued, "skipped", skipped, "errored", errored)
-		totalEnqueued += enqueued
-		totalSkipped += skipped
-		totalErrored += errored
-		if resp.Cursor != nil && *resp.Cursor != "" {
-			cursor = *resp.Cursor
-		} else {
-			break
-		}
-	}
-
-	log.Info("finished repo discovery", "totalEnqueued", totalEnqueued, "totalSkipped", totalSkipped, "totalErrored", totalErrored)
-}
-
-func (s *Server) handleCreateOrUpdate(ctx context.Context, rawDID string, path string, recP *typegen.CBORMarshaler, rcid *cid.Cid) error {
-	// Since this gets called in a backfill job, we need to check if the path is a post or profile
-	if !strings.Contains(path, "app.bsky.feed.post") && !strings.Contains(path, "app.bsky.actor.profile") {
-		return nil
-	}
-
-	did, err := syntax.ParseDID(rawDID)
-	if err != nil {
-		return fmt.Errorf("bad DID syntax in event: %w", err)
-	}
-
-	ident, err := s.dir.LookupDID(ctx, did)
-	if err != nil {
-		return fmt.Errorf("resolving identity: %w", err)
-	}
-	if ident == nil {
-		return fmt.Errorf("identity not found for did: %s", did.String())
-	}
-	rec := *recP
-
-	_ = rec
-	/* XXX:
-	switch rec := rec.(type) {
-	case *bsky.FeedPost:
-		// XXX: if err := s.indexPost(ctx, ident, rec, path, *rcid); err != nil {
-		_ = rec
-		if err := s.engine.ProcessCommit(ctx, evt); err != nil {
-			postsFailed.Inc()
-			return fmt.Errorf("processing post for %s: %w", did.String(), err)
-		}
-		postsIndexed.Inc()
-	case *bsky.ActorProfile:
-		// XXX: if err := s.indexProfile(ctx, ident, rec, path, *rcid); err != nil {
-		if err := s.engine.ProcessCommit(ctx, evt); err != nil {
-			profilesFailed.Inc()
-			return fmt.Errorf("processing profile for %s: %w", did.String(), err)
-		}
-		profilesIndexed.Inc()
-	default:
-	}
-	*/
-	return nil
-}
-
-func (s *Server) handleDelete(ctx context.Context, rawDID, path string) error {
-	// TODO: just ignoring for now
-	return nil
-}
-
-func (s *Server) processTooBigCommit(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) error {
-	repodata, err := comatproto.SyncGetRepo(ctx, s.bgsxrpc, evt.Repo, "")
-	if err != nil {
-		return err
-	}
-
-	r, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(repodata))
-	if err != nil {
-		return err
-	}
-
-	did, err := syntax.ParseDID(evt.Repo)
-	if err != nil {
-		return fmt.Errorf("bad DID in repo event: %w", err)
-	}
-
-	return r.ForEach(ctx, "", func(k string, v cid.Cid) error {
-		if strings.HasPrefix(k, "app.bsky.feed.post") || strings.HasPrefix(k, "app.bsky.actor.profile") {
-			rcid, rec, err := r.GetRecord(ctx, k)
-			if err != nil {
-				// TODO: handle this case (instead of return nil)
-				s.logger.Error("failed to get record from repo checkout", "path", k, "err", err)
-				return nil
-			}
-
-			// TODO: may want to treat this as a regular event?
-			_ = rcid
-			_ = did
-			_ = rec
-			/* XXX:
-			if err := s.engine.ProcessRecord(ctx, did, m, rec); err != nil {
-				return fmt.Errorf("processing record from tooBig commit: %w", err)
-			}
-			*/
-		}
-		return nil
-	})
-}

cmd/hepa/main.go

···
 import (
 	"context"
 	"fmt"
-	"log"
 	"log/slog"
 	"net/http"
 	"os"
 	"time"
 
 	"github.com/bluesky-social/indigo/atproto/identity"
-	"github.com/bluesky-social/indigo/util/cliutil"
 
 	"github.com/carlmjohnson/versioninfo"
 	_ "github.com/joho/godotenv/autoload"
 	cli "github.com/urfave/cli/v2"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
-	"go.opentelemetry.io/otel/sdk/resource"
-	tracesdk "go.opentelemetry.io/otel/sdk/trace"
-	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
 	"golang.org/x/time/rate"
 )
 
···
 			Usage:   "method, hostname, and port of PLC registry",
 			Value:   "https://plc.directory",
 			EnvVars: []string{"ATP_PLC_HOST"},
-		},
-		&cli.IntFlag{
-			Name:    "max-metadb-connections",
-			EnvVars: []string{"MAX_METADB_CONNECTIONS"},
-			Value:   40,
 		},
 	}
 
···
 
 var runCmd = &cli.Command{
 	Name:  "run",
-	Usage: "run the service",
+	Usage: "run the hepa daemon",
 	Flags: []cli.Flag{
 		&cli.StringFlag{
-			Name:    "database-url",
-			Value:   "sqlite://data/hepa/automod.db",
-			EnvVars: []string{"DATABASE_URL"},
-		},
-		&cli.BoolFlag{
-			Name:    "readonly",
-			EnvVars: []string{"HEPA_READONLY", "READONLY"},
-		},
-		&cli.StringFlag{
-			Name:    "bind",
-			Usage:   "IP or address, and port, to listen on for HTTP APIs",
-			Value:   ":3999",
-			EnvVars: []string{"HEPA_BIND"},
-		},
-		&cli.StringFlag{
 			Name:    "metrics-listen",
 			Usage:   "IP or address, and port, to listen on for metrics APIs",
-			Value:   ":3998",
+			Value:   ":3989",
 			EnvVars: []string{"HEPA_METRICS_LISTEN"},
 		},
 		&cli.IntFlag{
-			Name:    "bgs-sync-rate-limit",
-			Usage:   "max repo sync (checkout) requests per second to upstream (BGS)",
-			Value:   8,
-			EnvVars: []string{"HEPA_BGS_SYNC_RATE_LIMIT"},
-		},
-		&cli.IntFlag{
 			Name:    "plc-rate-limit",
 			Usage:   "max number of requests per second to PLC registry",
 			Value:   100,
···
 		}))
 		slog.SetDefault(logger)
 
-		// 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 != "" {
-			slog.Info("setting up trace exporter", "endpoint", ep)
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
+		configOTEL("hepa")
 
-			exp, err := otlptracehttp.New(ctx)
-			if err != nil {
-				log.Fatal("failed to create trace exporter", "error", err)
-			}
-			defer func() {
-				ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-				defer cancel()
-				if err := exp.Shutdown(ctx); err != nil {
-					slog.Error("failed to shutdown trace exporter", "error", err)
-				}
-			}()
-
-			tp := tracesdk.NewTracerProvider(
-				tracesdk.WithBatcher(exp),
-				tracesdk.WithResource(resource.NewWithAttributes(
-					semconv.SchemaURL,
-					semconv.ServiceNameKey.String("hepa"),
-					attribute.String("env", os.Getenv("ENVIRONMENT")),         // DataDog
-					attribute.String("environment", os.Getenv("ENVIRONMENT")), // Others
-					attribute.Int64("ID", 1),
-				)),
-			)
-			otel.SetTracerProvider(tp)
-		}
-
-		db, err := cliutil.SetupDatabase(cctx.String("database-url"), cctx.Int("max-metadb-connections"))
-		if err != nil {
-			return err
-		}
-
-		// TODO: replace this with "bingo" resolver?
-		base := identity.BaseDirectory{
+		baseDir := identity.BaseDirectory{
 			PLCURL: cctx.String("atp-plc-host"),
 			HTTPClient: http.Client{
 				Timeout: time.Second * 15,
···
 			TryAuthoritativeDNS:   true,
 			SkipDNSDomainSuffixes: []string{".bsky.social"},
 		}
-		dir := identity.NewCacheDirectory(&base, 1_500_000, time.Hour*24, time.Minute*2)
+		dir := identity.NewCacheDirectory(&baseDir, 1_500_000, time.Hour*24, time.Minute*2)
 
 		srv, err := NewServer(
-			db,
 			&dir,
 			Config{
-				BGSHost:          cctx.String("atp-bgs-host"),
-				Logger:           logger,
-				BGSSyncRateLimit: cctx.Int("bgs-sync-rate-limit"),
+				BGSHost: cctx.String("atp-bgs-host"),
+				Logger:  logger,
 			},
 		)
 		if err != nil {
 			return err
 		}
 
+		// prometheus HTTP endpoint: /metrics
 		go func() {
 			if err := srv.RunMetrics(cctx.String("metrics-listen")); err != nil {
 				slog.Error("failed to start metrics endpoint", "error", err)
···
 			}
 		}()
 
-		// TODO: if cctx.Bool("readonly") ...
-
-		if err := srv.Run(ctx); err != nil {
-			return fmt.Errorf("failed to run automod service: %w", err)
+		// the main service loop
+		if err := srv.RunConsumer(ctx); err != nil {
+			return fmt.Errorf("failure consuming and processing firehose: %w", err)
 		}
 		return nil
 	},

cmd/hepa/metrics.go

···
-package main
-
-import (
-	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
-	"go.opentelemetry.io/otel"
-)
-
-var tracer = otel.Tracer("hepa")
-
-var postsReceived = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_posts_received",
-	Help: "Number of posts received",
-})
-
-var postsIndexed = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_posts_indexed",
-	Help: "Number of posts indexed",
-})
-
-var postsFailed = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_posts_failed",
-	Help: "Number of posts that failed indexing",
-})
-
-var postsDeleted = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_posts_deleted",
-	Help: "Number of posts deleted",
-})
-
-var profilesReceived = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_profiles_received",
-	Help: "Number of profiles received",
-})
-
-var profilesIndexed = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_profiles_indexed",
-	Help: "Number of profiles indexed",
-})
-
-var profilesFailed = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_profiles_failed",
-	Help: "Number of profiles that failed indexing",
-})
-
-var profilesDeleted = promauto.NewCounter(prometheus.CounterOpts{
-	Name: "hepa_profiles_deleted",
-	Help: "Number of profiles deleted",
-})
-
-var currentSeq = promauto.NewGauge(prometheus.GaugeOpts{
-	Name: "hepa_current_seq",
-	Help: "Current sequence number",
-})

cmd/hepa/otel.go

···
+package main
+
+import (
+	"os"
+	"log/slog"
+	"log"
+	"context"
+	"time"
+
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
+	"go.opentelemetry.io/otel/sdk/resource"
+	tracesdk "go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+)
+
+var tracer = otel.Tracer("hepa")
+
+// 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
+// TODO: this should be in cliutil or something
+func configOTEL(serviceName string) {
+	if ep := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); ep != "" {
+		slog.Info("setting up trace exporter", "endpoint", ep)
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		exp, err := otlptracehttp.New(ctx)
+		if err != nil {
+			log.Fatal("failed to create trace exporter", "error", err)
+		}
+		defer func() {
+			ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+			defer cancel()
+			if err := exp.Shutdown(ctx); err != nil {
+				slog.Error("failed to shutdown trace exporter", "error", err)
+			}
+		}()
+
+		tp := tracesdk.NewTracerProvider(
+			tracesdk.WithBatcher(exp),
+			tracesdk.WithResource(resource.NewWithAttributes(
+				semconv.SchemaURL,
+				semconv.ServiceNameKey.String(serviceName),
+				attribute.String("env", os.Getenv("ENVIRONMENT")),         // DataDog
+				attribute.String("environment", os.Getenv("ENVIRONMENT")), // Others
+				attribute.Int64("ID", 1),
+			)),
+		)
+		otel.SetTracerProvider(tp)
+	}
+}

cmd/hepa/server.go

···
 
 	"github.com/bluesky-social/indigo/atproto/identity"
 	"github.com/bluesky-social/indigo/automod"
-	"github.com/bluesky-social/indigo/backfill"
-	"github.com/bluesky-social/indigo/xrpc"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
-	gorm "gorm.io/gorm"
 )
 
 type Server struct {
-	db           *gorm.DB
-	bgshost      string
-	bgsxrpc      *xrpc.Client
-	dir          identity.Directory
-	logger       *slog.Logger
-	engine       *automod.Engine
-	skipBackfill bool
-
-	bfs *backfill.Gormstore
-	bf  *backfill.Backfiller
-}
-
-type LastSeq struct {
-	ID  uint `gorm:"primarykey"`
-	Seq int64
+	bgshost string
+	logger  *slog.Logger
+	engine  *automod.Engine
 }
 
 type Config struct {
-	BGSHost             string
-	Logger              *slog.Logger
-	BGSSyncRateLimit    int
-	MaxEventConcurrency int
+	BGSHost string
+	Logger  *slog.Logger
 }
 
-func NewServer(db *gorm.DB, dir identity.Directory, config Config) (*Server, error) {
+func NewServer(dir identity.Directory, config Config) (*Server, error) {
 	logger := config.Logger
 	if logger == nil {
 		logger = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
···
 		}))
 	}
 
-	logger.Info("running database migrations")
-	db.AutoMigrate(&LastSeq{})
-	db.AutoMigrate(&backfill.GormDBJob{})
-
 	bgsws := config.BGSHost
 	if !strings.HasPrefix(bgsws, "ws") {
 		return nil, fmt.Errorf("specified bgs host must include 'ws://' or 'wss://'")
 	}
 
-	bgshttp := strings.Replace(bgsws, "ws", "http", 1)
-	bgsxrpc := &xrpc.Client{
-		Host: bgshttp,
-	}
-
 	engine := automod.Engine{
+		Logger: logger,
 		Directory: dir,
+		CountStore: automod.NewMemCountStore(),
+		// TODO: RulesMap (loaded/config from somewhere)
+		// TODO: AdminClient (XRPC with mod access)
 	}
 
 	s := &Server{
-		db:           db,
-		bgshost:      config.BGSHost, // NOTE: the original URL, not 'bgshttp'
-		bgsxrpc:      bgsxrpc,
-		dir:          dir,
-		logger:       logger,
-		engine:       &engine,
-		skipBackfill: true,
+		bgshost: config.BGSHost,
+		logger:  logger,
+		engine:  &engine,
 	}
-
-	bfstore := backfill.NewGormstore(db)
-	opts := backfill.DefaultBackfillOptions()
-	if config.BGSSyncRateLimit > 0 {
-		opts.SyncRequestsPerSecond = config.BGSSyncRateLimit
-		opts.ParallelBackfills = 2 * config.BGSSyncRateLimit
-	} else {
-		opts.SyncRequestsPerSecond = 8
-	}
-	opts.CheckoutPath = fmt.Sprintf("%s/xrpc/com.atproto.sync.getRepo", bgshttp)
-	if config.MaxEventConcurrency > 0 {
-		opts.ParallelRecordCreates = config.MaxEventConcurrency
-	} else {
-		opts.ParallelRecordCreates = 20
-	}
-	opts.NSIDFilter = "app.bsky."
-	bf := backfill.NewBackfiller(
-		"hepa",
-		bfstore,
-		s.handleCreateOrUpdate,
-		s.handleCreateOrUpdate,
-		s.handleDelete,
-		opts,
-	)
-
-	s.bfs = bfstore
-	s.bf = bf
 
 	return s, nil
 }