diff --git a/api/tangled/cbor_gen.go b/api/tangled/cbor_gen.go index eed70520..6464ec23 100644 --- a/api/tangled/cbor_gen.go +++ b/api/tangled/cbor_gen.go @@ -2699,7 +2699,11 @@ func (t *CiTrigger_PullRequest) MarshalCBOR(w io.Writer) error { } cw := cbg.NewCborWriter(w) - fieldCount := 6 + fieldCount := 7 + + if t.Action == nil { + fieldCount-- + } if t.Pull == nil { fieldCount-- @@ -2768,6 +2772,38 @@ func (t *CiTrigger_PullRequest) MarshalCBOR(w io.Writer) error { return err } + // t.Action (string) (string) + if t.Action != nil { + + if len("action") > 1000000 { + return xerrors.Errorf("Value in field \"action\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("action"))); err != nil { + return err + } + if _, err := cw.WriteString(string("action")); err != nil { + return err + } + + if t.Action == nil { + if _, err := cw.Write(cbg.CborNull); err != nil { + return err + } + } else { + if len(*t.Action) > 1000000 { + return xerrors.Errorf("Value in field t.Action was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(*t.Action))); err != nil { + return err + } + if _, err := cw.WriteString(string(*t.Action)); err != nil { + return err + } + } + } + // t.SourceSha (string) (string) if len("sourceSha") > 1000000 { return xerrors.Errorf("Value in field \"sourceSha\" was too long") @@ -2953,6 +2989,27 @@ func (t *CiTrigger_PullRequest) UnmarshalCBOR(r io.Reader) (err error) { t.LexiconTypeID = string(sval) } + // t.Action (string) (string) + case "action": + + { + b, err := cr.ReadByte() + if err != nil { + return err + } + if b != cbg.CborNull[0] { + if err := cr.UnreadByte(); err != nil { + return err + } + + sval, err := cbg.ReadStringWithMax(cr, 1000000) + if err != nil { + return err + } + + t.Action = (*string)(&sval) + } + } // t.SourceSha (string) (string) case "sourceSha": @@ -8387,7 +8444,11 @@ func (t *Pipeline_PullRequestTriggerData) MarshalCBOR(w io.Writer) error { } cw := cbg.NewCborWriter(w) - fieldCount := 4 + fieldCount := 5 + + if t.Action == nil { + fieldCount-- + } if t.Pull == nil { fieldCount-- @@ -8429,6 +8490,38 @@ func (t *Pipeline_PullRequestTriggerData) MarshalCBOR(w io.Writer) error { } } + // t.Action (string) (string) + if t.Action != nil { + + if len("action") > 1000000 { + return xerrors.Errorf("Value in field \"action\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("action"))); err != nil { + return err + } + if _, err := cw.WriteString(string("action")); err != nil { + return err + } + + if t.Action == nil { + if _, err := cw.Write(cbg.CborNull); err != nil { + return err + } + } else { + if len(*t.Action) > 1000000 { + return xerrors.Errorf("Value in field t.Action was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(*t.Action))); err != nil { + return err + } + if _, err := cw.WriteString(string(*t.Action)); err != nil { + return err + } + } + } + // t.SourceSha (string) (string) if len("sourceSha") > 1000000 { return xerrors.Errorf("Value in field \"sourceSha\" was too long") @@ -8562,6 +8655,27 @@ func (t *Pipeline_PullRequestTriggerData) UnmarshalCBOR(r io.Reader) (err error) t.Pull = (*string)(&sval) } } + // t.Action (string) (string) + case "action": + + { + b, err := cr.ReadByte() + if err != nil { + return err + } + if b != cbg.CborNull[0] { + if err := cr.UnreadByte(); err != nil { + return err + } + + sval, err := cbg.ReadStringWithMax(cr, 1000000) + if err != nil { + return err + } + + t.Action = (*string)(&sval) + } + } // t.SourceSha (string) (string) case "sourceSha": diff --git a/api/tangled/citrigger.go b/api/tangled/citrigger.go index 3229d527..181a3ba9 100644 --- a/api/tangled/citrigger.go +++ b/api/tangled/citrigger.go @@ -33,6 +33,8 @@ type CiTrigger_Pair struct { // RECORDTYPE: CiTrigger_PullRequest type CiTrigger_PullRequest struct { LexiconTypeID string `json:"$type,const=sh.tangled.ci.trigger#pullRequest" cborgen:"$type,const=sh.tangled.ci.trigger#pullRequest"` + // action: the pull request lifecycle action that produced this trigger + Action *string `json:"action,omitempty" cborgen:"action,omitempty"` // pull: AT-URI of the sh.tangled.repo.pull record this run belongs to Pull *string `json:"pull,omitempty" cborgen:"pull,omitempty"` SourceBranch *string `json:"sourceBranch,omitempty" cborgen:"sourceBranch,omitempty"` diff --git a/api/tangled/tangledpipeline.go b/api/tangled/tangledpipeline.go index da00f406..4f003e0a 100644 --- a/api/tangled/tangledpipeline.go +++ b/api/tangled/tangledpipeline.go @@ -49,6 +49,8 @@ type Pipeline_Pair struct { // Pipeline_PullRequestTriggerData is a "pullRequestTriggerData" in the sh.tangled.pipeline schema. type Pipeline_PullRequestTriggerData struct { + // action: the pull request lifecycle action that produced this trigger + Action *string `json:"action,omitempty" cborgen:"action,omitempty"` // pull: AT-URI of the sh.tangled.repo.pull record this run belongs to Pull *string `json:"pull,omitempty" cborgen:"pull,omitempty"` SourceBranch string `json:"sourceBranch" cborgen:"sourceBranch"` diff --git a/lexicons/ci/trigger.json b/lexicons/ci/trigger.json index 15c49e1d..9974ca7b 100644 --- a/lexicons/ci/trigger.json +++ b/lexicons/ci/trigger.json @@ -26,6 +26,11 @@ "required": ["targetBranch", "sourceSha"], "description": "TODO: reference PR record with strongRef instead of embedding raw values", "properties": { + "action": { + "type": "string", + "description": "the pull request lifecycle action that produced this trigger", + "enum": ["opened", "reopened", "closed", "merged", "synchronize"] + }, "sourceRepo": { "type": "string", "format": "did", diff --git a/lexicons/pipeline/pipeline.json b/lexicons/pipeline/pipeline.json index 9deb76fd..c710b9ff 100644 --- a/lexicons/pipeline/pipeline.json +++ b/lexicons/pipeline/pipeline.json @@ -126,6 +126,17 @@ "sourceSha" ], "properties": { + "action": { + "type": "string", + "description": "the pull request lifecycle action that produced this trigger", + "enum": [ + "opened", + "reopened", + "closed", + "merged", + "synchronize" + ] + }, "sourceBranch": { "type": "string" }, diff --git a/spindle/db/pipelines.go b/spindle/db/pipelines.go index 8cd4eeac..028f13f8 100644 --- a/spindle/db/pipelines.go +++ b/spindle/db/pipelines.go @@ -160,6 +160,7 @@ func (d *DB) mapToCiPipeline(rkey string, created int64, raw tangled.Pipeline) ( if raw.TriggerMetadata.PullRequest != nil { commitSha = raw.TriggerMetadata.PullRequest.SourceSha trigger.CiTrigger_PullRequest = &tangled.CiTrigger_PullRequest{ + Action: raw.TriggerMetadata.PullRequest.Action, SourceBranch: &raw.TriggerMetadata.PullRequest.SourceBranch, SourceRepo: raw.TriggerMetadata.SourceRepo, SourceSha: raw.TriggerMetadata.PullRequest.SourceSha, diff --git a/spindle/ingester.go b/spindle/ingester.go index 4f2909f4..aaefaa45 100644 --- a/spindle/ingester.go +++ b/spindle/ingester.go @@ -35,6 +35,10 @@ func (s *Spindle) ingest() Ingester { if evt, ok := jetstreamToTapEvent(e); ok { err = s.processPull(ctx, evt.Record) } + case tangled.RepoPullStatusNSID: + if evt, ok := jetstreamToTapEvent(e); ok { + err = s.processPullStatus(ctx, evt.Record) + } } if err != nil { diff --git a/spindle/server.go b/spindle/server.go index 839d2e8b..51dcc119 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -120,14 +120,16 @@ func New(ctx context.Context, cfg *config.Config, d *db.DB, engines map[string]m tangled.RepoNSID, tangled.RepoCollaboratorNSID, tangled.RepoPullNSID, + tangled.RepoPullStatusNSID, } jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, "spindle", collections, nil, log.SubLogger(logger, "jetstream"), d, true, true) if err != nil { return nil, fmt.Errorf("failed to setup jetstream client: %w", err) } jc.AddDid(cfg.Server.Owner) - // pull records are created by arbitrary users too, same hack as in tap + // pull (status) records are created by arbitrary users too, same hack as in tap jc.ExemptCollection(tangled.RepoPullNSID) + jc.ExemptCollection(tangled.RepoPullStatusNSID) // Check if the spindle knows about any Dids; dids, err := d.GetAllDids() diff --git a/spindle/tapclient.go b/spindle/tapclient.go index 62ceda04..bfe9d90f 100644 --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -13,6 +13,7 @@ import ( "sync" "time" + comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/syntax" indigoxrpc "github.com/bluesky-social/indigo/xrpc" "tangled.org/core/api/tangled" @@ -369,129 +370,251 @@ func (s *Spindle) processPull(ctx context.Context, evt *tapc.RecordEventData) er return fmt.Errorf("parsing record: %w", err) } - // ignore legacy records - if record.Target == nil { - l.Info("ignoring pull record: target repo is nil") - return nil + action := workflow.PullRequestActionOpened + if evt.Action == tapc.RecordUpdateAction { + action = workflow.PullRequestActionSynchronize } - // ignore patch-based and fork-based PRs - if record.Source == nil || record.Source.Repo != nil { - l.Info("ignoring pull record: not a branch-based pull request") - return nil - } + // for open/synchronize the event author is the pull record author. + pullAuthor := evt.Did.String() + return s.triggerPullRequestPipeline(ctx, l, pullAuthor, pullAuthor, evt.Rkey.String(), &record, action) + case tapc.RecordDeleteAction: + // no-op + } + return nil +} - // skip if target repo is unknown - 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") - } +// processPullStatus reacts to sh.tangled.repo.pull.status records, which record +// pull request state transitions (reopen/close/merge). Unlike the pull record +// itself, the status record only references the pull by AT-URI, so we resolve +// and fetch the pull record before building the trigger. +func (s *Spindle) processPullStatus(ctx context.Context, evt *tapc.RecordEventData) error { + l := s.l.With("component", "ingester", "collection", evt.Collection, "did", evt.Did, "rkey", evt.Rkey) - // only accept branch-based PR (excluding patch-based and fork-based) - if record.Source == nil || record.Source.Repo != nil { - l.Warn("skipping non-branch-based PR") - return nil - } + // only listen to live events + if !evt.Live { + l.Info("skipping backfill event", "event", evt.AtUri()) + return nil + } - // check if pull record author has push access to target repo - allowed, err := s.e.IsPushAllowed(evt.Did.String(), rbac.ThisServer, repo.RepoDid.String()) - if err != nil { - return fmt.Errorf("checking push access for pull record author: %w", err) - } - if !allowed { - l.Warn("rejecting pull-triggered pipeline. author has no push access", - "author", evt.Did, "repo", repo.RepoDid) - return nil - } + // status records are append-only; only creation is meaningful + if evt.Action != tapc.RecordCreateAction { + return nil + } - latestSubmission, err := s.fetchLatestSubmission(ctx, evt.Did.String(), evt.Rkey.String(), &record) - if err != nil { - return err - } - sourceSha := latestSubmission.SourceRev + record := tangled.RepoPullStatus{} + if err := json.Unmarshal(evt.Record, &record); err != nil { + l.Error("invalid record", "err", err) + return fmt.Errorf("parsing record: %w", err) + } - scheme := "https" - if s.cfg.Server.Dev { - scheme = "http" - } - client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} + action, ok := pullStatusAction(record.Status) + if !ok { + l.Info("ignoring pull status record: unknown status", "status", record.Status) + return nil + } - // fetch current default branch - defaultBranch, _ := func(repo syntax.DID) (string, error) { - defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) - if err != nil { - return "", err - } - return defaultBranchOut.Name, nil - }(repo.RepoDid) - - compiler := workflow.Compiler{ - Trigger: tangled.Pipeline_TriggerMetadata{ - Kind: string(workflow.TriggerKindPullRequest), - PullRequest: &tangled.Pipeline_PullRequestTriggerData{ - SourceBranch: record.Source.Branch, - SourceSha: sourceSha, - TargetBranch: record.Target.Branch, - }, - Repo: &tangled.Pipeline_TriggerRepo{ - Did: repo.Owner.String(), - Knot: repo.Knot, - Repo: (*string)(&repo.Rkey), - RepoDid: (*string)(&repo.RepoDid), - DefaultBranch: defaultBranch, - }, - }, - } + pullUri, err := syntax.ParseATURI(record.Pull) + if err != nil { + l.Error("invalid pull at-uri in status record", "pull", record.Pull, "err", err) + return nil + } + if pullUri.Collection().String() != tangled.RepoPullNSID { + l.Info("ignoring pull status record: subject is not a pull", "collection", pullUri.Collection()) + return nil + } - repoUri := s.newRepoCloneUrl(repo.Knot, repo.RepoDid) - repoPath := s.newRepoPath(repo.RepoDid) + pullDid := pullUri.Authority().String() + pullRkey := pullUri.RecordKey().String() + actorDid := evt.Did.String() - // load workflow definitions from rev (without spindle context) - rawPipeline, err := s.loadPipeline(ctx, repoUri, repoPath, sourceSha) - if err != nil { - // don't retry - l.Error("failed loading pipeline", "err", err) - return nil - } - if len(rawPipeline) == 0 { - l.Info("no workflow definition find for the repo. skipping the event") - return nil - } - tpl := compiler.Compile(compiler.Parse(rawPipeline)) - // TODO: pass compile error to workflow log - for _, w := range compiler.Diagnostics.Errors { - l.Error(w.String()) - } - for _, w := range compiler.Diagnostics.Warnings { - l.Warn(w.String()) - } - if len(tpl.Workflows) == 0 { - l.Info("no workflow matching trigger 'pull_request'. skipping the event") - return nil - } + pull, err := s.fetchPullRecord(ctx, pullDid, pullRkey) + if err != nil { + l.Error("failed to fetch pull record for status event", "pull", record.Pull, "err", err) + return fmt.Errorf("fetch pull record: %w", err) + } - pipelineId := models.PipelineId{ - Knot: tpl.TriggerMetadata.Repo.Knot, - Rkey: tid.TID(), - } - 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 := s.resolvePipelineSourceRepo(ctx, tpl.TriggerMetadata) - if err != nil { - l.Error("failed resolving pipeline source repo", "err", err) - return nil - } - err = s.processPipeline(repo.RepoDid, tpl, pipelineId, sourceRepo) + l = l.With("pull", record.Pull, "action", action, "actor", actorDid) + return s.triggerPullRequestPipeline(ctx, l, actorDid, pullDid, pullRkey, pull, action) +} + +// pullStatusAction maps a sh.tangled.repo.pull.status variant to the +// corresponding pull_request trigger action. A status.open record is only ever +// written on reopen (initial creation emits no status record), so it maps to +// "reopened". +func pullStatusAction(status string) (string, bool) { + switch status { + case tangled.RepoPullStatusOpen: + return workflow.PullRequestActionReopened, true + case tangled.RepoPullStatusClosed: + return workflow.PullRequestActionClosed, true + case tangled.RepoPullStatusMerged: + return workflow.PullRequestActionMerged, true + default: + return "", false + } +} + +// fetchPullRecord retrieves a sh.tangled.repo.pull record from its author's PDS. +func (s *Spindle) fetchPullRecord(ctx context.Context, did, rkey string) (*tangled.RepoPull, error) { + ident, err := s.res.ResolveIdent(ctx, did) + if err != nil || ident.Handle.IsInvalidHandle() { + return nil, fmt.Errorf("failed to resolve pull owner: %w", err) + } + + client := &indigoxrpc.Client{Host: ident.PDSEndpoint()} + resp, err := comatproto.RepoGetRecord(ctx, client, "", tangled.RepoPullNSID, did, rkey) + if err != nil { + return nil, fmt.Errorf("fetching pull record: %w", err) + } + + pull, ok := resp.Value.Val.(*tangled.RepoPull) + if !ok { + return nil, fmt.Errorf("record %s/%s is not a pull record", did, rkey) + } + return pull, nil +} + +// isPullTriggerAuthorized reports whether a pull_request pipeline may be +// triggered on repoDid. The pull author must always have push access to the +// target repo; the event actor must either be the pull author or also have push +// access. For open/synchronize the actor and pull author are the same DID, so +// this reduces to the pull author's push check. +func (s *Spindle) isPullTriggerAuthorized(eventDid, pullDid, repoDid string) (bool, error) { + pullHasPush, err := s.e.IsPushAllowed(pullDid, rbac.ThisServer, repoDid) + if err != nil || !pullHasPush { + return false, err + } + + if eventDid == pullDid { + return true, nil + } + + return s.e.IsPushAllowed(eventDid, rbac.ThisServer, repoDid) +} + +// triggerPullRequestPipeline builds and runs a pull_request-triggered pipeline +// for the given pull record. eventDid is the DID that authored the firehose +// event (the actor); pullDid/pullRkey identify the sh.tangled.repo.pull record +// (used to fetch the latest submission and as the pull author for the +// authorization check); action is the pull_request lifecycle action carried +// into the trigger metadata for `types` matching. +func (s *Spindle) triggerPullRequestPipeline(ctx context.Context, l *slog.Logger, eventDid, pullDid, pullRkey string, record *tangled.RepoPull, action string) error { + // ignore legacy records + if record.Target == nil { + l.Info("ignoring pull record: target repo is nil") + return nil + } + + // ignore patch-based and fork-based PRs + if record.Source == nil || record.Source.Repo != nil { + l.Info("ignoring pull record: not a branch-based pull request") + return nil + } + + // skip if target repo is unknown + 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") + } + + // authorize the actor against the target repo + allowed, err := s.isPullTriggerAuthorized(eventDid, pullDid, repo.RepoDid.String()) + if err != nil { + return fmt.Errorf("authorizing pull-triggered pipeline: %w", err) + } + if !allowed { + l.Warn("rejecting pull-triggered pipeline: actor is not authorized", + "actor", eventDid, "author", pullDid, "repo", repo.RepoDid) + return nil + } + + latestSubmission, err := s.fetchLatestSubmission(ctx, pullDid, pullRkey, record) + if err != nil { + return err + } + sourceSha := latestSubmission.SourceRev + + scheme := "https" + if s.cfg.Server.Dev { + scheme = "http" + } + client := &indigoxrpc.Client{Host: fmt.Sprintf("%s://%s", scheme, repo.Knot)} + + // fetch current default branch + defaultBranch, _ := func(repo syntax.DID) (string, error) { + defaultBranchOut, err := tangled.RepoGetDefaultBranch(ctx, client, repo.String()) if err != nil { - // don't retry - l.Error("failed processing pipeline", "err", err) - return nil - } - case tapc.RecordDeleteAction: - // no-op + return "", err + } + return defaultBranchOut.Name, nil + }(repo.RepoDid) + + compiler := workflow.Compiler{ + Trigger: tangled.Pipeline_TriggerMetadata{ + Kind: string(workflow.TriggerKindPullRequest), + PullRequest: &tangled.Pipeline_PullRequestTriggerData{ + Action: &action, + SourceBranch: record.Source.Branch, + SourceSha: sourceSha, + TargetBranch: record.Target.Branch, + }, + Repo: &tangled.Pipeline_TriggerRepo{ + Did: repo.Owner.String(), + Knot: repo.Knot, + Repo: (*string)(&repo.Rkey), + RepoDid: (*string)(&repo.RepoDid), + DefaultBranch: defaultBranch, + }, + }, + } + + repoUri := s.newRepoCloneUrl(repo.Knot, repo.RepoDid) + repoPath := s.newRepoPath(repo.RepoDid) + + // load workflow definitions from rev (without spindle context) + rawPipeline, err := s.loadPipeline(ctx, repoUri, repoPath, sourceSha) + if err != nil { + // don't retry + l.Error("failed loading pipeline", "err", err) + return nil + } + if len(rawPipeline) == 0 { + l.Info("no workflow definition find for the repo. skipping the event") + return nil + } + tpl := compiler.Compile(compiler.Parse(rawPipeline)) + // TODO: pass compile error to workflow log + for _, w := range compiler.Diagnostics.Errors { + l.Error(w.String()) + } + for _, w := range compiler.Diagnostics.Warnings { + l.Warn(w.String()) + } + if len(tpl.Workflows) == 0 { + l.Info("no workflow matching trigger 'pull_request'. skipping the event") + return nil + } + + pipelineId := models.PipelineId{ + Knot: tpl.TriggerMetadata.Repo.Knot, + Rkey: tid.TID(), + } + 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 := s.resolvePipelineSourceRepo(ctx, tpl.TriggerMetadata) + if err != nil { + l.Error("failed resolving pipeline source repo", "err", err) + return nil + } + err = s.processPipeline(repo.RepoDid, tpl, pipelineId, sourceRepo) + if err != nil { + // don't retry + l.Error("failed processing pipeline", "err", err) + return nil } return nil } diff --git a/spindle/tapclient_test.go b/spindle/tapclient_test.go index 7cf72670..4bcaf738 100644 --- a/spindle/tapclient_test.go +++ b/spindle/tapclient_test.go @@ -17,6 +17,7 @@ import ( "tangled.org/core/rbac" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" + "tangled.org/core/workflow" "tangled.org/core/tapc" ) @@ -672,3 +673,182 @@ func TestProcessCollaborator_ForgeDeleteRejection(t *testing.T) { t.Fatal("collaborator policy was removed from Casbin by forged delete") } } + +func TestPullStatusAction(t *testing.T) { + tests := []struct { + name string + status string + wantAction string + wantOK bool + }{ + {"open maps to reopened", tangled.RepoPullStatusOpen, workflow.PullRequestActionReopened, true}, + {"closed maps to closed", tangled.RepoPullStatusClosed, workflow.PullRequestActionClosed, true}, + {"merged maps to merged", tangled.RepoPullStatusMerged, workflow.PullRequestActionMerged, true}, + {"unknown status is rejected", "sh.tangled.repo.pull.status.bogus", "", false}, + {"empty status is rejected", "", "", false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + action, ok := pullStatusAction(tt.status) + if ok != tt.wantOK { + t.Fatalf("pullStatusAction(%q) ok = %v, want %v", tt.status, ok, tt.wantOK) + } + if action != tt.wantAction { + t.Fatalf("pullStatusAction(%q) action = %q, want %q", tt.status, action, tt.wantAction) + } + }) + } +} + +func TestProcessPullStatus(t *testing.T) { + d, e := newTestSpindleDB(t) + + cfg := &config.Config{} + cfg.Server.Hostname = "spindle.test" + + jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false) + if jcerr != nil { + t.Fatalf("NewJetstreamClient: %v", jcerr) + } + s := &Spindle{ + db: d, + e: e, + l: slog.Default(), + cfg: cfg, + res: idresolver.DefaultResolver("https://plc.test"), + jc: jc, + rootCtx: context.Background(), + } + + statusRecord := func(status, pullUri string) []byte { + rec := tangled.RepoPullStatus{ + Status: status, + Pull: pullUri, + CreatedAt: time.Now().Format(time.RFC3339), + } + b, _ := json.Marshal(rec) + return b + } + + validPullUri := "at://did:plc:pullowner/sh.tangled.repo.pull/pull-rkey-1" + + // non-create actions are ignored + if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{ + Live: true, + Did: syntax.DID("did:plc:actor"), + Rkey: "status-rkey", + Collection: syntax.NSID(tangled.RepoPullStatusNSID), + Action: tapc.RecordUpdateAction, + Record: statusRecord(tangled.RepoPullStatusClosed, validPullUri), + }); err != nil { + t.Fatalf("update action should be a no-op, got: %v", err) + } + + // unknown status variant is skipped without error + if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{ + Live: true, + Did: syntax.DID("did:plc:actor"), + Rkey: "status-rkey", + Collection: syntax.NSID(tangled.RepoPullStatusNSID), + Action: tapc.RecordCreateAction, + Record: statusRecord("sh.tangled.repo.pull.status.bogus", validPullUri), + }); err != nil { + t.Fatalf("unknown status should be skipped, got: %v", err) + } + + // a malformed pull at-uri is skipped without error + if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{ + Live: true, + Did: syntax.DID("did:plc:actor"), + Rkey: "status-rkey", + Collection: syntax.NSID(tangled.RepoPullStatusNSID), + Action: tapc.RecordCreateAction, + Record: statusRecord(tangled.RepoPullStatusClosed, "not-an-at-uri"), + }); err != nil { + t.Fatalf("invalid pull at-uri should be skipped, got: %v", err) + } + + // a status pointing at a non-pull subject is skipped without error + if err := s.processPullStatus(context.Background(), &tapc.RecordEventData{ + Live: true, + Did: syntax.DID("did:plc:actor"), + Rkey: "status-rkey", + Collection: syntax.NSID(tangled.RepoPullStatusNSID), + Action: tapc.RecordCreateAction, + Record: statusRecord(tangled.RepoPullStatusClosed, "at://did:plc:x/sh.tangled.repo.issue/y"), + }); err != nil { + t.Fatalf("non-pull subject should be skipped, got: %v", err) + } + + // a valid close event resolves the pull record; fetch fails because plc/pds + // are not real, confirming we reached the fetch stage with a mapped action. + err := s.processPullStatus(context.Background(), &tapc.RecordEventData{ + Live: true, + Did: syntax.DID("did:plc:actor"), + Rkey: "status-rkey", + Collection: syntax.NSID(tangled.RepoPullStatusNSID), + Action: tapc.RecordCreateAction, + Record: statusRecord(tangled.RepoPullStatusClosed, validPullUri), + }) + if err == nil { + t.Fatal("expected error fetching pull record against fake pds, got nil") + } + if !strings.Contains(err.Error(), "fetch pull record") { + t.Fatalf("expected fetch pull record error, got: %v", err) + } +} + +func TestIsPullTriggerAuthorized(t *testing.T) { + d, e := newTestSpindleDB(t) + s := &Spindle{ + db: d, + e: e, + l: slog.Default(), + } + + repoDid := syntax.DID("did:plc:targetrepo") + ownerDid := syntax.DID("did:plc:owner") // has push (repo owner) + collaboratorDid := "did:plc:collaborator" // has push + noPushAuthorDid := "did:plc:nopushauthor" // pull author without push + strangerDid := "did:plc:stranger" // no push, not the author + + if err := e.AddRepo(ownerDid.String(), rbac.ThisServer, repoDid.String()); err != nil { + t.Fatalf("AddRepo permissions: %v", err) + } + if err := e.AddCollaborator(collaboratorDid, rbac.ThisServer, repoDid.String()); err != nil { + t.Fatalf("AddCollaborator: %v", err) + } + + cases := []struct { + name string + eventDid string + pullDid string + allowed bool + }{ + // direct pr path: event author == pull author + {"direct pr by pushing author", ownerDid.String(), ownerDid.String(), true}, + {"direct pr by non-pushing author", noPushAuthorDid, noPushAuthorDid, false}, + + // status path: actor differs from pull author, pull author has push + {"actor with push acts on authorized pr", collaboratorDid, ownerDid.String(), true}, + {"pull author acts on own authorized pr", ownerDid.String(), ownerDid.String(), true}, + {"stranger without push acts on authorized pr", strangerDid, ownerDid.String(), false}, + + // pull author must always have push, even if the actor does + {"pushing actor on unauthorized pull author", ownerDid.String(), noPushAuthorDid, false}, + {"non-pushing actor on unauthorized pull author", strangerDid, noPushAuthorDid, false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + allowed, err := s.isPullTriggerAuthorized(tc.eventDid, tc.pullDid, repoDid.String()) + if err != nil { + t.Fatalf("isPullTriggerAuthorized: %v", err) + } + if allowed != tc.allowed { + t.Fatalf("event=%q pull=%q allowed=%v, want %v", tc.eventDid, tc.pullDid, allowed, tc.allowed) + } + }) + } +} diff --git a/workflow/def.go b/workflow/def.go index ddd9b7d0..0d52175e 100644 --- a/workflow/def.go +++ b/workflow/def.go @@ -34,6 +34,7 @@ type ( Constraint struct { Event StringList `yaml:"event"` + Types StringList `yaml:"types"` // optional; only applies to pull_request events. defaults to opened, reopened and synchronize Branch StringList `yaml:"branch"` // required for pull_request; for push, either branch or tag must be specified Tag StringList `yaml:"tag"` // optional; only applies to push events Paths StringList `yaml:"paths"` // optional; only run if any changed file matches a glob pattern @@ -57,8 +58,26 @@ const ( TriggerKindPush TriggerKind = "push" TriggerKindPullRequest TriggerKind = "pull_request" TriggerKindManual TriggerKind = "manual" + + // pull_request lifecycle actions, carried in the trigger metadata and + // matched against a constraint's `types` list. + PullRequestActionOpened = "opened" + PullRequestActionReopened = "reopened" + PullRequestActionClosed = "closed" + PullRequestActionMerged = "merged" + PullRequestActionSynchronize = "synchronize" ) +// DefaultPullRequestActions is the set of pull_request actions a constraint +// matches when it does not specify an explicit `types` list. This preserves +// the historic behaviour of firing on PR creation and resubmission, plus +// reopen, while leaving close/merge opt-in. +var DefaultPullRequestActions = []string{ + PullRequestActionOpened, + PullRequestActionReopened, + PullRequestActionSynchronize, +} + func (t TriggerKind) String() string { return strings.ReplaceAll(string(t), "_", " ") } @@ -131,13 +150,17 @@ func (c *Constraint) Match(trigger tangled.Pipeline_TriggerMetadata, changedFile // apply event constraints match = match && c.MatchEvent(trigger.Kind) - // apply branch constraints for PRs + // apply branch and action constraints for PRs if trigger.PullRequest != nil { matched, err := c.MatchBranch(trigger.PullRequest.TargetBranch) if err != nil { return false, err } - match = match && matched + action := "" + if trigger.PullRequest.Action != nil { + action = *trigger.PullRequest.Action + } + match = match && matched && c.MatchTypes(action) } // apply ref constraints for pushes @@ -202,6 +225,21 @@ func (c *Constraint) MatchEvent(event string) bool { return slices.Contains(c.Event, event) } +// MatchTypes reports whether a pull_request action satisfies this constraint's +// `types` filter. An empty `types` list falls back to DefaultPullRequestActions. +// A missing action (e.g. legacy trigger metadata) is treated as "opened" so +// existing pull_request workflows keep matching. +func (c *Constraint) MatchTypes(action string) bool { + if action == "" { + action = PullRequestActionOpened + } + types := []string(c.Types) + if len(types) == 0 { + types = DefaultPullRequestActions + } + return slices.Contains(types, action) +} + // Custom unmarshaller for StringList func (s *StringList) UnmarshalYAML(unmarshal func(any) error) error { var stringType string diff --git a/workflow/def_test.go b/workflow/def_test.go index 6f246f9c..eae79e4e 100644 --- a/workflow/def_test.go +++ b/workflow/def_test.go @@ -497,6 +497,105 @@ func TestConstraintMatchTag_GlobPatterns(t *testing.T) { } } +func TestUnmarshalWorkflowWithTypes(t *testing.T) { + yamlData := ` +when: + - event: pull_request + types: [opened, reopened, closed] + branch: main` + + wf, err := FromFile("test.yml", []byte(yamlData)) + assert.NoError(t, err, "YAML should unmarshal without error") + assert.Len(t, wf.When, 1) + assert.ElementsMatch(t, []string{"opened", "reopened", "closed"}, wf.When[0].Types) + assert.ElementsMatch(t, []string{"pull_request"}, wf.When[0].Event) +} + +func TestConstraintMatchTypes(t *testing.T) { + tests := []struct { + name string + constraint Constraint + action string + expected bool + }{ + {"empty types defaults to opened", Constraint{}, PullRequestActionOpened, true}, + {"empty types defaults to reopened", Constraint{}, PullRequestActionReopened, true}, + {"empty types defaults to synchronize", Constraint{}, PullRequestActionSynchronize, true}, + {"empty types does not match closed", Constraint{}, PullRequestActionClosed, false}, + {"empty types does not match merged", Constraint{}, PullRequestActionMerged, false}, + {"missing action treated as opened", Constraint{}, "", true}, + {"explicit closed matches", Constraint{Types: []string{"closed"}}, PullRequestActionClosed, true}, + {"explicit closed does not match opened", Constraint{Types: []string{"closed"}}, PullRequestActionOpened, false}, + {"merged is distinct from closed", Constraint{Types: []string{"closed"}}, PullRequestActionMerged, false}, + {"explicit merged matches", Constraint{Types: []string{"merged"}}, PullRequestActionMerged, true}, + {"multiple types", Constraint{Types: []string{"opened", "closed"}}, PullRequestActionClosed, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.expected, tt.constraint.MatchTypes(tt.action)) + }) + } +} + +func TestConstraintMatch_PullRequestTypes(t *testing.T) { + prTrigger := func(action string) tangled.Pipeline_TriggerMetadata { + return tangled.Pipeline_TriggerMetadata{ + Kind: string(TriggerKindPullRequest), + PullRequest: &tangled.Pipeline_PullRequestTriggerData{ + Action: &action, + TargetBranch: "main", + }, + } + } + + tests := []struct { + name string + constraint Constraint + action string + expected bool + }{ + { + name: "no types matches opened on target branch", + constraint: Constraint{Event: []string{"pull_request"}, Branch: []string{"main"}}, + action: PullRequestActionOpened, + expected: true, + }, + { + name: "no types does not match closed", + constraint: Constraint{Event: []string{"pull_request"}, Branch: []string{"main"}}, + action: PullRequestActionClosed, + expected: false, + }, + { + name: "explicit closed type matches close event", + constraint: Constraint{Event: []string{"pull_request"}, Types: []string{"closed"}, Branch: []string{"main"}}, + action: PullRequestActionClosed, + expected: true, + }, + { + name: "closed type does not fire on open", + constraint: Constraint{Event: []string{"pull_request"}, Types: []string{"closed"}, Branch: []string{"main"}}, + action: PullRequestActionOpened, + expected: false, + }, + { + name: "branch mismatch overrides matching type", + constraint: Constraint{Event: []string{"pull_request"}, Types: []string{"closed"}, Branch: []string{"release"}}, + action: PullRequestActionClosed, + expected: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result, err := tt.constraint.Match(prTrigger(tt.action), nil) + assert.NoError(t, err) + assert.Equal(t, tt.expected, result) + }) + } +} + func TestMatch_ManualDispatch(t *testing.T) { // manual dispatch is policy-free: every workflow matches regardless of its // declared event/branch/tag/path constraints. Selection is the caller's job.