Decouple HookTask from Repository (#17940)

At the moment a repository reference is needed for webhooks. With the
upcoming package PR we need to send webhooks without a repository
reference. For example a package is uploaded to an organization. In
theory this enables the usage of webhooks for future user actions.

This PR removes the repository id from `HookTask` and changes how the
hooks are processed (see `services/webhook/deliver.go`). In a follow up
PR I want to remove the usage of the `UniqueQueue´ and replace it with a
normal queue because there is no reason to be unique.

Co-authored-by: 6543 <6543@obermui.de>

3 files changed, 67 insertions, 96 deletions

diff --git a/services/webhook/deliver.go b/services/webhook/deliver.go
index 77744473f1..74a69c297c 100644
--- a/services/webhook/deliver.go
+++ b/services/webhook/deliver.go
@@ -23,7 +23,6 @@ import (
 	"code.gitea.io/gitea/modules/graceful"
 	"code.gitea.io/gitea/modules/hostmatcher"
 	"code.gitea.io/gitea/modules/log"
-	"code.gitea.io/gitea/modules/process"
 	"code.gitea.io/gitea/modules/proxy"
 	"code.gitea.io/gitea/modules/queue"
 	"code.gitea.io/gitea/modules/setting"
@@ -44,7 +43,7 @@ func Deliver(ctx context.Context, t *webhook_model.HookTask) error {
 			return
 		}
 		// There was a panic whilst delivering a hook...
-		log.Error("PANIC whilst trying to deliver webhook[%d] for repo[%d] to %s Panic: %v\nStacktrace: %s", t.ID, t.RepoID, w.URL, err, log.Stack(2))
+		log.Error("PANIC whilst trying to deliver webhook[%d] to %s Panic: %v\nStacktrace: %s", t.ID, w.URL, err, log.Stack(2))
 	}()
 
 	t.IsDelivered = true
@@ -202,35 +201,6 @@ func Deliver(ctx context.Context, t *webhook_model.HookTask) error {
 	return nil
 }
 
-// populateDeliverHooks checks and delivers undelivered hooks.
-func populateDeliverHooks(ctx context.Context) {
-	select {
-	case <-ctx.Done():
-		return
-	default:
-	}
-	ctx, _, finished := process.GetManager().AddTypedContext(ctx, "Service: DeliverHooks", process.SystemProcessType, true)
-	defer finished()
-	tasks, err := webhook_model.FindUndeliveredHookTasks()
-	if err != nil {
-		log.Error("DeliverHooks: %v", err)
-		return
-	}
-
-	// Update hook task status.
-	for _, t := range tasks {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
-
-		if err := addToTask(t.RepoID); err != nil {
-			log.Error("DeliverHook failed [%d]: %v", t.RepoID, err)
-		}
-	}
-}
-
 var (
 	webhookHTTPClient *http.Client
 	once              sync.Once
@@ -281,13 +251,23 @@ func Init() error {
 		},
 	}
 
-	hookQueue = queue.CreateUniqueQueue("webhook_sender", handle, "")
+	hookQueue = queue.CreateUniqueQueue("webhook_sender", handle, int64(0))
 	if hookQueue == nil {
 		return fmt.Errorf("Unable to create webhook_sender Queue")
 	}
 	go graceful.GetManager().RunWithShutdownFns(hookQueue.Run)
 
-	populateDeliverHooks(graceful.GetManager().HammerContext())
+	tasks, err := webhook_model.FindUndeliveredHookTasks(graceful.GetManager().HammerContext())
+	if err != nil {
+		log.Error("FindUndeliveredHookTasks failed: %v", err)
+		return err
+	}
+
+	for _, task := range tasks {
+		if err := enqueueHookTask(task); err != nil {
+			log.Error("enqueueHookTask failed: %v", err)
+		}
+	}
 
 	return nil
 }
diff --git a/services/webhook/webhook.go b/services/webhook/webhook.go
index 767c3701f3..e877e16eda 100644
--- a/services/webhook/webhook.go
+++ b/services/webhook/webhook.go
@@ -7,11 +7,10 @@ package webhook
 import (
 	"context"
 	"fmt"
-	"strconv"
 	"strings"
 
-	"code.gitea.io/gitea/models/db"
 	repo_model "code.gitea.io/gitea/models/repo"
+	user_model "code.gitea.io/gitea/models/user"
 	webhook_model "code.gitea.io/gitea/models/webhook"
 	"code.gitea.io/gitea/modules/git"
 	"code.gitea.io/gitea/modules/graceful"
@@ -104,49 +103,38 @@ func getPayloadBranch(p api.Payloader) string {
 	return ""
 }
 
-// handle passed PR IDs and test the PRs
-func handle(data ...queue.Data) []queue.Data {
-	for _, datum := range data {
-		repoIDStr := datum.(string)
-		log.Trace("DeliverHooks [repo_id: %v]", repoIDStr)
+// EventSource represents the source of a webhook action. Repository and/or Owner must be set.
+type EventSource struct {
+	Repository *repo_model.Repository
+	Owner      *user_model.User
+}
 
-		repoID, err := strconv.ParseInt(repoIDStr, 10, 64)
-		if err != nil {
-			log.Error("Invalid repo ID: %s", repoIDStr)
-			continue
-		}
+// handle delivers hook tasks
+func handle(data ...queue.Data) []queue.Data {
+	ctx := graceful.GetManager().HammerContext()
 
-		tasks, err := webhook_model.FindRepoUndeliveredHookTasks(repoID)
+	for _, taskID := range data {
+		task, err := webhook_model.GetHookTaskByID(ctx, taskID.(int64))
 		if err != nil {
-			log.Error("Get repository [%d] hook tasks: %v", repoID, err)
-			continue
-		}
-		for _, t := range tasks {
-			if err = Deliver(graceful.GetManager().HammerContext(), t); err != nil {
-				log.Error("deliver: %v", err)
+			log.Error("GetHookTaskByID failed: %v", err)
+		} else {
+			if err := Deliver(ctx, task); err != nil {
+				log.Error("webhook.Deliver failed: %v", err)
 			}
 		}
 	}
+
 	return nil
 }
 
-func addToTask(repoID int64) error {
-	err := hookQueue.PushFunc(strconv.FormatInt(repoID, 10), nil)
+func enqueueHookTask(task *webhook_model.HookTask) error {
+	err := hookQueue.PushFunc(task.ID, nil)
 	if err != nil && err != queue.ErrAlreadyInQueue {
 		return err
 	}
 	return nil
 }
 
-// PrepareWebhook adds special webhook to task queue for given payload.
-func PrepareWebhook(w *webhook_model.Webhook, repo *repo_model.Repository, event webhook_model.HookEventType, p api.Payloader) error {
-	if err := prepareWebhook(w, repo, event, p); err != nil {
-		return err
-	}
-
-	return addToTask(repo.ID)
-}
-
 func checkBranch(w *webhook_model.Webhook, branch string) bool {
 	if w.BranchFilter == "" || w.BranchFilter == "*" {
 		return true
@@ -162,7 +150,8 @@ func checkBranch(w *webhook_model.Webhook, branch string) bool {
 	return g.Match(branch)
 }
 
-func prepareWebhook(w *webhook_model.Webhook, repo *repo_model.Repository, event webhook_model.HookEventType, p api.Payloader) error {
+// PrepareWebhook creates a hook task and enqueues it for processing
+func PrepareWebhook(ctx context.Context, w *webhook_model.Webhook, event webhook_model.HookEventType, p api.Payloader) error {
 	// Skip sending if webhooks are disabled.
 	if setting.DisableWebhooks {
 		return nil
@@ -207,44 +196,45 @@ func prepareWebhook(w *webhook_model.Webhook, repo *repo_model.Repository, event
 		payloader = p
 	}
 
-	if err = webhook_model.CreateHookTask(&webhook_model.HookTask{
-		RepoID:    repo.ID,
+	task, err := webhook_model.CreateHookTask(ctx, &webhook_model.HookTask{
 		HookID:    w.ID,
 		Payloader: payloader,
 		EventType: event,
-	}); err != nil {
+	})
+	if err != nil {
 		return fmt.Errorf("CreateHookTask: %v", err)
 	}
-	return nil
+
+	return enqueueHookTask(task)
 }
 
 // PrepareWebhooks adds new webhooks to task queue for given payload.
-func PrepareWebhooks(repo *repo_model.Repository, event webhook_model.HookEventType, p api.Payloader) error {
-	if err := prepareWebhooks(db.DefaultContext, repo, event, p); err != nil {
-		return err
-	}
+func PrepareWebhooks(ctx context.Context, source EventSource, event webhook_model.HookEventType, p api.Payloader) error {
+	owner := source.Owner
 
-	return addToTask(repo.ID)
-}
+	var ws []*webhook_model.Webhook
 
-func prepareWebhooks(ctx context.Context, repo *repo_model.Repository, event webhook_model.HookEventType, p api.Payloader) error {
-	ws, err := webhook_model.ListWebhooksByOpts(ctx, &webhook_model.ListWebhookOptions{
-		RepoID:   repo.ID,
-		IsActive: util.OptionalBoolTrue,
-	})
-	if err != nil {
-		return fmt.Errorf("GetActiveWebhooksByRepoID: %v", err)
+	if source.Repository != nil {
+		repoHooks, err := webhook_model.ListWebhooksByOpts(ctx, &webhook_model.ListWebhookOptions{
+			RepoID:   source.Repository.ID,
+			IsActive: util.OptionalBoolTrue,
+		})
+		if err != nil {
+			return fmt.Errorf("ListWebhooksByOpts: %v", err)
+		}
+		ws = append(ws, repoHooks...)
+
+		owner = source.Repository.MustOwner()
 	}
 
-	// check if repo belongs to org and append additional webhooks
-	if repo.MustOwner().IsOrganization() {
-		// get hooks for org
+	// check if owner is an org and append additional webhooks
+	if owner != nil && owner.IsOrganization() {
 		orgHooks, err := webhook_model.ListWebhooksByOpts(ctx, &webhook_model.ListWebhookOptions{
-			OrgID:    repo.OwnerID,
+			OrgID:    owner.ID,
 			IsActive: util.OptionalBoolTrue,
 		})
 		if err != nil {
-			return fmt.Errorf("GetActiveWebhooksByOrgID: %v", err)
+			return fmt.Errorf("ListWebhooksByOpts: %v", err)
 		}
 		ws = append(ws, orgHooks...)
 	}
@@ -261,7 +251,7 @@ func prepareWebhooks(ctx context.Context, repo *repo_model.Repository, event web
 	}
 
 	for _, w := range ws {
-		if err = prepareWebhook(w, repo, event, p); err != nil {
+		if err := PrepareWebhook(ctx, w, event, p); err != nil {
 			return err
 		}
 	}
@@ -269,11 +259,11 @@ func prepareWebhooks(ctx context.Context, repo *repo_model.Repository, event web
 }
 
 // ReplayHookTask replays a webhook task
-func ReplayHookTask(w *webhook_model.Webhook, uuid string) error {
-	t, err := webhook_model.ReplayHookTask(w.ID, uuid)
+func ReplayHookTask(ctx context.Context, w *webhook_model.Webhook, uuid string) error {
+	task, err := webhook_model.ReplayHookTask(ctx, w.ID, uuid)
 	if err != nil {
 		return err
 	}
 
-	return addToTask(t.RepoID)
+	return enqueueHookTask(task)
 }
diff --git a/services/webhook/webhook_test.go b/services/webhook/webhook_test.go
index 1887cc71fe..8d44aa504a 100644
--- a/services/webhook/webhook_test.go
+++ b/services/webhook/webhook_test.go
@@ -7,6 +7,7 @@ package webhook
 import (
 	"testing"
 
+	"code.gitea.io/gitea/models/db"
 	repo_model "code.gitea.io/gitea/models/repo"
 	"code.gitea.io/gitea/models/unittest"
 	webhook_model "code.gitea.io/gitea/models/webhook"
@@ -32,12 +33,12 @@ func TestPrepareWebhooks(t *testing.T) {
 
 	repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 1})
 	hookTasks := []*webhook_model.HookTask{
-		{RepoID: repo.ID, HookID: 1, EventType: webhook_model.HookEventPush},
+		{HookID: 1, EventType: webhook_model.HookEventPush},
 	}
 	for _, hookTask := range hookTasks {
 		unittest.AssertNotExistsBean(t, hookTask)
 	}
-	assert.NoError(t, PrepareWebhooks(repo, webhook_model.HookEventPush, &api.PushPayload{Commits: []*api.PayloadCommit{{}}}))
+	assert.NoError(t, PrepareWebhooks(db.DefaultContext, EventSource{Repository: repo}, webhook_model.HookEventPush, &api.PushPayload{Commits: []*api.PayloadCommit{{}}}))
 	for _, hookTask := range hookTasks {
 		unittest.AssertExistsAndLoadBean(t, hookTask)
 	}
@@ -48,13 +49,13 @@ func TestPrepareWebhooksBranchFilterMatch(t *testing.T) {
 
 	repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 2})
 	hookTasks := []*webhook_model.HookTask{
-		{RepoID: repo.ID, HookID: 4, EventType: webhook_model.HookEventPush},
+		{HookID: 4, EventType: webhook_model.HookEventPush},
 	}
 	for _, hookTask := range hookTasks {
 		unittest.AssertNotExistsBean(t, hookTask)
 	}
 	// this test also ensures that * doesn't handle / in any special way (like shell would)
-	assert.NoError(t, PrepareWebhooks(repo, webhook_model.HookEventPush, &api.PushPayload{Ref: "refs/heads/feature/7791", Commits: []*api.PayloadCommit{{}}}))
+	assert.NoError(t, PrepareWebhooks(db.DefaultContext, EventSource{Repository: repo}, webhook_model.HookEventPush, &api.PushPayload{Ref: "refs/heads/feature/7791", Commits: []*api.PayloadCommit{{}}}))
 	for _, hookTask := range hookTasks {
 		unittest.AssertExistsAndLoadBean(t, hookTask)
 	}
@@ -65,12 +66,12 @@ func TestPrepareWebhooksBranchFilterNoMatch(t *testing.T) {
 
 	repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: 2})
 	hookTasks := []*webhook_model.HookTask{
-		{RepoID: repo.ID, HookID: 4, EventType: webhook_model.HookEventPush},
+		{HookID: 4, EventType: webhook_model.HookEventPush},
 	}
 	for _, hookTask := range hookTasks {
 		unittest.AssertNotExistsBean(t, hookTask)
 	}
-	assert.NoError(t, PrepareWebhooks(repo, webhook_model.HookEventPush, &api.PushPayload{Ref: "refs/heads/fix_weird_bug"}))
+	assert.NoError(t, PrepareWebhooks(db.DefaultContext, EventSource{Repository: repo}, webhook_model.HookEventPush, &api.PushPayload{Ref: "refs/heads/fix_weird_bug"}))
 
 	for _, hookTask := range hookTasks {
 		unittest.AssertNotExistsBean(t, hookTask)