From 1887c9525483957cfc2a1af5bccfd7d03b41f015 Mon Sep 17 00:00:00 2001 From: KN4CK3R Date: Fri, 21 Oct 2022 18:21:56 +0200 Subject: [PATCH] Decouple HookTask from Repository (#17940) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 - models/repo.go | 6 +- models/webhook/hooktask.go | 88 ++++++----- models/webhook/webhook.go | 10 +- models/webhook/webhook_test.go | 28 ++-- modules/notification/webhook/webhook.go | 190 ++++++++++-------------- routers/api/v1/repo/hook.go | 2 +- routers/api/v1/repo/hook_test.go | 1 - routers/web/repo/webhook.go | 6 +- services/webhook/deliver.go | 46 ++---- services/webhook/webhook.go | 110 +++++++------- services/webhook/webhook_test.go | 13 +- 12 files changed, 219 insertions(+), 282 deletions(-) diff --git a/models/fixtures/hook_task.yml b/models/fixtures/hook_task.yml index bb662345cd..6dbb10151a 100644 --- a/models/fixtures/hook_task.yml +++ b/models/fixtures/hook_task.yml @@ -1,6 +1,5 @@ - id: 1 - repo_id: 1 hook_id: 1 uuid: uuid1 is_delivered: true diff --git a/models/repo.go b/models/repo.go index 65159f14af..08fbb0abea 100644 --- a/models/repo.go +++ b/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}, diff --git a/models/webhook/hooktask.go b/models/webhook/hooktask.go index 2adfcaa60d..2b9b63c09b 100644 --- a/models/webhook/hooktask.go +++ b/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) { - var newTask *HookTask - - err := db.WithTx(func(ctx context.Context) error { - task := &HookTask{ +func ReplayHookTask(ctx context.Context, hookID int64, uuid string) (*HookTask, error) { + task := &HookTask{ + HookID: hookID, + UUID: uuid, + } + 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 nil, err - } - 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 + return tasks, db.GetEngine(ctx). + Where("is_delivered=?", false). + Find(&tasks) } // 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) diff --git a/models/webhook/webhook.go b/models/webhook/webhook.go index 83200a3d1c..4551dcff5f 100644 --- a/models/webhook/webhook.go +++ b/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 { diff --git a/models/webhook/webhook_test.go b/models/webhook/webhook_test.go index 7ec7edc0b7..8c4838ebdc 100644 --- a/models/webhook/webhook_test.go +++ b/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)) diff --git a/modules/notification/webhook/webhook.go b/modules/notification/webhook/webhook.go index b93e90368a..630b565984 100644 --- a/modules/notification/webhook/webhook.go +++ b/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{ - 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, - }) + eventType = webhook.HookEventPullRequestComment } else { - err = webhook_services.PrepareWebhooks(c.Issue.Repo, webhook.HookEventIssueComment, &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: false, - }) + eventType = webhook.HookEventIssueComment } - 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 err error + var eventType webhook.HookEventType if issue.IsPull { - err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventPullRequestComment, &api.IssueCommentPayload{ - Action: api.HookIssueCommentCreated, - Issue: convert.ToAPIIssue(issue), - Comment: convert.ToComment(comment), - Repository: convert.ToRepo(repo, mode), - Sender: convert.ToUser(doer, nil), - IsPull: true, - }) + eventType = webhook.HookEventPullRequestComment } else { - err = webhook_services.PrepareWebhooks(issue.Repo, webhook.HookEventIssueComment, &api.IssueCommentPayload{ - Action: api.HookIssueCommentCreated, - Issue: convert.ToAPIIssue(issue), - Comment: convert.ToComment(comment), - Repository: convert.ToRepo(repo, mode), - Sender: convert.ToUser(doer, nil), - IsPull: false, - }) + eventType = webhook.HookEventIssueComment } - 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{ - 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, - }) + eventType = webhook.HookEventPullRequestComment } else { - err = webhook_services.PrepareWebhooks(comment.Issue.Repo, webhook.HookEventIssueComment, &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: false, - }) + eventType = webhook.HookEventIssueComment } - 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 { - // TODO https://github.com/go-gitea/gitea/pull/17940 - return + source := webhook_services.EventSource{ + Repository: pd.Repository, + 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), diff --git a/routers/api/v1/repo/hook.go b/routers/api/v1/repo/hook.go index 86361817cb..5956fe9da9 100644 --- a/routers/api/v1/repo/hook.go +++ b/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, diff --git a/routers/api/v1/repo/hook_test.go b/routers/api/v1/repo/hook_test.go index 07f1532f82..fd9a165bf3 100644 --- a/routers/api/v1/repo/hook_test.go +++ b/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)) } diff --git a/routers/web/repo/webhook.go b/routers/web/repo/webhook.go index 425198ce24..ee980333b7 100644 --- a/routers/web/repo/webhook.go +++ b/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 { 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 +// 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 +} + +// handle delivers hook tasks func handle(data ...queue.Data) []queue.Data { - for _, datum := range data { - repoIDStr := datum.(string) - log.Trace("DeliverHooks [repo_id: %v]", repoIDStr) + ctx := graceful.GetManager().HammerContext() - repoID, err := strconv.ParseInt(repoIDStr, 10, 64) + for _, taskID := range data { + task, err := webhook_model.GetHookTaskByID(ctx, taskID.(int64)) if err != nil { - log.Error("Invalid repo ID: %s", repoIDStr) - continue - } - - tasks, err := webhook_model.FindRepoUndeliveredHookTasks(repoID) - 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) - } - - // check if repo belongs to org and append additional webhooks - if repo.MustOwner().IsOrganization() { - // get hooks for org - orgHooks, err := webhook_model.ListWebhooksByOpts(ctx, &webhook_model.ListWebhookOptions{ - OrgID: repo.OwnerID, + 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("GetActiveWebhooksByOrgID: %v", err) + return fmt.Errorf("ListWebhooksByOpts: %v", err) + } + ws = append(ws, repoHooks...) + + owner = source.Repository.MustOwner() + } + + // 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: owner.ID, + IsActive: util.OptionalBoolTrue, + }) + if err != nil { + 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)