diff --git a/cmd/tap/crawler.go b/cmd/tap/crawler.go --- a/cmd/tap/crawler.go +++ b/cmd/tap/crawler.go @@ -19,17 +19,27 @@ 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) 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 @@ 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 @@ 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) 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 @@ 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 @@ 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) 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 --- a/cmd/tap/event_manager.go +++ b/cmd/tap/event_manager.go @@ -28,6 +28,16 @@ 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 --- a/cmd/tap/firehose.go +++ b/cmd/tap/firehose.go @@ -40,6 +40,21 @@ 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) 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 @@ 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 --- a/cmd/tap/main.go +++ b/cmd/tap/main.go @@ -224,7 +224,7 @@ } if !config.OutboxOnly { - go tap.Crawler.Run(ctx) + go tap.crawler.Run(ctx) } svcErr := make(chan error, 1) @@ -232,7 +232,7 @@ 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 @@ 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 @@ 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 --- a/cmd/tap/outbox.go +++ b/cmd/tap/outbox.go @@ -3,6 +3,7 @@ import ( "context" "log/slog" + "net/http" "sync" "time" @@ -41,6 +42,27 @@ outgoing chan *OutboxEvt 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. diff --git a/cmd/tap/repo_manager.go b/cmd/tap/repo_manager.go --- a/cmd/tap/repo_manager.go +++ b/cmd/tap/repo_manager.go @@ -13,8 +13,17 @@ 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) { @@ -39,11 +48,11 @@ 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 --- a/cmd/tap/resyncer.go +++ b/cmd/tap/resyncer.go @@ -38,6 +38,19 @@ 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 @@ 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 --- a/cmd/tap/server.go +++ b/cmd/tap/server.go @@ -27,6 +27,18 @@ 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 --- a/cmd/tap/tap.go +++ b/cmd/tap/tap.go @@ -4,13 +4,11 @@ "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 @@ 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 @@ 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 @@ // 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 {