diff --git a/spindle/embedtap.go b/spindle/embedtap.go --- a/spindle/embedtap.go +++ b/spindle/embedtap.go @@ -50,12 +50,8 @@ logger *slog.Logger closed atomic.Bool } -func startEmbeddedTap(ctx context.Context, cfg *config.Config, logger *slog.Logger) (*embeddedTap, error) { - if err := assertLoopbackBind(cfg.Server.Tap.Bind); err != nil { - return nil, err - } - - tcfg := tap.Config{ +func newEmbeddedTapConfig(cfg *config.Config) tap.Config { + return tap.Config{ DatabaseURL: "sqlite://" + cfg.Server.Tap.DBPath, DBMaxConns: 32, PLCURL: cfg.Server.PlcUrl, @@ -67,11 +63,18 @@ FirehoseCursorSaveInterval: time.Second, RepoFetchTimeout: 5 * time.Minute, IdentityCacheSize: 50_000, EventCacheSize: 10_000, - SignalCollection: tangled.RepoPullNSID, // HACK: to ingest PRs from any users - CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID}, + CollectionFilters: []string{tangled.RepoNSID, tangled.RepoCollaboratorNSID}, AdminPassword: cfg.Server.Tap.AdminPassword, RetryTimeout: 60 * time.Second, } +} + +func startEmbeddedTap(ctx context.Context, cfg *config.Config, logger *slog.Logger) (*embeddedTap, error) { + if err := assertLoopbackBind(cfg.Server.Tap.Bind); err != nil { + return nil, err + } + + tcfg := newEmbeddedTapConfig(cfg) t, err := tap.New(tcfg) if err != nil { diff --git a/spindle/ingester.go b/spindle/ingester.go --- a/spindle/ingester.go +++ b/spindle/ingester.go @@ -27,9 +27,13 @@ var err error switch e.Commit.Collection { case tangled.SpindleMemberNSID: err = s.ingestMember(ctx, e) - case tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID: + case tangled.RepoNSID, tangled.RepoCollaboratorNSID: if evt, ok := jetstreamToTapEvent(e); ok { err = s.tap.processEvent(ctx, evt) + } + case tangled.RepoPullNSID: + if evt, ok := jetstreamToTapEvent(e); ok { + err = s.processPull(ctx, evt.Record) } } diff --git a/spindle/ingester_test.go b/spindle/ingester_test.go new file mode 100644 --- /dev/null +++ b/spindle/ingester_test.go @@ -0,0 +1,87 @@ +package spindle + +import ( + "context" + "encoding/json" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/bluesky-social/jetstream/pkg/models" + + "tangled.org/core/api/tangled" + "tangled.org/core/spindle/config" + "tangled.org/core/tapc" +) + +func TestTapProcessEventIgnoresPullRecords(t *testing.T) { + client := &Tap{} + err := client.processEvent(context.Background(), tapc.Event{ + Type: tapc.EvtRecord, + Record: &tapc.RecordEventData{ + Live: true, + Did: syntax.DID("did:plc:jge3zxi7lgrfnvhzcgrimeo7"), + Collection: syntax.NSID(tangled.RepoPullNSID), + Rkey: syntax.RecordKey("3mrhpypucbsg4"), + Action: tapc.RecordCreateAction, + Record: json.RawMessage(`{`), + }, + }) + if err != nil { + t.Fatalf("Tap.processEvent() returned an error for a pull record: %v", err) + } +} + +func TestJetstreamToTapEventMarksPullRecordsLive(t *testing.T) { + tests := []struct { + name string + operation string + action tapc.RecordAction + }{ + {name: "create", operation: models.CommitOperationCreate, action: tapc.RecordCreateAction}, + {name: "update", operation: models.CommitOperationUpdate, action: tapc.RecordUpdateAction}, + {name: "delete", operation: models.CommitOperationDelete, action: tapc.RecordDeleteAction}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + event, ok := jetstreamToTapEvent(&models.Event{ + Did: "did:plc:jge3zxi7lgrfnvhzcgrimeo7", + Kind: models.EventKindCommit, + Commit: &models.Commit{ + Operation: tt.operation, + Collection: tangled.RepoPullNSID, + RKey: "3mrhpypucbsg4", + Record: json.RawMessage(`{"title":"test"}`), + }, + }) + if !ok { + t.Fatal("jetstreamToTapEvent() rejected a valid pull event") + } + if event.Record == nil { + t.Fatal("jetstreamToTapEvent() returned no record") + } + if !event.Record.Live { + t.Error("converted pull event is not live") + } + if event.Record.Collection.String() != tangled.RepoPullNSID { + t.Errorf("collection = %q, want %q", event.Record.Collection, tangled.RepoPullNSID) + } + if event.Record.Action != tt.action { + t.Errorf("action = %q, want %q", event.Record.Action, tt.action) + } + }) + } +} + +func TestEmbeddedTapDoesNotSubscribeToPullRecords(t *testing.T) { + tcfg := newEmbeddedTapConfig(&config.Config{}) + + if tcfg.SignalCollection == tangled.RepoPullNSID { + t.Errorf("SignalCollection = %q, must not ingest pull records", tcfg.SignalCollection) + } + for _, collection := range tcfg.CollectionFilters { + if collection == tangled.RepoPullNSID { + t.Errorf("CollectionFilters includes %q", tangled.RepoPullNSID) + } + } +} diff --git a/spindle/tapclient.go b/spindle/tapclient.go --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -82,8 +82,6 @@ case tangled.RepoNSID: return t.processRepo(ctx, evt.Record) case tangled.RepoCollaboratorNSID: return t.processCollaborator(ctx, evt.Record) - case tangled.RepoPullNSID: - return t.processPull(ctx, evt.Record) } return nil } @@ -315,8 +313,8 @@ } return nil } -func (t *Tap) processPull(ctx context.Context, evt *tapc.RecordEventData) error { - l := t.logger.With("collection", evt.Collection, "did", evt.Did, "rkey", evt.Rkey) +func (s *Spindle) processPull(ctx context.Context, evt *tapc.RecordEventData) error { + l := s.l.With("component", "ingester", "collection", evt.Collection, "did", evt.Did, "rkey", evt.Rkey) // only listen to live events if !evt.Live { @@ -345,7 +343,7 @@ return nil } // skip if target repo is unknown - repo, err := t.spindle.db.GetRepoByDid(syntax.DID(record.Target.Repo)) + repo, err := s.db.GetRepoByDid(syntax.DID(record.Target.Repo)) if err != nil { l.Warn("target repo is not ingested yet", "repo", record.Target.Repo, "err", err) return fmt.Errorf("target repo is unknown") @@ -357,14 +355,14 @@ l.Warn("skipping non-branch-based PR") return nil } - latestSubmission, err := t.fetchLatestSubmission(ctx, evt.Did.String(), evt.Rkey.String(), &record) + latestSubmission, err := s.fetchLatestSubmission(ctx, evt.Did.String(), evt.Rkey.String(), &record) if err != nil { return err } sourceSha := latestSubmission.SourceRev scheme := "https" - if t.spindle.cfg.Server.Dev { + if s.cfg.Server.Dev { scheme = "http" } client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} @@ -396,11 +394,11 @@ }, }, } - repoUri := t.spindle.newRepoCloneUrl(repo.Knot, repo.RepoDid) - repoPath := t.spindle.newRepoPath(repo.RepoDid) + repoUri := s.newRepoCloneUrl(repo.Knot, repo.RepoDid) + repoPath := s.newRepoPath(repo.RepoDid) // load workflow definitions from rev (without spindle context) - rawPipeline, err := t.spindle.loadPipeline(ctx, repoUri, repoPath, sourceSha) + rawPipeline, err := s.loadPipeline(ctx, repoUri, repoPath, sourceSha) if err != nil { // don't retry l.Error("failed loading pipeline", "err", err) @@ -427,16 +425,16 @@ pipelineId := models.PipelineId{ Knot: tpl.TriggerMetadata.Repo.Knot, Rkey: tid.TID(), } - if err := t.spindle.db.CreatePipelineEvent(pipelineId.Rkey, tpl, t.spindle.n); err != nil { + if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { l.Error("failed to create pipeline event", "err", err) return nil } - sourceRepo, err := t.spindle.resolvePipelineSourceRepo(ctx, tpl.TriggerMetadata) + sourceRepo, err := s.resolvePipelineSourceRepo(ctx, tpl.TriggerMetadata) if err != nil { l.Error("failed resolving pipeline source repo", "err", err) return nil } - err = t.spindle.processPipeline(repo.RepoDid, tpl, pipelineId, sourceRepo) + err = s.processPipeline(repo.RepoDid, tpl, pipelineId, sourceRepo) if err != nil { // don't retry l.Error("failed processing pipeline", "err", err) @@ -516,9 +514,9 @@ t.logger.Warn("expired buffered collaborator events without matching repo arrival", "count", expired, "ttl", pendingCollabTTL) } } -func (t *Tap) fetchLatestSubmission(ctx context.Context, did, rkey string, record *tangled.RepoPull) (*avmodels.PullSubmission, error) { +func (s *Spindle) fetchLatestSubmission(ctx context.Context, did, rkey string, record *tangled.RepoPull) (*avmodels.PullSubmission, error) { // resolve the PR owner's identity to fetch the blob from their PDS - prOwnerIdent, err := t.spindle.res.ResolveIdent(ctx, did) + prOwnerIdent, err := s.res.ResolveIdent(ctx, did) if err != nil || prOwnerIdent.Handle.IsInvalidHandle() { return nil, fmt.Errorf("failed to resolve PR owner handle: %w", err) }