From cfe65508c4937b7e7142a84f6398839766741355 Mon Sep 17 00:00:00 2001 From: Anirudh Oppiliappan Date: Thu, 6 Aug 2026 15:56:01 +0300 Subject: [PATCH] spindle: fire webhooks from push, pull and rename events push from processKnotStream (even when CI is skipped), pull_request:* from processPull/processPullStatus, and repository:renamed from rename detection in processRepo. Signed-off-by: dawn --- spindle/server.go | 14 +++++++ spindle/tapclient.go | 79 +++++++++++++++++++++++++++++++++++++++ spindle/tapclient_test.go | 11 ++++++ 3 files changed, 104 insertions(+) diff --git a/spindle/server.go b/spindle/server.go index 98b653a0..e42275ea 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -53,6 +53,7 @@ import ( "tangled.org/core/spindle/quota" "tangled.org/core/spindle/secrets" "tangled.org/core/spindle/storage" + "tangled.org/core/spindle/webhook" "tangled.org/core/spindle/xrpc" "tangled.org/core/tid" "tangled.org/core/workflow" @@ -86,6 +87,7 @@ type Spindle struct { e *rbac.Enforcer l *slog.Logger n *notifier.Notifier + wh *webhook.Service metrics *observability.Metrics engs map[string]models.Engine jobWake chan struct{} @@ -163,6 +165,7 @@ func New(ctx context.Context, cfg *config.Config, d *db.DB, engines map[string]m db: d, l: logger, n: &n, + wh: webhook.New(d, cfg.Server.Dev), metrics: metrics, engs: engines, cfg: cfg, @@ -495,6 +498,12 @@ func (s *Spindle) Start(ctx context.Context) error { cancelDrain() } + if s.cfg.Role != config.RoleExecutor { + webhookCtx, cancelWebhooks := context.WithTimeout(context.Background(), serverShutdownTimeout) + s.wh.Close(webhookCtx) + cancelWebhooks() + } + cancelRun() waitCtx, cancelWait := context.WithTimeout(context.Background(), executorShutdownTimeout) if err := waitForExecutor(waitCtx); err != nil { @@ -802,6 +811,7 @@ func (s *Spindle) XrpcRouter() http.Handler { Resolver: s.res, Vault: s.vault, Notifier: s.Notifier(), + Webhooks: s.wh, ServiceAuth: serviceAuth, Trigger: s, QuotaStore: db.NewQuotaStore(s.db, quota.Defaults{}), @@ -859,6 +869,10 @@ func (s *Spindle) processKnotStreamInner(ctx context.Context, src eventconsumer. return fmt.Errorf("repo knot does not match event source: %s != %s", src.Host, repo.Knot) } + // deliver push webhooks independently of CI; a skip-ci push option + // should not suppress webhook notifications + s.wh.FirePush(ctx, repo, event.CommitterDid, event.Ref, event.OldSha, event.NewSha) + if kgit.HasSkipCIPushOption(event.PushOptions) { l.Info("push event requested ci skip, skipping the event") return nil diff --git a/spindle/tapclient.go b/spindle/tapclient.go index 927c61b1..81eb06dd 100644 --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -191,6 +191,22 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error return nil } + // a rename changes the record key while the repoDid stays constant, so + // the repoDid already existing under a different rkey signals a rename. + repoName := rkey.String() + if record.Name != nil && *record.Name != "" { + repoName = *record.Name + } + renamed := false + oldName := "" + if !knownRepo && existingRepo != nil && existingRepo.Owner == ownerDid && existingRepo.Rkey != rkey { + renamed = true + oldName = existingRepo.Name + if oldName == "" { + oldName = existingRepo.Rkey.String() + } + } + if err := t.spindle.e.AddRepo(ownerDid.String(), rbac.ThisServer, repoDid.String()); err != nil { l.Error("failed to add repo policy", "err", err) return fmt.Errorf("add repo policy: %w", err) @@ -204,6 +220,7 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error Owner: ownerDid, Rkey: rkey, RepoDid: repoDid, + Name: repoName, CreatedAt: record.CreatedAt, } @@ -231,6 +248,11 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error l.Info("collapsed rename leftovers", "owner", ownerDid, "repo_did", repoDid, "removed", removed) } + if renamed { + l.Info("repo renamed, firing webhook", "old", oldName, "new", repoName) + t.spindle.wh.FireRename(ctx, ownerDid, &repo, oldName, repoName) + } + if e := t.spindle.embedTap; e == nil || !e.closed.Load() { if err := t.tap.AddRepos(ctx, []syntax.DID{ownerDid}); err != nil { l.Warn("tap AddRepos rejected", "did", ownerDid, "err", err) @@ -383,6 +405,24 @@ func (s *Spindle) processPull(ctx context.Context, evt *tapc.RecordEventData) er // for open/synchronize the event author is the pull record author. pullAuthor := evt.Did.String() + + appended := false + if record.Target != nil { + var err error + appended, err = s.db.ObservePullRounds( + syntax.DID(record.Target.Repo), evt.Rkey.String(), len(record.Rounds), + ) + if err != nil { + l.Warn("failed to record pull rounds", "err", err) + } + } + switch { + case evt.Action == tapc.RecordCreateAction: + s.firePullWebhook(ctx, models.WebhookEventPullRequestCreated, "created", evt.Did, evt.Did, &record) + case appended: + s.firePullWebhook(ctx, models.WebhookEventPullRequestResubmitted, "resubmitted", evt.Did, evt.Did, &record) + } + return s.triggerPullRequestPipeline(ctx, l, pullAuthor, pullAuthor, evt.Rkey.String(), &record, action) case tapc.RecordDeleteAction: // no-op @@ -441,9 +481,48 @@ func (s *Spindle) processPullStatus(ctx context.Context, evt *tapc.RecordEventDa } l = l.With("pull", record.Pull, "action", action, "actor", actorDid) + + if whEvent, ok := pullStatusWebhookEvent(record.Status); ok { + s.firePullWebhook(ctx, whEvent, action, syntax.DID(pullDid), evt.Did, pull) + } + return s.triggerPullRequestPipeline(ctx, l, actorDid, pullDid, pullRkey, pull, action) } +// pullStatusWebhookEvent maps a pull.status variant to its webhook event. +func pullStatusWebhookEvent(status string) (models.WebhookEvent, bool) { + switch status { + case tangled.RepoPullStatusOpen: + return models.WebhookEventPullRequestReopened, true + case tangled.RepoPullStatusClosed: + return models.WebhookEventPullRequestClosed, true + case tangled.RepoPullStatusMerged: + return models.WebhookEventPullRequestMerged, true + default: + return "", false + } +} + +// firePullWebhook delivers a pull_request:* webhook for the target repo, if that +// repo is served by this spindle and has matching webhooks configured. +func (s *Spindle) firePullWebhook(ctx context.Context, event models.WebhookEvent, action string, pullAuthor, sender syntax.DID, record *tangled.RepoPull) { + if record.Target == nil { + return + } + repo, err := s.db.GetRepoByDid(syntax.DID(record.Target.Repo)) + if err != nil { + // target repo is not served by this spindle; nothing to deliver + return + } + if ban, err := s.db.IsBanned(repo.RepoDid, repo.Owner); err != nil { + s.l.Warn("failed to check bans before firing webhook", "repo", repo.RepoDid, "err", err) + return + } else if ban != nil { + return + } + s.wh.FirePullRequest(ctx, event, action, repo, record, pullAuthor, sender) +} + // 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 diff --git a/spindle/tapclient_test.go b/spindle/tapclient_test.go index f7501a93..a197a49b 100644 --- a/spindle/tapclient_test.go +++ b/spindle/tapclient_test.go @@ -25,6 +25,7 @@ import ( "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/secrets" + "tangled.org/core/spindle/webhook" "tangled.org/core/tapc" "tangled.org/core/workflow" ) @@ -56,6 +57,7 @@ func TestDeclareTapInterestBootstrapsConfiguredOwner(t *testing.T) { db: d, l: slog.Default(), cfg: cfg, + wh: webhook.New(d, false), } tapClient := tapc.NewClient(server.URL, "") s.tap = &Tap{spindle: s, tap: tapClient} @@ -94,6 +96,7 @@ func TestProcessRepo_MembershipCheck(t *testing.T) { ks: ks, jc: jc, rootCtx: context.Background(), + wh: webhook.New(d, false), } tap := &Tap{ @@ -182,6 +185,7 @@ func TestProcessPull_PushAllowedCheck(t *testing.T) { res: idresolver.DefaultResolver("https://plc.test"), jc: jc, rootCtx: context.Background(), + wh: webhook.New(d, false), } repoOwnerDid := syntax.DID("did:plc:repoowner") @@ -273,6 +277,7 @@ func TestProcessRepo_HijackRepoDidCheck(t *testing.T) { ks: ks, jc: jc, rootCtx: context.Background(), + wh: webhook.New(d, false), } tap := &Tap{ @@ -368,6 +373,7 @@ func TestProcessCollaborator_RBAC(t *testing.T) { res: resolver, jc: jc, rootCtx: context.Background(), + wh: webhook.New(d, false), } tap := &Tap{ @@ -510,6 +516,7 @@ func TestTeardownRepo_RBAC(t *testing.T) { n: ptr(notifier.New()), engs: map[string]models.Engine{}, rootCtx: context.Background(), + wh: webhook.New(d, false), } tap := &Tap{ @@ -622,6 +629,7 @@ func TestProcessRepo_ForgeDeleteRejection(t *testing.T) { cfg: cfg, jc: jc, rootCtx: context.Background(), + wh: webhook.New(d, false), } tap := &Tap{ @@ -691,6 +699,7 @@ func TestProcessCollaborator_ForgeDeleteRejection(t *testing.T) { res: idresolver.DefaultResolver("https://plc.test"), jc: jc, rootCtx: context.Background(), + wh: webhook.New(d, false), } tap := &Tap{ @@ -802,6 +811,7 @@ func TestProcessPullStatus(t *testing.T) { res: idresolver.DefaultResolver("https://plc.test"), jc: jc, rootCtx: context.Background(), + wh: webhook.New(d, false), } statusRecord := func(status, pullUri string) []byte { @@ -888,6 +898,7 @@ func TestIsPullTriggerAuthorized(t *testing.T) { db: d, e: e, l: slog.Default(), + wh: webhook.New(d, false), } repoDid := syntax.DID("did:plc:targetrepo") -- 2.54.0