From c739bd8d23f0cbfb1cdfd1eddaf2f93343507495 Mon Sep 17 00:00:00 2001 From: Dave Cheney Date: Thu, 29 Jan 2026 14:50:11 +1100 Subject: [PATCH] cmd/tap: move component initalisation into constructor fns This PR moves the declaration of the various tap components from runTap to various constructor functions. Along the way I also took the liberty of unexporting various fields, and plumbed the outboxOnly mode down to main.Tap rather than the whole config object. There is no functional change, just breaking out the initalisation for each component so in tests we can use the same logic as in runTap. --- cmd/tap/crawler.go | 40 ++++++++----- cmd/tap/event_manager.go | 10 ++++ cmd/tap/firehose.go | 19 +++++- cmd/tap/main.go | 8 +-- cmd/tap/outbox.go | 22 +++++++ cmd/tap/repo_manager.go | 15 ++++- cmd/tap/resyncer.go | 15 ++++- cmd/tap/server.go | 12 ++++ cmd/tap/tap.go | 121 +++++++++------------------------------ 9 files changed, 144 insertions(+), 118 deletions(-) diff --git a/cmd/tap/crawler.go b/cmd/tap/crawler.go index 2470bfaf..585ae5ba 100644 --- a/cmd/tap/crawler.go +++ b/cmd/tap/crawler.go @@ -19,17 +19,27 @@ type Crawler struct { db *gorm.DB logger *slog.Logger - FullNetworkMode bool - RelayUrl string - SignalCollection string + fullNetworkMode bool + relayUrl string + signalCollection string +} + +func NewCrawler(logger *slog.Logger, db *gorm.DB, config *TapConfig) *Crawler { + return &Crawler{ + logger: logger.With("component", "crawler"), + db: db, + fullNetworkMode: config.FullNetworkMode, + relayUrl: config.RelayUrl, + signalCollection: config.SignalCollection, + } } func (c *Crawler) Run(ctx context.Context) { for { var err error - if c.SignalCollection != "" { - err = c.EnumerateNetworkByCollection(ctx, c.SignalCollection) - } else if c.FullNetworkMode { + if c.signalCollection != "" { + err = c.EnumerateNetworkByCollection(ctx, c.signalCollection) + } else if c.fullNetworkMode { err = c.EnumerateNetwork(ctx) } var d time.Duration @@ -50,9 +60,9 @@ func (c *Crawler) Run(ctx context.Context) { } func (c *Crawler) GetCursor(ctx context.Context) (string, error) { - if c.SignalCollection != "" { - return c.getCollectionCursor(ctx, c.SignalCollection) - } else if c.FullNetworkMode { + if c.signalCollection != "" { + return c.getCollectionCursor(ctx, c.signalCollection) + } else if c.fullNetworkMode { return c.getListReposCursor(ctx) } return "", nil @@ -68,7 +78,7 @@ func (c *Crawler) EnumerateNetwork(ctx context.Context) error { return err } - client := atclient.NewAPIClient(c.RelayUrl) + client := atclient.NewAPIClient(c.relayUrl) client.Headers.Set("User-Agent", userAgent()) client.Client = &http.Client{ Timeout: 30 * time.Second, @@ -116,7 +126,7 @@ func (c *Crawler) EnumerateNetwork(ctx context.Context) error { cursor = *repoList.Cursor if err := c.db.WithContext(ctx).Save(&models.ListReposCursor{ - Url: c.RelayUrl, + Url: c.relayUrl, Cursor: cursor, }).Error; err != nil { c.logger.Error("failed to save list repos cursor", "error", err) @@ -129,7 +139,7 @@ func (c *Crawler) EnumerateNetwork(ctx context.Context) error { func (c *Crawler) getListReposCursor(ctx context.Context) (string, error) { var dbCursor models.ListReposCursor - err := c.db.WithContext(ctx).Where("url = ?", c.RelayUrl).First(&dbCursor).Error + err := c.db.WithContext(ctx).Where("url = ?", c.relayUrl).First(&dbCursor).Error if err != nil { if err != gorm.ErrRecordNotFound { return "", fmt.Errorf("failed to read list repos cursor: %w", err) @@ -150,7 +160,7 @@ func (c *Crawler) EnumerateNetworkByCollection(ctx context.Context, collection s return err } - client := atclient.NewAPIClient(c.RelayUrl) + client := atclient.NewAPIClient(c.relayUrl) client.Headers.Set("User-Agent", userAgent()) client.Client = &http.Client{ Timeout: 30 * time.Second, @@ -195,7 +205,7 @@ func (c *Crawler) EnumerateNetworkByCollection(ctx context.Context, collection s cursor = *repoList.Cursor if err := c.db.WithContext(ctx).Save(&models.CollectionCursor{ - Url: c.RelayUrl, + Url: c.relayUrl, Collection: collection, Cursor: cursor, }).Error; err != nil { @@ -209,7 +219,7 @@ func (c *Crawler) EnumerateNetworkByCollection(ctx context.Context, collection s func (c *Crawler) getCollectionCursor(ctx context.Context, collection string) (string, error) { var dbCursor models.CollectionCursor - err := c.db.WithContext(ctx).Where("url = ? AND collection = ?", c.RelayUrl, collection).First(&dbCursor).Error + err := c.db.WithContext(ctx).Where("url = ? AND collection = ?", c.relayUrl, collection).First(&dbCursor).Error if err != nil { if err != gorm.ErrRecordNotFound { return "", fmt.Errorf("failed to read collection cursor: %w", err) diff --git a/cmd/tap/event_manager.go b/cmd/tap/event_manager.go index 891fba63..9058145b 100644 --- a/cmd/tap/event_manager.go +++ b/cmd/tap/event_manager.go @@ -28,6 +28,16 @@ type EventManager struct { pendingIDs chan uint } +func NewEventManager(logger *slog.Logger, db *gorm.DB, config *TapConfig) *EventManager { + return &EventManager{ + logger: logger.With("component", "event_manager"), + db: db, + cacheSize: config.EventCacheSize, + cache: make(map[uint]*OutboxEvt), + pendingIDs: make(chan uint, config.EventCacheSize*2), // give us some buffer room in channel since we can overshoot + } +} + type DBCallback = func(tx *gorm.DB) error func (em *EventManager) IsFull() bool { diff --git a/cmd/tap/firehose.go b/cmd/tap/firehose.go index 3d52c108..02f82f6d 100644 --- a/cmd/tap/firehose.go +++ b/cmd/tap/firehose.go @@ -40,6 +40,21 @@ type FirehoseProcessor struct { lastSeq atomic.Int64 } +func NewFirehoseProcessor(logger *slog.Logger, db *gorm.DB, events *EventManager, repos *RepoManager, config *TapConfig) *FirehoseProcessor { + return &FirehoseProcessor{ + logger: logger.With("component", "firehose"), + db: db, + events: events, + repos: repos, + relayUrl: config.RelayUrl, + fullNetworkMode: config.FullNetworkMode, + signalCollection: config.SignalCollection, + collectionFilters: config.CollectionFilters, + parallelism: config.FirehoseParallelism, + cursorSaveInterval: config.FirehoseCursorSaveInterval, + } +} + func (fp *FirehoseProcessor) updateLastSeq(seq int64) { fp.lastSeq.Store(seq) firehoseLastSeq.Set(float64(seq)) @@ -123,7 +138,7 @@ func (fp *FirehoseProcessor) ProcessCommit(ctx context.Context, evt *comatproto. } func (fp *FirehoseProcessor) validateCommitAndFilterOps(ctx context.Context, evt *comatproto.SyncSubscribeRepos_Commit) (*Commit, error) { - if err := repo.VerifyCommitSignature(ctx, fp.repos.IdDir, evt); err != nil { + if err := repo.VerifyCommitSignature(ctx, fp.repos.idDir, evt); err != nil { return nil, err } @@ -216,7 +231,7 @@ func (fp *FirehoseProcessor) ProcessSync(ctx context.Context, evt *comatproto.Sy return nil } - commit, err := repo.VerifySyncMessage(ctx, fp.repos.IdDir, evt) + commit, err := repo.VerifySyncMessage(ctx, fp.repos.idDir, evt) if err != nil { return fmt.Errorf("failed to verify sync message: %w", err) } diff --git a/cmd/tap/main.go b/cmd/tap/main.go index 40a7eace..214279c2 100644 --- a/cmd/tap/main.go +++ b/cmd/tap/main.go @@ -224,7 +224,7 @@ func runTap(ctx context.Context, cmd *cli.Command) error { } if !config.OutboxOnly { - go tap.Crawler.Run(ctx) + go tap.crawler.Run(ctx) } svcErr := make(chan error, 1) @@ -232,7 +232,7 @@ func runTap(ctx context.Context, cmd *cli.Command) error { if !config.OutboxOnly { go func() { logger.Info("starting firehose consumer") - if err := tap.Firehose.Run(ctx); err != nil { + if err := tap.firehose.Run(ctx); err != nil { svcErr <- err } }() @@ -242,7 +242,7 @@ func runTap(ctx context.Context, cmd *cli.Command) error { go func() { logger.Info("starting HTTP server", "addr", cmd.String("bind")) - if err := tap.Server.Start(cmd.String("bind")); err != nil { + if err := tap.server.Start(cmd.String("bind")); err != nil { svcErr <- err } }() @@ -273,7 +273,7 @@ func runTap(ctx context.Context, cmd *cli.Command) error { shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) defer shutdownCancel() - if err := tap.Server.Shutdown(shutdownCtx); err != nil { + if err := tap.server.Shutdown(shutdownCtx); err != nil { logger.Error("error during shutdown", "error", err) return err } diff --git a/cmd/tap/outbox.go b/cmd/tap/outbox.go index d308eee5..b2610806 100644 --- a/cmd/tap/outbox.go +++ b/cmd/tap/outbox.go @@ -3,6 +3,7 @@ package main import ( "context" "log/slog" + "net/http" "sync" "time" @@ -43,6 +44,27 @@ type Outbox struct { ctx context.Context } +func NewOutbox(logger *slog.Logger, events *EventManager, config *TapConfig) *Outbox { + return &Outbox{ + logger: logger.With("component", "outbox"), + mode: parseOutboxMode(config.WebhookURL, config.DisableAcks), + parallelism: config.OutboxParallelism, + retryTimeout: config.RetryTimeout, + webhook: &WebhookClient{ + logger: logger.With("component", "webhook_client"), + webhookURL: config.WebhookURL, + adminPassword: config.AdminPassword, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + }, + events: events, + didWorkers: xsync.NewMap[string, *DIDWorker](), + acks: make(chan uint, config.OutboxParallelism*10000), + outgoing: make(chan *OutboxEvt, config.OutboxParallelism*10000), + } +} + // Run starts the outbox workers for event delivery and cleanup. func (o *Outbox) Run(ctx context.Context) { o.ctx = ctx diff --git a/cmd/tap/repo_manager.go b/cmd/tap/repo_manager.go index 373a65b3..8709ac59 100644 --- a/cmd/tap/repo_manager.go +++ b/cmd/tap/repo_manager.go @@ -13,10 +13,19 @@ import ( type RepoManager struct { logger *slog.Logger db *gorm.DB - IdDir identity.Directory + idDir identity.Directory events *EventManager } +func NewRepoManager(logger *slog.Logger, db *gorm.DB, events *EventManager, idDir identity.Directory) *RepoManager { + return &RepoManager{ + logger: logger.With("component", "repo_manager"), + db: db, + idDir: idDir, + events: events, + } +} + func (rm *RepoManager) GetRepoState(ctx context.Context, did string) (*models.Repo, error) { var r models.Repo if err := rm.db.WithContext(ctx).First(&r, "did = ?", did).Error; err != nil { @@ -39,11 +48,11 @@ func (rm *RepoManager) RefreshIdentity(ctx context.Context, did string) error { ctx, span := tracer.Start(ctx, "RefreshIdentity") defer span.End() - if err := rm.IdDir.Purge(ctx, syntax.DID(did).AtIdentifier()); err != nil { + if err := rm.idDir.Purge(ctx, syntax.DID(did).AtIdentifier()); err != nil { rm.logger.Error("failed to purge identity cache", "did", did, "error", err) } - id, err := rm.IdDir.LookupDID(ctx, syntax.DID(did)) + id, err := rm.idDir.LookupDID(ctx, syntax.DID(did)) if err != nil { return err } diff --git a/cmd/tap/resyncer.go b/cmd/tap/resyncer.go index c21f4a3c..3cc91c3c 100644 --- a/cmd/tap/resyncer.go +++ b/cmd/tap/resyncer.go @@ -38,6 +38,19 @@ type Resyncer struct { pdsBackoffMu sync.RWMutex } +func NewResyncer(logger *slog.Logger, db *gorm.DB, repos *RepoManager, events *EventManager, config *TapConfig) *Resyncer { + return &Resyncer{ + logger: logger.With("component", "resyncer"), + db: db, + events: events, + repos: repos, + repoFetchTimeout: config.RepoFetchTimeout, + collectionFilters: config.CollectionFilters, + parallelism: config.ResyncParallelism, + pdsBackoff: make(map[string]time.Time), + } +} + func (r *Resyncer) run(ctx context.Context) { for i := 0; i < r.parallelism; i++ { go r.runResyncWorker(ctx, i) @@ -146,7 +159,7 @@ func (r *Resyncer) doResync(ctx context.Context, did string) (bool, error) { span.SetAttributes(attribute.String("did", did)) defer span.End() - ident, err := r.repos.IdDir.LookupDID(ctx, syntax.DID(did)) + ident, err := r.repos.idDir.LookupDID(ctx, syntax.DID(did)) if err != nil { return false, fmt.Errorf("failed to resolve DID: %w", err) } diff --git a/cmd/tap/server.go b/cmd/tap/server.go index 361b371e..1bab969c 100644 --- a/cmd/tap/server.go +++ b/cmd/tap/server.go @@ -27,6 +27,18 @@ type TapServer struct { crawler *Crawler } +func NewTapServer(logger *slog.Logger, db *gorm.DB, outbox *Outbox, idDir identity.Directory, firehose *FirehoseProcessor, crawler *Crawler, config *TapConfig) *TapServer { + return &TapServer{ + logger: logger.With("component", "server"), + db: db, + outbox: outbox, + adminPassword: config.AdminPassword, + idDir: idDir, + firehose: firehose, + crawler: crawler, + } +} + func (ts *TapServer) Start(address string) error { ts.echo = echo.New() ts.echo.HideBanner = true diff --git a/cmd/tap/tap.go b/cmd/tap/tap.go index d2c73c12..0c39cf8f 100644 --- a/cmd/tap/tap.go +++ b/cmd/tap/tap.go @@ -4,13 +4,11 @@ import ( "context" "fmt" "log/slog" - "net/http" "strings" "time" "github.com/bluesky-social/indigo/atproto/identity" "github.com/bluesky-social/indigo/cmd/tap/models" - "github.com/puzpuzpuz/xsync/v4" "gorm.io/driver/postgres" "gorm.io/driver/sqlite" "gorm.io/gorm" @@ -21,16 +19,16 @@ type Tap struct { db *gorm.DB logger *slog.Logger - Firehose *FirehoseProcessor - Events *EventManager - Repos *RepoManager - Resyncer *Resyncer - Crawler *Crawler + firehose *FirehoseProcessor + events *EventManager + repos *RepoManager + resyncer *Resyncer + crawler *Crawler - Server *TapServer - Outbox *Outbox + server *TapServer + outbox *Outbox - config TapConfig + outboxOnly bool } type TapConfig struct { @@ -70,98 +68,35 @@ func NewTap(config TapConfig) (*Tap, error) { logger := slog.Default().With("system", "tap") - evtMngr := &EventManager{ - logger: logger.With("component", "event_manager"), - db: db, - cacheSize: config.EventCacheSize, - cache: make(map[uint]*OutboxEvt), - pendingIDs: make(chan uint, config.EventCacheSize*2), // give us some buffer room in channel since we can overshoot - } + evtMngr := NewEventManager(logger, db, &config) - repoMngr := &RepoManager{ - logger: logger.With("component", "server"), - db: db, - IdDir: &cdir, - events: evtMngr, - } + repoMngr := NewRepoManager(logger, db, evtMngr, &cdir) - resyncer := &Resyncer{ - logger: logger.With("component", "resyncer"), - db: db, - events: evtMngr, - repos: repoMngr, - repoFetchTimeout: config.RepoFetchTimeout, - collectionFilters: config.CollectionFilters, - parallelism: config.ResyncParallelism, - pdsBackoff: make(map[string]time.Time), - } + resyncer := NewResyncer(logger, db, repoMngr, evtMngr, &config) - firehose := &FirehoseProcessor{ - logger: logger.With("component", "firehose"), - db: db, - events: evtMngr, - repos: repoMngr, - relayUrl: config.RelayUrl, - fullNetworkMode: config.FullNetworkMode, - signalCollection: config.SignalCollection, - collectionFilters: config.CollectionFilters, - parallelism: config.FirehoseParallelism, - cursorSaveInterval: config.FirehoseCursorSaveInterval, - } + firehose := NewFirehoseProcessor(logger, db, evtMngr, repoMngr, &config) - crawler := &Crawler{ - logger: logger.With("component", "crawler"), - db: db, - FullNetworkMode: config.FullNetworkMode, - RelayUrl: config.RelayUrl, - SignalCollection: config.SignalCollection, - } + crawler := NewCrawler(logger, db, &config) - outbox := &Outbox{ - logger: logger.With("component", "outbox"), - mode: parseOutboxMode(config.WebhookURL, config.DisableAcks), - parallelism: config.OutboxParallelism, - retryTimeout: config.RetryTimeout, - webhook: &WebhookClient{ - logger: logger.With("component", "webhook_client"), - webhookURL: config.WebhookURL, - adminPassword: config.AdminPassword, - httpClient: &http.Client{ - Timeout: 30 * time.Second, - }, - }, - events: evtMngr, - didWorkers: xsync.NewMap[string, *DIDWorker](), - acks: make(chan uint, config.OutboxParallelism*10000), - outgoing: make(chan *OutboxEvt, config.OutboxParallelism*10000), - } + outbox := NewOutbox(logger, evtMngr, &config) - server := &TapServer{ - logger: logger.With("component", "server"), - db: db, - outbox: outbox, - adminPassword: config.AdminPassword, - idDir: repoMngr.IdDir, - firehose: firehose, - crawler: crawler, - } + server := NewTapServer(logger, db, outbox, &cdir, firehose, crawler, &config) t := &Tap{ db: db, logger: slog.Default().With("system", "tap"), - Firehose: firehose, - Events: evtMngr, - Repos: repoMngr, - Resyncer: resyncer, - Crawler: crawler, - Server: server, - Outbox: outbox, - - config: config, + firehose: firehose, + events: evtMngr, + repos: repoMngr, + resyncer: resyncer, + crawler: crawler, + server: server, + outbox: outbox, + outboxOnly: config.OutboxOnly, } - if err := t.Resyncer.resetPartiallyResynced(context.Background()); err != nil { + if err := t.resyncer.resetPartiallyResynced(context.Background()); err != nil { return nil, err } @@ -170,13 +105,13 @@ func NewTap(config TapConfig) (*Tap, error) { // Run starts internal background workers for resync, cursor saving, and outbox delivery. func (t *Tap) Run(ctx context.Context) { - go t.Events.LoadEvents(ctx) + go t.events.LoadEvents(ctx) - if !t.config.OutboxOnly { - go t.Resyncer.run(ctx) + if !t.outboxOnly { + go t.resyncer.run(ctx) } - go t.Outbox.Run(ctx) + go t.outbox.Run(ctx) } func (t *Tap) CloseDb(ctx context.Context) error { -- 2.51.2