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>

models/fixtures/hook_task.yml

@@ -1,6 +1,5 @@
 -
   id: 1
-  repo_id: 1
   hook_id: 1
   uuid: uuid1
   is_delivered: true

models/repo.go

@@ -123,6 +123,11 @@ func DeleteRepository(doer *user_model.User, uid, repoID int64) error {
 		return err
 	}
 
+	if _, err := db.GetEngine(ctx).In("hook_id", builder.Select("id").From("webhook").Where(builder.Eq{"webhook.repo_id": repo.ID})).
+		Delete(&webhook.HookTask{}); err != nil {
+		return err
+	}
+
 	if err := db.DeleteBeans(ctx,
 		&access_model.Access{RepoID: repo.ID},
 		&activities_model.Action{RepoID: repo.ID},
@@ -130,7 +135,6 @@ func DeleteRepository(doer *user_model.User, uid, repoID int64) error {
 		&issues_model.Comment{RefRepoID: repoID},
 		&git_model.CommitStatus{RepoID: repoID},
 		&git_model.DeletedBranch{RepoID: repoID},
-		&webhook.HookTask{RepoID: repoID},
 		&git_model.LFSLock{RepoID: repoID},
 		&repo_model.LanguageStat{RepoID: repoID},
 		&issues_model.Milestone{RepoID: repoID},

models/webhook/hooktask.go

@@ -104,7 +104,6 @@ type HookResponse struct {
 // HookTask represents a hook task.
 type HookTask struct {
 	ID              int64 `xorm:"pk autoincr"`
-	RepoID          int64 `xorm:"INDEX"`
 	HookID          int64
 	UUID            string
 	api.Payloader   `xorm:"-"`
@@ -178,14 +177,29 @@ func HookTasks(hookID int64, page int) ([]*HookTask, error) {
 
 // CreateHookTask creates a new hook task,
 // it handles conversion from Payload to PayloadContent.
-func CreateHookTask(t *HookTask) error {
+func CreateHookTask(ctx context.Context, t *HookTask) (*HookTask, error) {
 	data, err := t.Payloader.JSONPayload()
 	if err != nil {
-		return err
+		return nil, err
 	}
 	t.UUID = gouuid.New().String()
 	t.PayloadContent = string(data)
-	return db.Insert(db.DefaultContext, t)
+	return t, db.Insert(ctx, t)
+}
+
+func GetHookTaskByID(ctx context.Context, id int64) (*HookTask, error) {
+	t := &HookTask{}
+
+	has, err := db.GetEngine(ctx).ID(id).Get(t)
+	if err != nil {
+		return nil, err
+	}
+	if !has {
+		return nil, ErrHookTaskNotExist{
+			TaskID: id,
+		}
+	}
+	return t, nil
 }
 
 // UpdateHookTask updates information of hook task.
@@ -195,53 +209,36 @@ func UpdateHookTask(t *HookTask) error {
 }
 
 // ReplayHookTask copies a hook task to get re-delivered
-func ReplayHookTask(hookID int64, uuid string) (*HookTask, error) {
+func ReplayHookTask(ctx context.Context, hookID int64, uuid string) (*HookTask, error) {
-	var newTask *HookTask
+	task := &HookTask{
-
+		HookID: hookID,
-	err := db.WithTx(func(ctx context.Context) error {
+		UUID:   uuid,
-		task := &HookTask{
+	}
+	has, err := db.GetByBean(ctx, task)
+	if err != nil {
+		return nil, err
+	} else if !has {
+		return nil, ErrHookTaskNotExist{
 			HookID: hookID,
 			UUID:   uuid,
 		}
-		has, err := db.GetByBean(ctx, task)
+	}
-		if err != nil {
-			return err
-		} else if !has {
-			return ErrHookTaskNotExist{
-				HookID: hookID,
-				UUID:   uuid,
-			}
-		}
-
-		newTask = &HookTask{
-			UUID:           gouuid.New().String(),
-			RepoID:         task.RepoID,
-			HookID:         task.HookID,
-			PayloadContent: task.PayloadContent,
-			EventType:      task.EventType,
-		}
-		return db.Insert(ctx, newTask)
-	})
 
-	return newTask, err
+	newTask := &HookTask{
+		UUID:           gouuid.New().String(),
+		HookID:         task.HookID,
+		PayloadContent: task.PayloadContent,
+		EventType:      task.EventType,
+	}
+	return newTask, db.Insert(ctx, newTask)
 }
 
 // FindUndeliveredHookTasks represents find the undelivered hook tasks
-func FindUndeliveredHookTasks() ([]*HookTask, error) {
+func FindUndeliveredHookTasks(ctx context.Context) ([]*HookTask, error) {
 	tasks := make([]*HookTask, 0, 10)
-	if err := db.GetEngine(db.DefaultContext).Where("is_delivered=?", false).Find(&tasks); err != nil {
+	return tasks, db.GetEngine(ctx).
-		return nil, err
+		Where("is_delivered=?", false).
-	}
+		Find(&tasks)
-	return tasks, nil
-}
-
-// FindRepoUndeliveredHookTasks represents find the undelivered hook tasks of one repository
-func FindRepoUndeliveredHookTasks(repoID int64) ([]*HookTask, error) {
-	tasks := make([]*HookTask, 0, 5)
-	if err := db.GetEngine(db.DefaultContext).Where("repo_id=? AND is_delivered=?", repoID, false).Find(&tasks); err != nil {
-		return nil, err
-	}
-	return tasks, nil
 }
 
 // CleanupHookTaskTable deletes rows from hook_task as needed.
@@ -250,7 +247,7 @@ func CleanupHookTaskTable(ctx context.Context, cleanupType HookTaskCleanupType,
 
 	if cleanupType == OlderThan {
 		deleteOlderThan := time.Now().Add(-olderThan).UnixNano()
-		deletes, err := db.GetEngine(db.DefaultContext).
+		deletes, err := db.GetEngine(ctx).
 			Where("is_delivered = ? and delivered < ?", true, deleteOlderThan).
 			Delete(new(HookTask))
 		if err != nil {
@@ -259,7 +256,8 @@ func CleanupHookTaskTable(ctx context.Context, cleanupType HookTaskCleanupType,
 		log.Trace("Deleted %d rows from hook_task", deletes)
 	} else if cleanupType == PerWebhook {
 		hookIDs := make([]int64, 0, 10)
-		err := db.GetEngine(db.DefaultContext).Table("webhook").
+		err := db.GetEngine(ctx).
+			Table("webhook").
 			Where("id > 0").
 			Cols("id").
 			Find(&hookIDs)

models/webhook/webhook.go

@@ -19,13 +19,6 @@ import (
 	"xorm.io/builder"
 )
 
-//  __      __      ___.   .__                   __
-// /  \    /  \ ____\_ |__ |  |__   ____   ____ |  | __
-// \   \/\/   // __ \| __ \|  |  \ /  _ \ /  _ \|  |/ /
-//  \        /\  ___/| \_\ \   Y  (  <_> |  <_> )    <
-//   \__/\  /  \___  >___  /___|  /\____/ \____/|__|_ \
-//        \/       \/    \/     \/                   \/
-
 // ErrWebhookNotExist represents a "WebhookNotExist" kind of error.
 type ErrWebhookNotExist struct {
 	ID int64
@@ -47,6 +40,7 @@ func (err ErrWebhookNotExist) Unwrap() error {
 
 // ErrHookTaskNotExist represents a "HookTaskNotExist" kind of error.
 type ErrHookTaskNotExist struct {
+	TaskID int64
 	HookID int64
 	UUID   string
 }
@@ -58,7 +52,7 @@ func IsErrHookTaskNotExist(err error) bool {
 }
 
 func (err ErrHookTaskNotExist) Error() string {
-	return fmt.Sprintf("hook task does not exist [hook: %d, uuid: %s]", err.HookID, err.UUID)
+	return fmt.Sprintf("hook task does not exist [task: %d, hook: %d, uuid: %s]", err.TaskID, err.HookID, err.UUID)
 }
 
 func (err ErrHookTaskNotExist) Unwrap() error {

models/webhook/webhook_test.go

@@ -208,12 +208,12 @@ func TestHookTasks(t *testing.T) {
 func TestCreateHookTask(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	hookTask := &HookTask{
-		RepoID:    3,
 		HookID:    3,
 		Payloader: &api.PushPayload{},
 	}
 	unittest.AssertNotExistsBean(t, hookTask)
-	assert.NoError(t, CreateHookTask(hookTask))
+	_, err := CreateHookTask(db.DefaultContext, hookTask)
+	assert.NoError(t, err)
 	unittest.AssertExistsAndLoadBean(t, hookTask)
 }
 
@@ -232,14 +232,14 @@ func TestUpdateHookTask(t *testing.T) {
 func TestCleanupHookTaskTable_PerWebhook_DeletesDelivered(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	hookTask := &HookTask{
-		RepoID:      3,
 		HookID:      3,
 		Payloader:   &api.PushPayload{},
 		IsDelivered: true,
 		Delivered:   time.Now().UnixNano(),
 	}
 	unittest.AssertNotExistsBean(t, hookTask)
-	assert.NoError(t, CreateHookTask(hookTask))
+	_, err := CreateHookTask(db.DefaultContext, hookTask)
+	assert.NoError(t, err)
 	unittest.AssertExistsAndLoadBean(t, hookTask)
 
 	assert.NoError(t, CleanupHookTaskTable(context.Background(), PerWebhook, 168*time.Hour, 0))
@@ -249,13 +249,13 @@ func TestCleanupHookTaskTable_PerWebhook_DeletesDelivered(t *testing.T) {
 func TestCleanupHookTaskTable_PerWebhook_LeavesUndelivered(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	hookTask := &HookTask{
-		RepoID:      2,
 		HookID:      4,
 		Payloader:   &api.PushPayload{},
 		IsDelivered: false,
 	}
 	unittest.AssertNotExistsBean(t, hookTask)
-	assert.NoError(t, CreateHookTask(hookTask))
+	_, err := CreateHookTask(db.DefaultContext, hookTask)
+	assert.NoError(t, err)
 	unittest.AssertExistsAndLoadBean(t, hookTask)
 
 	assert.NoError(t, CleanupHookTaskTable(context.Background(), PerWebhook, 168*time.Hour, 0))
@@ -265,14 +265,14 @@ func TestCleanupHookTaskTable_PerWebhook_LeavesUndelivered(t *testing.T) {
 func TestCleanupHookTaskTable_PerWebhook_LeavesMostRecentTask(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	hookTask := &HookTask{
-		RepoID:      2,
 		HookID:      4,
 		Payloader:   &api.PushPayload{},
 		IsDelivered: true,
 		Delivered:   time.Now().UnixNano(),
 	}
 	unittest.AssertNotExistsBean(t, hookTask)
-	assert.NoError(t, CreateHookTask(hookTask))
+	_, err := CreateHookTask(db.DefaultContext, hookTask)
+	assert.NoError(t, err)
 	unittest.AssertExistsAndLoadBean(t, hookTask)
 
 	assert.NoError(t, CleanupHookTaskTable(context.Background(), PerWebhook, 168*time.Hour, 1))
@@ -282,14 +282,14 @@ func TestCleanupHookTaskTable_PerWebhook_LeavesMostRecentTask(t *testing.T) {
 func TestCleanupHookTaskTable_OlderThan_DeletesDelivered(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	hookTask := &HookTask{
-		RepoID:      3,
 		HookID:      3,
 		Payloader:   &api.PushPayload{},
 		IsDelivered: true,
 		Delivered:   time.Now().AddDate(0, 0, -8).UnixNano(),
 	}
 	unittest.AssertNotExistsBean(t, hookTask)
-	assert.NoError(t, CreateHookTask(hookTask))
+	_, err := CreateHookTask(db.DefaultContext, hookTask)
+	assert.NoError(t, err)
 	unittest.AssertExistsAndLoadBean(t, hookTask)
 
 	assert.NoError(t, CleanupHookTaskTable(context.Background(), OlderThan, 168*time.Hour, 0))
@@ -299,13 +299,13 @@ func TestCleanupHookTaskTable_OlderThan_DeletesDelivered(t *testing.T) {
 func TestCleanupHookTaskTable_OlderThan_LeavesUndelivered(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	hookTask := &HookTask{
-		RepoID:      2,
 		HookID:      4,
 		Payloader:   &api.PushPayload{},
 		IsDelivered: false,
 	}
 	unittest.AssertNotExistsBean(t, hookTask)
-	assert.NoError(t, CreateHookTask(hookTask))
+	_, err := CreateHookTask(db.DefaultContext, hookTask)
+	assert.NoError(t, err)
 	unittest.AssertExistsAndLoadBean(t, hookTask)
 
 	assert.NoError(t, CleanupHookTaskTable(context.Background(), OlderThan, 168*time.Hour, 0))
@@ -315,14 +315,14 @@ func TestCleanupHookTaskTable_OlderThan_LeavesUndelivered(t *testing.T) {
 func TestCleanupHookTaskTable_OlderThan_LeavesTaskEarlierThanAgeToDelete(t *testing.T) {
 	assert.NoError(t, unittest.PrepareTestDatabase())
 	hookTask := &HookTask{
-		RepoID:      2,
 		HookID:      4,
 		Payloader:   &api.PushPayload{},
 		IsDelivered: true,
 		Delivered:   time.Now().AddDate(0, 0, -6).UnixNano(),
 	}
 	unittest.AssertNotExistsBean(t, hookTask)
-	assert.NoError(t, CreateHookTask(hookTask))
+	_, err := CreateHookTask(db.DefaultContext, hookTask)
+	assert.NoError(t, err)
 	unittest.AssertExistsAndLoadBean(t, hookTask)
 
 	assert.NoError(t, CleanupHookTaskTable(context.Background(), OlderThan, 168*time.Hour, 0))

modules/notification/webhook/webhook.go

@@ -61,7 +61,7 @@ func (m *webhookNotifier) NotifyIssueClearLabels(doer *user_model.User, issue *i
 			return
 		}
 
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequestLabel, &api.PullRequestPayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequestLabel, &api.PullRequestPayload{
 			Action:      api.HookIssueLabelCleared,
 			Index:       issue.Index,
 			PullRequest: convert.ToAPIPullRequest(ctx, issue.PullRequest, nil),
@@ -69,7 +69,7 @@ func (m *webhookNotifier) NotifyIssueClearLabels(doer *user_model.User, issue *i
 			Sender:      convert.ToUser(doer, nil),
 		})
 	} else {
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssueLabel, &api.IssuePayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssueLabel, &api.IssuePayload{
 			Action:     api.HookIssueLabelCleared,
 			Index:      issue.Index,
 			Issue:      convert.ToAPIIssue(issue),
@@ -87,7 +87,7 @@ func (m *webhookNotifier) NotifyForkRepository(doer *user_model.User, oldRepo, r
 	mode, _ := access_model.AccessLevel(doer, repo)
 
 	// forked webhook
-	if err := webhook_services.PrepareWebhooks(oldRepo, webhook.HookEventFork, &api.ForkPayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: oldRepo}, webhook.HookEventFork, &api.ForkPayload{
 		Forkee: convert.ToRepo(oldRepo, oldMode),
 		Repo:   convert.ToRepo(repo, mode),
 		Sender: convert.ToUser(doer, nil),
@@ -99,7 +99,7 @@ func (m *webhookNotifier) NotifyForkRepository(doer *user_model.User, oldRepo, r
 
 	// Add to hook queue for created repo after session commit.
 	if u.IsOrganization() {
-		if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventRepository, &api.RepositoryPayload{
+		if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventRepository, &api.RepositoryPayload{
 			Action:       api.HookRepoCreated,
 			Repository:   convert.ToRepo(repo, perm.AccessModeOwner),
 			Organization: convert.ToUser(u, nil),
@@ -112,7 +112,7 @@ func (m *webhookNotifier) NotifyForkRepository(doer *user_model.User, oldRepo, r
 
 func (m *webhookNotifier) NotifyCreateRepository(doer, u *user_model.User, repo *repo_model.Repository) {
 	// Add to hook queue for created repo after session commit.
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventRepository, &api.RepositoryPayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventRepository, &api.RepositoryPayload{
 		Action:       api.HookRepoCreated,
 		Repository:   convert.ToRepo(repo, perm.AccessModeOwner),
 		Organization: convert.ToUser(u, nil),
@@ -125,7 +125,7 @@ func (m *webhookNotifier) NotifyCreateRepository(doer, u *user_model.User, repo
 func (m *webhookNotifier) NotifyDeleteRepository(doer *user_model.User, repo *repo_model.Repository) {
 	u := repo.MustOwner()
 
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventRepository, &api.RepositoryPayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventRepository, &api.RepositoryPayload{
 		Action:       api.HookRepoDeleted,
 		Repository:   convert.ToRepo(repo, perm.AccessModeOwner),
 		Organization: convert.ToUser(u, nil),
@@ -137,7 +137,7 @@ func (m *webhookNotifier) NotifyDeleteRepository(doer *user_model.User, repo *re
 
 func (m *webhookNotifier) NotifyMigrateRepository(doer, u *user_model.User, repo *repo_model.Repository) {
 	// Add to hook queue for created repo after session commit.
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventRepository, &api.RepositoryPayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventRepository, &api.RepositoryPayload{
 		Action:       api.HookRepoCreated,
 		Repository:   convert.ToRepo(repo, perm.AccessModeOwner),
 		Organization: convert.ToUser(u, nil),
@@ -171,7 +171,7 @@ func (m *webhookNotifier) NotifyIssueChangeAssignee(doer *user_model.User, issue
 			apiPullRequest.Action = api.HookIssueAssigned
 		}
 		// Assignee comment triggers a webhook
-		if err := webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequestAssign, apiPullRequest); err != nil {
+		if err := webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequestAssign, apiPullRequest); err != nil {
 			log.Error("PrepareWebhooks [is_pull: %v, remove_assignee: %v]: %v", issue.IsPull, removed, err)
 			return
 		}
@@ -189,7 +189,7 @@ func (m *webhookNotifier) NotifyIssueChangeAssignee(doer *user_model.User, issue
 			apiIssue.Action = api.HookIssueAssigned
 		}
 		// Assignee comment triggers a webhook
-		if err := webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssueAssign, apiIssue); err != nil {
+		if err := webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssueAssign, apiIssue); err != nil {
 			log.Error("PrepareWebhooks [is_pull: %v, remove_assignee: %v]: %v", issue.IsPull, removed, err)
 			return
 		}
@@ -208,7 +208,7 @@ func (m *webhookNotifier) NotifyIssueChangeTitle(doer *user_model.User, issue *i
 			return
 		}
 		issue.PullRequest.Issue = issue
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequest, &api.PullRequestPayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequest, &api.PullRequestPayload{
 			Action: api.HookIssueEdited,
 			Index:  issue.Index,
 			Changes: &api.ChangesPayload{
@@ -221,7 +221,7 @@ func (m *webhookNotifier) NotifyIssueChangeTitle(doer *user_model.User, issue *i
 			Sender:      convert.ToUser(doer, nil),
 		})
 	} else {
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssues, &api.IssuePayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssues, &api.IssuePayload{
 			Action: api.HookIssueEdited,
 			Index:  issue.Index,
 			Changes: &api.ChangesPayload{
@@ -263,7 +263,7 @@ func (m *webhookNotifier) NotifyIssueChangeStatus(doer *user_model.User, issue *
 		} else {
 			apiPullRequest.Action = api.HookIssueReOpened
 		}
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequest, apiPullRequest)
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequest, apiPullRequest)
 	} else {
 		apiIssue := &api.IssuePayload{
 			Index:      issue.Index,
@@ -276,7 +276,7 @@ func (m *webhookNotifier) NotifyIssueChangeStatus(doer *user_model.User, issue *
 		} else {
 			apiIssue.Action = api.HookIssueReOpened
 		}
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssues, apiIssue)
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssues, apiIssue)
 	}
 	if err != nil {
 		log.Error("PrepareWebhooks [is_pull: %v, is_closed: %v]: %v", issue.IsPull, isClosed, err)
@@ -294,7 +294,7 @@ func (m *webhookNotifier) NotifyNewIssue(issue *issues_model.Issue, mentions []*
 	}
 
 	mode, _ := access_model.AccessLevel(issue.Poster, issue.Repo)
-	if err := webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssues, &api.IssuePayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssues, &api.IssuePayload{
 		Action:     api.HookIssueOpened,
 		Index:      issue.Index,
 		Issue:      convert.ToAPIIssue(issue),
@@ -323,7 +323,7 @@ func (m *webhookNotifier) NotifyNewPullRequest(pull *issues_model.PullRequest, m
 	}
 
 	mode, _ := access_model.AccessLevel(pull.Issue.Poster, pull.Issue.Repo)
-	if err := webhook_services.PrepareWebhooks(pull.Issue.Repo, webhook.HookEventPullRequest, &api.PullRequestPayload{
+	if err := webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: pull.Issue.Repo}, webhook.HookEventPullRequest, &api.PullRequestPayload{
 		Action:      api.HookIssueOpened,
 		Index:       pull.Issue.Index,
 		PullRequest: convert.ToAPIPullRequest(ctx, pull, nil),
@@ -342,7 +342,7 @@ func (m *webhookNotifier) NotifyIssueChangeContent(doer *user_model.User, issue
 	var err error
 	if issue.IsPull {
 		issue.PullRequest.Issue = issue
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequest, &api.PullRequestPayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequest, &api.PullRequestPayload{
 			Action: api.HookIssueEdited,
 			Index:  issue.Index,
 			Changes: &api.ChangesPayload{
@@ -355,7 +355,7 @@ func (m *webhookNotifier) NotifyIssueChangeContent(doer *user_model.User, issue
 			Sender:      convert.ToUser(doer, nil),
 		})
 	} else {
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssues, &api.IssuePayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssues, &api.IssuePayload{
 			Action: api.HookIssueEdited,
 			Index:  issue.Index,
 			Changes: &api.ChangesPayload{
@@ -374,54 +374,41 @@ func (m *webhookNotifier) NotifyIssueChangeContent(doer *user_model.User, issue
 }
 
 func (m *webhookNotifier) NotifyUpdateComment(doer *user_model.User, c *issues_model.Comment, oldContent string) {
-	var err error
+	if err := c.LoadPoster(); err != nil {
-
-	if err = c.LoadPoster(); err != nil {
 		log.Error("LoadPoster: %v", err)
 		return
 	}
-	if err = c.LoadIssue(); err != nil {
+	if err := c.LoadIssue(); err != nil {
 		log.Error("LoadIssue: %v", err)
 		return
 	}
 
-	if err = c.Issue.LoadAttributes(db.DefaultContext); err != nil {
+	if err := c.Issue.LoadAttributes(db.DefaultContext); err != nil {
 		log.Error("LoadAttributes: %v", err)
 		return
 	}
 
-	mode, _ := access_model.AccessLevel(doer, c.Issue.Repo)
+	var eventType webhook.HookEventType
 	if c.Issue.IsPull {
-		err = webhook_services.PrepareWebhooks(c.Issue.Repo, webhook.HookEventPullRequestComment, &api.IssueCommentPayload{
+		eventType = webhook.HookEventPullRequestComment
-			Action:  api.HookIssueCommentEdited,
-			Issue:   convert.ToAPIIssue(c.Issue),
-			Comment: convert.ToComment(c),
-			Changes: &api.ChangesPayload{
-				Body: &api.ChangesFromPayload{
-					From: oldContent,
-				},
-			},
-			Repository: convert.ToRepo(c.Issue.Repo, mode),
-			Sender:     convert.ToUser(doer, nil),
-			IsPull:     true,
-		})
 	} else {
-		err = webhook_services.PrepareWebhooks(c.Issue.Repo, webhook.HookEventIssueComment, &api.IssueCommentPayload{
+		eventType = webhook.HookEventIssueComment
-			Action:  api.HookIssueCommentEdited,
-			Issue:   convert.ToAPIIssue(c.Issue),
-			Comment: convert.ToComment(c),
-			Changes: &api.ChangesPayload{
-				Body: &api.ChangesFromPayload{
-					From: oldContent,
-				},
-			},
-			Repository: convert.ToRepo(c.Issue.Repo, mode),
-			Sender:     convert.ToUser(doer, nil),
-			IsPull:     false,
-		})
 	}
 
-	if err != nil {
+	mode, _ := access_model.AccessLevel(doer, c.Issue.Repo)
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: c.Issue.Repo}, eventType, &api.IssueCommentPayload{
+		Action:  api.HookIssueCommentEdited,
+		Issue:   convert.ToAPIIssue(c.Issue),
+		Comment: convert.ToComment(c),
+		Changes: &api.ChangesPayload{
+			Body: &api.ChangesFromPayload{
+				From: oldContent,
+			},
+		},
+		Repository: convert.ToRepo(c.Issue.Repo, mode),
+		Sender:     convert.ToUser(doer, nil),
+		IsPull:     c.Issue.IsPull,
+	}); err != nil {
 		log.Error("PrepareWebhooks [comment_id: %d]: %v", c.ID, err)
 	}
 }
@@ -429,30 +416,22 @@ func (m *webhookNotifier) NotifyUpdateComment(doer *user_model.User, c *issues_m
 func (m *webhookNotifier) NotifyCreateIssueComment(doer *user_model.User, repo *repo_model.Repository,
 	issue *issues_model.Issue, comment *issues_model.Comment, mentions []*user_model.User,
 ) {
-	mode, _ := access_model.AccessLevel(doer, repo)
+	var eventType webhook.HookEventType
-
-	var err error
 	if issue.IsPull {
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequestComment, &api.IssueCommentPayload{
+		eventType = webhook.HookEventPullRequestComment
-			Action:     api.HookIssueCommentCreated,
-			Issue:      convert.ToAPIIssue(issue),
-			Comment:    convert.ToComment(comment),
-			Repository: convert.ToRepo(repo, mode),
-			Sender:     convert.ToUser(doer, nil),
-			IsPull:     true,
-		})
 	} else {
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssueComment, &api.IssueCommentPayload{
+		eventType = webhook.HookEventIssueComment
-			Action:     api.HookIssueCommentCreated,
-			Issue:      convert.ToAPIIssue(issue),
-			Comment:    convert.ToComment(comment),
-			Repository: convert.ToRepo(repo, mode),
-			Sender:     convert.ToUser(doer, nil),
-			IsPull:     false,
-		})
 	}
 
-	if err != nil {
+	mode, _ := access_model.AccessLevel(doer, repo)
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: issue.Repo}, eventType, &api.IssueCommentPayload{
+		Action:     api.HookIssueCommentCreated,
+		Issue:      convert.ToAPIIssue(issue),
+		Comment:    convert.ToComment(comment),
+		Repository: convert.ToRepo(repo, mode),
+		Sender:     convert.ToUser(doer, nil),
+		IsPull:     issue.IsPull,
+	}); err != nil {
 		log.Error("PrepareWebhooks [comment_id: %d]: %v", comment.ID, err)
 	}
 }
@@ -474,36 +453,29 @@ func (m *webhookNotifier) NotifyDeleteComment(doer *user_model.User, comment *is
 		return
 	}
 
-	mode, _ := access_model.AccessLevel(doer, comment.Issue.Repo)
+	var eventType webhook.HookEventType
-
 	if comment.Issue.IsPull {
-		err = webhook_services.PrepareWebhooks(comment.Issue.Repo, webhook.HookEventPullRequestComment, &api.IssueCommentPayload{
+		eventType = webhook.HookEventPullRequestComment
-			Action:     api.HookIssueCommentDeleted,
-			Issue:      convert.ToAPIIssue(comment.Issue),
-			Comment:    convert.ToComment(comment),
-			Repository: convert.ToRepo(comment.Issue.Repo, mode),
-			Sender:     convert.ToUser(doer, nil),
-			IsPull:     true,
-		})
 	} else {
-		err = webhook_services.PrepareWebhooks(comment.Issue.Repo, webhook.HookEventIssueComment, &api.IssueCommentPayload{
+		eventType = webhook.HookEventIssueComment
-			Action:     api.HookIssueCommentDeleted,
-			Issue:      convert.ToAPIIssue(comment.Issue),
-			Comment:    convert.ToComment(comment),
-			Repository: convert.ToRepo(comment.Issue.Repo, mode),
-			Sender:     convert.ToUser(doer, nil),
-			IsPull:     false,
-		})
 	}
 
-	if err != nil {
+	mode, _ := access_model.AccessLevel(doer, comment.Issue.Repo)
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: comment.Issue.Repo}, eventType, &api.IssueCommentPayload{
+		Action:     api.HookIssueCommentDeleted,
+		Issue:      convert.ToAPIIssue(comment.Issue),
+		Comment:    convert.ToComment(comment),
+		Repository: convert.ToRepo(comment.Issue.Repo, mode),
+		Sender:     convert.ToUser(doer, nil),
+		IsPull:     comment.Issue.IsPull,
+	}); err != nil {
 		log.Error("PrepareWebhooks [comment_id: %d]: %v", comment.ID, err)
 	}
 }
 
 func (m *webhookNotifier) NotifyNewWikiPage(doer *user_model.User, repo *repo_model.Repository, page, comment string) {
 	// Add to hook queue for created wiki page.
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventWiki, &api.WikiPayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventWiki, &api.WikiPayload{
 		Action:     api.HookWikiCreated,
 		Repository: convert.ToRepo(repo, perm.AccessModeOwner),
 		Sender:     convert.ToUser(doer, nil),
@@ -516,7 +488,7 @@ func (m *webhookNotifier) NotifyNewWikiPage(doer *user_model.User, repo *repo_mo
 
 func (m *webhookNotifier) NotifyEditWikiPage(doer *user_model.User, repo *repo_model.Repository, page, comment string) {
 	// Add to hook queue for edit wiki page.
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventWiki, &api.WikiPayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventWiki, &api.WikiPayload{
 		Action:     api.HookWikiEdited,
 		Repository: convert.ToRepo(repo, perm.AccessModeOwner),
 		Sender:     convert.ToUser(doer, nil),
@@ -529,7 +501,7 @@ func (m *webhookNotifier) NotifyEditWikiPage(doer *user_model.User, repo *repo_m
 
 func (m *webhookNotifier) NotifyDeleteWikiPage(doer *user_model.User, repo *repo_model.Repository, page string) {
 	// Add to hook queue for edit wiki page.
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventWiki, &api.WikiPayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventWiki, &api.WikiPayload{
 		Action:     api.HookWikiDeleted,
 		Repository: convert.ToRepo(repo, perm.AccessModeOwner),
 		Sender:     convert.ToUser(doer, nil),
@@ -567,7 +539,7 @@ func (m *webhookNotifier) NotifyIssueChangeLabels(doer *user_model.User, issue *
 			log.Error("LoadIssue: %v", err)
 			return
 		}
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequestLabel, &api.PullRequestPayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequestLabel, &api.PullRequestPayload{
 			Action:      api.HookIssueLabelUpdated,
 			Index:       issue.Index,
 			PullRequest: convert.ToAPIPullRequest(ctx, issue.PullRequest, nil),
@@ -575,7 +547,7 @@ func (m *webhookNotifier) NotifyIssueChangeLabels(doer *user_model.User, issue *
 			Sender:      convert.ToUser(doer, nil),
 		})
 	} else {
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssueLabel, &api.IssuePayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssueLabel, &api.IssuePayload{
 			Action:     api.HookIssueLabelUpdated,
 			Index:      issue.Index,
 			Issue:      convert.ToAPIIssue(issue),
@@ -612,7 +584,7 @@ func (m *webhookNotifier) NotifyIssueChangeMilestone(doer *user_model.User, issu
 			log.Error("LoadIssue: %v", err)
 			return
 		}
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequestMilestone, &api.PullRequestPayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequestMilestone, &api.PullRequestPayload{
 			Action:      hookAction,
 			Index:       issue.Index,
 			PullRequest: convert.ToAPIPullRequest(ctx, issue.PullRequest, nil),
@@ -620,7 +592,7 @@ func (m *webhookNotifier) NotifyIssueChangeMilestone(doer *user_model.User, issu
 			Sender:      convert.ToUser(doer, nil),
 		})
 	} else {
-		err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssueMilestone, &api.IssuePayload{
+		err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventIssueMilestone, &api.IssuePayload{
 			Action:     hookAction,
 			Index:      issue.Index,
 			Issue:      convert.ToAPIIssue(issue),
@@ -644,7 +616,7 @@ func (m *webhookNotifier) NotifyPushCommits(pusher *user_model.User, repo *repo_
 		return
 	}
 
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventPush, &api.PushPayload{
+	if err := webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: repo}, webhook.HookEventPush, &api.PushPayload{
 		Ref:          opts.RefFullName,
 		Before:       opts.OldCommitID,
 		After:        opts.NewCommitID,
@@ -695,7 +667,7 @@ func (*webhookNotifier) NotifyMergePullRequest(pr *issues_model.PullRequest, doe
 		Action:      api.HookIssueClosed,
 	}
 
-	err = webhook_services.PrepareWebhooks(pr.Issue.Repo, webhook.HookEventPullRequest, apiPullRequest)
+	err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: pr.Issue.Repo}, webhook.HookEventPullRequest, apiPullRequest)
 	if err != nil {
 		log.Error("PrepareWebhooks: %v", err)
 	}
@@ -717,7 +689,7 @@ func (m *webhookNotifier) NotifyPullRequestChangeTargetBranch(doer *user_model.U
 	}
 	issue.PullRequest.Issue = issue
 	mode, _ := access_model.AccessLevel(issue.Poster, issue.Repo)
-	err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequest, &api.PullRequestPayload{
+	err = webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: issue.Repo}, webhook.HookEventPullRequest, &api.PullRequestPayload{
 		Action: api.HookIssueEdited,
 		Index:  issue.Index,
 		Changes: &api.ChangesPayload{
@@ -764,7 +736,7 @@ func (m *webhookNotifier) NotifyPullRequestReview(pr *issues_model.PullRequest,
 		log.Error("models.AccessLevel: %v", err)
 		return
 	}
-	if err := webhook_services.PrepareWebhooks(review.Issue.Repo, reviewHookType, &api.PullRequestPayload{
+	if err := webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: review.Issue.Repo}, reviewHookType, &api.PullRequestPayload{
 		Action:      api.HookIssueReviewed,
 		Index:       review.Issue.Index,
 		PullRequest: convert.ToAPIPullRequest(ctx, pr, nil),
@@ -784,7 +756,7 @@ func (m *webhookNotifier) NotifyCreateRef(pusher *user_model.User, repo *repo_mo
 	apiRepo := convert.ToRepo(repo, perm.AccessModeNone)
 	refName := git.RefEndName(refFullName)
 
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventCreate, &api.CreatePayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventCreate, &api.CreatePayload{
 		Ref:     refName,
 		Sha:     refID,
 		RefType: refType,
@@ -808,7 +780,7 @@ func (m *webhookNotifier) NotifyPullRequestSynchronized(doer *user_model.User, p
 		return
 	}
 
-	if err := webhook_services.PrepareWebhooks(pr.Issue.Repo, webhook.HookEventPullRequestSync, &api.PullRequestPayload{
+	if err := webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: pr.Issue.Repo}, webhook.HookEventPullRequestSync, &api.PullRequestPayload{
 		Action:      api.HookIssueSynchronized,
 		Index:       pr.Issue.Index,
 		PullRequest: convert.ToAPIPullRequest(ctx, pr, nil),
@@ -824,7 +796,7 @@ func (m *webhookNotifier) NotifyDeleteRef(pusher *user_model.User, repo *repo_mo
 	apiRepo := convert.ToRepo(repo, perm.AccessModeNone)
 	refName := git.RefEndName(refFullName)
 
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventDelete, &api.DeletePayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: repo}, webhook.HookEventDelete, &api.DeletePayload{
 		Ref:        refName,
 		RefType:    refType,
 		PusherType: api.PusherTypeUser,
@@ -842,7 +814,7 @@ func sendReleaseHook(doer *user_model.User, rel *repo_model.Release, action api.
 	}
 
 	mode, _ := access_model.AccessLevel(doer, rel.Repo)
-	if err := webhook_services.PrepareWebhooks(rel.Repo, webhook.HookEventRelease, &api.ReleasePayload{
+	if err := webhook_services.PrepareWebhooks(db.DefaultContext, webhook_services.EventSource{Repository: rel.Repo}, webhook.HookEventRelease, &api.ReleasePayload{
 		Action:     action,
 		Release:    convert.ToRelease(rel),
 		Repository: convert.ToRepo(rel.Repo, mode),
@@ -875,7 +847,7 @@ func (m *webhookNotifier) NotifySyncPushCommits(pusher *user_model.User, repo *r
 		return
 	}
 
-	if err := webhook_services.PrepareWebhooks(repo, webhook.HookEventPush, &api.PushPayload{
+	if err := webhook_services.PrepareWebhooks(ctx, webhook_services.EventSource{Repository: repo}, webhook.HookEventPush, &api.PushPayload{
 		Ref:          opts.RefFullName,
 		Before:       opts.OldCommitID,
 		After:        opts.NewCommitID,
@@ -908,9 +880,9 @@ func (m *webhookNotifier) NotifyPackageDelete(doer *user_model.User, pd *package
 }
 
 func notifyPackage(sender *user_model.User, pd *packages_model.PackageDescriptor, action api.HookPackageAction) {
-	if pd.Repository == nil {
+	source := webhook_services.EventSource{
-		// TODO https://github.com/go-gitea/gitea/pull/17940
+		Repository: pd.Repository,
-		return
+		Owner:      pd.Owner,
 	}
 
 	ctx, _, finished := process.GetManager().AddContext(graceful.GetManager().HammerContext(), fmt.Sprintf("webhook.notifyPackage Package: %s[%d]", pd.Package.Name, pd.Package.ID))
@@ -922,7 +894,7 @@ func notifyPackage(sender *user_model.User, pd *packages_model.PackageDescriptor
 		return
 	}
 
-	if err := webhook_services.PrepareWebhooks(pd.Repository, webhook.HookEventPackage, &api.PackagePayload{
+	if err := webhook_services.PrepareWebhooks(ctx, source, webhook.HookEventPackage, &api.PackagePayload{
 		Action:  action,
 		Package: apiPackage,
 		Sender:  convert.ToUser(sender, nil),

routers/api/v1/repo/hook.go

@@ -168,7 +168,7 @@ func TestHook(ctx *context.APIContext) {
 	commit := convert.ToPayloadCommit(ctx.Repo.Repository, ctx.Repo.Commit)
 
 	commitID := ctx.Repo.Commit.ID.String()
-	if err := webhook_service.PrepareWebhook(hook, ctx.Repo.Repository, webhook.HookEventPush, &api.PushPayload{
+	if err := webhook_service.PrepareWebhook(ctx, hook, webhook.HookEventPush, &api.PushPayload{
 		Ref:          ref,
 		Before:       commitID,
 		After:        commitID,

routers/api/v1/repo/hook_test.go

@@ -28,7 +28,6 @@ func TestTestHook(t *testing.T) {
 	assert.EqualValues(t, http.StatusNoContent, ctx.Resp.Status())
 
 	unittest.AssertExistsAndLoadBean(t, &webhook.HookTask{
-		RepoID: 1,
 		HookID: 1,
 	}, unittest.Cond("is_delivered=?", false))
 }

routers/web/repo/webhook.go

@@ -633,7 +633,7 @@ func TestWebhook(ctx *context.Context) {
 	hookID := ctx.ParamsInt64(":id")
 	w, err := webhook.GetWebhookByRepoID(ctx.Repo.Repository.ID, hookID)
 	if err != nil {
-		ctx.Flash.Error("GetWebhookByID: " + err.Error())
+		ctx.Flash.Error("GetWebhookByRepoID: " + err.Error())
 		ctx.Status(http.StatusInternalServerError)
 		return
 	}
@@ -679,7 +679,7 @@ func TestWebhook(ctx *context.Context) {
 		Pusher:       apiUser,
 		Sender:       apiUser,
 	}
-	if err := webhook_service.PrepareWebhook(w, ctx.Repo.Repository, webhook.HookEventPush, p); err != nil {
+	if err := webhook_service.PrepareWebhook(ctx, w, webhook.HookEventPush, p); err != nil {
 		ctx.Flash.Error("PrepareWebhook: " + err.Error())
 		ctx.Status(http.StatusInternalServerError)
 	} else {
@@ -697,7 +697,7 @@ func ReplayWebhook(ctx *context.Context) {
 		return
 	}
 
-	if err := webhook_service.ReplayHookTask(w, hookTaskUUID); err != nil {
+	if err := webhook_service.ReplayHookTask(ctx, w, hookTaskUUID); err != nil {
 		if webhook.IsErrHookTaskNotExist(err) {
 			ctx.NotFound("ReplayHookTask", nil)
 		} else {

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
 }

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
+// EventSource represents the source of a webhook action. Repository and/or Owner must be set.
-func handle(data ...queue.Data) []queue.Data {
+type EventSource struct {
-	for _, datum := range data {
+	Repository *repo_model.Repository
-		repoIDStr := datum.(string)
+	Owner      *user_model.User
-		log.Trace("DeliverHooks [repo_id: %v]", repoIDStr)
+}
 
-		repoID, err := strconv.ParseInt(repoIDStr, 10, 64)
+// handle delivers hook tasks
-		if err != nil {
+func handle(data ...queue.Data) []queue.Data {
-			log.Error("Invalid repo ID: %s", repoIDStr)
+	ctx := graceful.GetManager().HammerContext()
-			continue
-		}
 
-		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)
+			log.Error("GetHookTaskByID failed: %v", err)
-			continue
+		} else {
-		}
+			if err := Deliver(ctx, task); err != nil {
-		for _, t := range tasks {
+				log.Error("webhook.Deliver failed: %v", err)
-			if err = Deliver(graceful.GetManager().HammerContext(), t); err != nil {
-				log.Error("deliver: %v", err)
 			}
 		}
 	}
+
 	return nil
 }
 
-func addToTask(repoID int64) error {
+func enqueueHookTask(task *webhook_model.HookTask) error {
-	err := hookQueue.PushFunc(strconv.FormatInt(repoID, 10), nil)
+	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{
+	task, err := webhook_model.CreateHookTask(ctx, &webhook_model.HookTask{
-		RepoID:    repo.ID,
 		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 {
+func PrepareWebhooks(ctx context.Context, source EventSource, event webhook_model.HookEventType, p api.Payloader) error {
-	if err := prepareWebhooks(db.DefaultContext, repo, event, p); err != nil {
+	owner := source.Owner
-		return err
-	}
 
-	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 {
+	if source.Repository != nil {
-	ws, err := webhook_model.ListWebhooksByOpts(ctx, &webhook_model.ListWebhookOptions{
+		repoHooks, err := webhook_model.ListWebhooksByOpts(ctx, &webhook_model.ListWebhookOptions{
-		RepoID:   repo.ID,
+			RepoID:   source.Repository.ID,
-		IsActive: util.OptionalBoolTrue,
+			IsActive: util.OptionalBoolTrue,
-	})
+		})
-	if err != nil {
+		if err != nil {
-		return fmt.Errorf("GetActiveWebhooksByRepoID: %v", err)
+			return fmt.Errorf("ListWebhooksByOpts: %v", err)
+		}
+		ws = append(ws, repoHooks...)
+
+		owner = source.Repository.MustOwner()
 	}
 
-	// check if repo belongs to org and append additional webhooks
+	// check if owner is an org and append additional webhooks
-	if repo.MustOwner().IsOrganization() {
+	if owner != nil && owner.IsOrganization() {
-		// get hooks for org
 		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 {
+func ReplayHookTask(ctx context.Context, w *webhook_model.Webhook, uuid string) error {
-	t, err := webhook_model.ReplayHookTask(w.ID, uuid)
+	task, err := webhook_model.ReplayHookTask(ctx, w.ID, uuid)
 	if err != nil {
 		return err
 	}
 
-	return addToTask(t.RepoID)
+	return enqueueHookTask(task)
 }

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)