refactor: refact the webhook API and life process (#18255)

refactor: refact the notification job API and life process

1. Introduce new APIs for webhook jobs management.
2. Refact legacy APIs for backforward compatible.
3. Migrate the webhook jobs process to unified execution/task framework.

Closes: #18210

Signed-off-by: chlins <chenyuzh@vmware.com>

api/v2.0/swagger.yaml

@@ -2665,8 +2665,128 @@ paths:
           $ref: '#/responses/404'
         '500':
           $ref: '#/responses/500'
+  '/projects/{project_name_or_id}/webhook/policies/{webhook_policy_id}/executions':
+    get:
+      summary: List executions for a specific webhook policy
+      description: |
+        This endpoint returns the executions of a specific webhook policy.
+      tags:
+        - webhook
+      operationId: ListExecutionsOfWebhookPolicy
+      parameters:
+        - $ref: '#/parameters/requestId'
+        - $ref: '#/parameters/isResourceName'
+        - $ref: '#/parameters/projectNameOrId'
+        - $ref: '#/parameters/webhookPolicyId'
+        - $ref: '#/parameters/page'
+        - $ref: '#/parameters/pageSize'
+        - $ref: '#/parameters/query'
+        - $ref: '#/parameters/sort'
+      responses:
+        '200':
+          description: List webhook executions success
+          headers:
+            X-Total-Count:
+              description: The total count of executions
+              type: integer
+            Link:
+              description: Link refers to the previous page and next page
+              type: string
+          schema:
+            type: array
+            items:
+              $ref: '#/definitions/Execution'
+        '400':
+          $ref: '#/responses/400'
+        '401':
+          $ref: '#/responses/401'
+        '404':
+          $ref: '#/responses/404'
+        '403':
+          $ref: '#/responses/403'
+        '500':
+          $ref: '#/responses/500'
+  '/projects/{project_name_or_id}/webhook/policies/{webhook_policy_id}/executions/{execution_id}/tasks':
+    get:
+      summary: List tasks for a specific webhook execution
+      description: |
+        This endpoint returns the tasks of a specific webhook execution.
+      tags:
+        - webhook
+      operationId: ListTasksOfWebhookExecution
+      parameters:
+        - $ref: '#/parameters/requestId'
+        - $ref: '#/parameters/isResourceName'
+        - $ref: '#/parameters/projectNameOrId'
+        - $ref: '#/parameters/webhookPolicyId'
+        - $ref: '#/parameters/executionId'
+        - $ref: '#/parameters/page'
+        - $ref: '#/parameters/pageSize'
+        - $ref: '#/parameters/query'
+        - $ref: '#/parameters/sort'
+      responses:
+        '200':
+          description: List tasks of webhook executions success
+          headers:
+            X-Total-Count:
+              description: The total count of tasks
+              type: integer
+            Link:
+              description: Link refers to the previous page and next page
+              type: string
+          schema:
+            type: array
+            items:
+              $ref: '#/definitions/Task'
+        '400':
+          $ref: '#/responses/400'
+        '401':
+          $ref: '#/responses/401'
+        '404':
+          $ref: '#/responses/404'
+        '403':
+          $ref: '#/responses/403'
+        '500':
+          $ref: '#/responses/500'
+  '/projects/{project_name_or_id}/webhook/policies/{webhook_policy_id}/executions/{execution_id}/tasks/{task_id}/log':
+    get:
+      summary: Get logs for a specific webhook task
+      description: |
+        This endpoint returns the logs of a specific webhook task.
+      tags:
+        - webhook
+      operationId: GetLogsOfWebhookTask
+      produces:
+        - text/plain
+      parameters:
+        - $ref: '#/parameters/requestId'
+        - $ref: '#/parameters/isResourceName'
+        - $ref: '#/parameters/projectNameOrId'
+        - $ref: '#/parameters/webhookPolicyId'
+        - $ref: '#/parameters/executionId'
+        - $ref: '#/parameters/taskId'
+      responses:
+        '200':
+          description: Get log success
+          headers:
+            Content-Type:
+              description: Content type of response
+              type: string
+          schema:
+            type: string
+        '400':
+          $ref: '#/responses/400'
+        '401':
+          $ref: '#/responses/401'
+        '404':
+          $ref: '#/responses/404'
+        '403':
+          $ref: '#/responses/403'
+        '500':
+          $ref: '#/responses/500'
   '/projects/{project_name_or_id}/webhook/lasttrigger':
     get:
+      deprecated: true
       summary: Get project webhook policy last trigger info
       description: |
         This endpoint returns last trigger information of project webhook policy.
@@ -2694,6 +2814,7 @@ paths:
           $ref: '#/responses/500'
   '/projects/{project_name_or_id}/webhook/jobs':
     get:
+      deprecated: true
       summary: List project webhook jobs
       description: |
         This endpoint returns webhook jobs of a project.
@@ -2746,7 +2867,7 @@ paths:
   '/projects/{project_name_or_id}/webhook/events':
     get:
       summary: Get supported event types and notify types.
-      description: Get supportted event types and notify types.
+      description: Get supported event types and notify types.
       tags:
         - webhook
       operationId: GetSupportedEventTypes
@@ -8343,7 +8464,7 @@ definitions:
         description: 'The group type, 1 for LDAP group, 2 for HTTP group, 3 for OIDC group.'
   SupportedWebhookEventTypes:
     type: object
-    description: Supportted webhook event types and notify types.
+    description: Supported webhook event types and notify types.
     properties:
       event_type:
         type: array
@@ -8353,14 +8474,33 @@ definitions:
         type: array
         items:
           $ref: '#/definitions/NotifyType'
+      payload_formats:
+        type: array
+        items:
+          $ref: '#/definitions/PayloadFormat'
   EventType:
     type: string
-    description: Webhook supportted event type.
-    example: 'pullImage'
+    description: Webhook supported event type.
+    example: 'PULL_ARTIFACT'
   NotifyType:
     type: string
-    description: Webhook supportted notify type.
+    description: Webhook supported notify type.
     example: 'http'
+  PayloadFormatType:
+    type: string
+    description: The type of webhook paylod format.
+    example: 'cloudevent'
+  PayloadFormat:
+    type: object
+    description: Webhook supported payload format type collections.
+    properties:
+      notify_type:
+        $ref: '#/definitions/NotifyType'
+      formats:
+        type: array
+        description: The supported payload formats for this notify type.
+        items:
+          $ref: '#/definitions/PayloadFormatType'
 
   WebhookTargetObject:
     type: object
@@ -8378,6 +8518,8 @@ definitions:
       skip_cert_verify:
         type: boolean
         description: Whether or not to skip cert verify.
+      payload_format:
+        $ref: '#/definitions/PayloadFormatType'
   WebhookPolicy:
     type: object
     description: The webhook policy object

src/controller/webhook/controller.go

@@ -0,0 +1,195 @@
+//  Copyright Project Harbor Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+package webhook
+
+import (
+	"context"
+	"time"
+
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/lib/errors"
+	"github.com/goharbor/harbor/src/lib/q"
+	"github.com/goharbor/harbor/src/pkg/notification/policy"
+	"github.com/goharbor/harbor/src/pkg/notification/policy/model"
+	"github.com/goharbor/harbor/src/pkg/task"
+)
+
+var (
+	// Ctl is a global webhook controller instance
+	Ctl = NewController()
+
+	// webhookJobVendors represents webhook(http) or slack.
+	webhookJobVendors = q.NewOrList([]interface{}{job.WebhookJobVendorType, job.SlackJobVendorType})
+)
+
+type Controller interface {
+	// CreatePolicy creates webhook policy
+	CreatePolicy(ctx context.Context, policy *model.Policy) (int64, error)
+	// ListPolicies lists webhook policies filter by query
+	ListPolicies(ctx context.Context, query *q.Query) ([]*model.Policy, error)
+	// CountPolicies counts webhook policies filter by query
+	CountPolicies(ctx context.Context, query *q.Query) (int64, error)
+	// GetPolicy gets webhook policy by specified ID
+	GetPolicy(ctx context.Context, id int64) (*model.Policy, error)
+	// UpdatePolicy updates webhook policy
+	UpdatePolicy(ctx context.Context, policy *model.Policy) error
+	// DeletePolicy deletes webhook policy by specified ID
+	DeletePolicy(ctx context.Context, policyID int64) error
+	// GetRelatedPolices gets related policies by the input project id and event type
+	GetRelatedPolices(ctx context.Context, projectID int64, eventType string) ([]*model.Policy, error)
+
+	// CountExecutions counts executions under the webhook policy
+	CountExecutions(ctx context.Context, policyID int64, query *q.Query) (int64, error)
+	// ListExecutions lists executions under the webhook policy
+	ListExecutions(ctx context.Context, policyID int64, query *q.Query) ([]*task.Execution, error)
+	// CountTasks counts tasks under the webhook execution
+	CountTasks(ctx context.Context, execID int64, query *q.Query) (int64, error)
+	// ListTasks lists tasks under the webhook execution
+	ListTasks(ctx context.Context, execID int64, query *q.Query) ([]*task.Task, error)
+	// GetTask gets the webhook task by the specified ID
+	GetTask(ctx context.Context, taskID int64) (*task.Task, error)
+	// GetTaskLog gets task log
+	GetTaskLog(ctx context.Context, taskID int64) ([]byte, error)
+
+	// GetLastTriggerTime gets policy last trigger time group by event type
+	GetLastTriggerTime(ctx context.Context, eventType string, policyID int64) (time.Time, error)
+}
+
+type controller struct {
+	policyMgr policy.Manager
+	execMgr   task.ExecutionManager
+	taskMgr   task.Manager
+}
+
+func NewController() Controller {
+	return &controller{
+		policyMgr: policy.Mgr,
+		execMgr:   task.ExecMgr,
+		taskMgr:   task.Mgr,
+	}
+}
+
+func (c *controller) CreatePolicy(ctx context.Context, policy *model.Policy) (int64, error) {
+	return c.policyMgr.Create(ctx, policy)
+}
+
+func (c *controller) ListPolicies(ctx context.Context, query *q.Query) ([]*model.Policy, error) {
+	return c.policyMgr.List(ctx, query)
+}
+
+func (c *controller) CountPolicies(ctx context.Context, query *q.Query) (int64, error) {
+	return c.policyMgr.Count(ctx, query)
+}
+
+func (c *controller) GetPolicy(ctx context.Context, id int64) (*model.Policy, error) {
+	return c.policyMgr.Get(ctx, id)
+}
+
+func (c *controller) UpdatePolicy(ctx context.Context, policy *model.Policy) error {
+	return c.policyMgr.Update(ctx, policy)
+}
+
+func (c *controller) DeletePolicy(ctx context.Context, policyID int64) error {
+	// delete executions under the webhook policy,
+	// there are two vendor types(webhook & slack) needs to be deleted.
+	if err := c.execMgr.DeleteByVendor(ctx, job.WebhookJobVendorType, policyID); err != nil {
+		return errors.Wrapf(err, "failed to delete executions for webhook of policy %d", policyID)
+	}
+	if err := c.execMgr.DeleteByVendor(ctx, job.SlackJobVendorType, policyID); err != nil {
+		return errors.Wrapf(err, "failed to delete executions for slack of policy %d", policyID)
+	}
+
+	return c.policyMgr.Delete(ctx, policyID)
+}
+
+func (c *controller) GetRelatedPolices(ctx context.Context, projectID int64, eventType string) ([]*model.Policy, error) {
+	return c.policyMgr.GetRelatedPolices(ctx, projectID, eventType)
+}
+
+func (c *controller) CountExecutions(ctx context.Context, policyID int64, query *q.Query) (int64, error) {
+	return c.execMgr.Count(ctx, buildExecutionQuery(policyID, query))
+}
+
+func (c *controller) ListExecutions(ctx context.Context, policyID int64, query *q.Query) ([]*task.Execution, error) {
+	return c.execMgr.List(ctx, buildExecutionQuery(policyID, query))
+}
+
+func (c *controller) CountTasks(ctx context.Context, execID int64, query *q.Query) (int64, error) {
+	return c.taskMgr.Count(ctx, buildTaskQuery(execID, query))
+}
+
+func (c *controller) ListTasks(ctx context.Context, execID int64, query *q.Query) ([]*task.Task, error) {
+	return c.taskMgr.List(ctx, buildTaskQuery(execID, query))
+}
+
+func (c *controller) GetTask(ctx context.Context, taskID int64) (*task.Task, error) {
+	query := q.New(q.KeyWords{
+		"id":          taskID,
+		"vendor_type": webhookJobVendors,
+	})
+	tasks, err := c.taskMgr.List(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(tasks) == 0 {
+		return nil, errors.New(nil).WithCode(errors.NotFoundCode).
+			WithMessage("webhook task %d not found", taskID)
+	}
+	return tasks[0], nil
+}
+
+func (c *controller) GetTaskLog(ctx context.Context, taskID int64) ([]byte, error) {
+	// ensure the webhook task exist
+	_, err := c.GetTask(ctx, taskID)
+	if err != nil {
+		return nil, err
+	}
+
+	return c.taskMgr.GetLog(ctx, taskID)
+}
+
+func buildExecutionQuery(policyID int64, query *q.Query) *q.Query {
+	query = q.MustClone(query)
+	query.Keywords["vendor_type"] = webhookJobVendors
+	query.Keywords["vendor_id"] = policyID
+	return query
+}
+
+func buildTaskQuery(execID int64, query *q.Query) *q.Query {
+	query = q.MustClone(query)
+	query.Keywords["vendor_type"] = webhookJobVendors
+	query.Keywords["execution_id"] = execID
+	return query
+}
+
+func (c *controller) GetLastTriggerTime(ctx context.Context, eventType string, policyID int64) (time.Time, error) {
+	query := q.New(q.KeyWords{
+		"vendor_type":     webhookJobVendors,
+		"vendor_id":       policyID,
+		"ExtraAttrs.type": eventType,
+	})
+	// fetch the latest execution sort by start_time
+	execs, err := c.execMgr.List(ctx, query.First(q.NewSort("start_time", true)))
+	if err != nil {
+		return time.Time{}, err
+	}
+
+	if len(execs) > 0 {
+		return execs[0].StartTime, nil
+	}
+
+	return time.Time{}, nil
+}

src/controller/webhook/controller_test.go

@@ -0,0 +1,166 @@
+//  Copyright Project Harbor Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+package webhook
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/goharbor/harbor/src/lib/q"
+	"github.com/goharbor/harbor/src/pkg/notification/policy/model"
+	task_model "github.com/goharbor/harbor/src/pkg/task"
+	"github.com/goharbor/harbor/src/testing/mock"
+	"github.com/goharbor/harbor/src/testing/pkg/notification/policy"
+	"github.com/goharbor/harbor/src/testing/pkg/task"
+	"github.com/stretchr/testify/suite"
+)
+
+type controllerTestSuite struct {
+	suite.Suite
+	ctl       *controller
+	policyMgr *policy.Manager
+	taskMgr   *task.Manager
+	execMgr   *task.ExecutionManager
+}
+
+func TestControllerTestSuite(t *testing.T) {
+	suite.Run(t, &controllerTestSuite{})
+}
+
+func (c *controllerTestSuite) SetupTest() {
+	c.policyMgr = &policy.Manager{}
+	c.taskMgr = &task.Manager{}
+	c.execMgr = &task.ExecutionManager{}
+	c.ctl = &controller{
+		policyMgr: c.policyMgr,
+		taskMgr:   c.taskMgr,
+		execMgr:   c.execMgr,
+	}
+}
+
+func (c *controllerTestSuite) TestCreatePolicy() {
+	c.policyMgr.On("Create", mock.Anything, mock.Anything).Return(int64(1), nil)
+	id, err := c.ctl.CreatePolicy(context.TODO(), &model.Policy{Name: "test-policy"})
+	c.NoError(err)
+	c.Equal(int64(1), id)
+}
+
+func (c *controllerTestSuite) TestListPolicies() {
+	c.policyMgr.On("List", mock.Anything, mock.Anything).Return([]*model.Policy{{Name: "test-policy-1"}, {Name: "test-policy-2"}}, nil)
+	policies, err := c.ctl.ListPolicies(context.TODO(), q.MustClone(nil))
+	c.NoError(err)
+	c.Len(policies, 2)
+}
+
+func (c *controllerTestSuite) TestCountPolicies() {
+	c.policyMgr.On("Count", mock.Anything, mock.Anything).Return(int64(3), nil)
+	cnt, err := c.ctl.CountPolicies(context.TODO(), q.MustClone(nil))
+	c.NoError(err)
+	c.Equal(int64(3), cnt)
+}
+
+func (c *controllerTestSuite) TestGetPolicy() {
+	c.policyMgr.On("Get", mock.Anything, mock.Anything).Return(&model.Policy{Name: "test-policy"}, nil)
+	p, err := c.ctl.GetPolicy(context.TODO(), 1)
+	c.NoError(err)
+	c.Equal("test-policy", p.Name)
+}
+
+func (c *controllerTestSuite) TestUpdatePolicy() {
+	c.policyMgr.On("Update", mock.Anything, mock.Anything).Return(nil)
+	err := c.ctl.UpdatePolicy(context.TODO(), &model.Policy{})
+	c.NoError(err)
+}
+
+func (c *controllerTestSuite) TestDeletePolicy() {
+	delExecErr := errors.New("delete executions error")
+	// failed to delete policy due to webhook executions deletion error
+	c.execMgr.On("DeleteByVendor", mock.Anything, "WEBHOOK", mock.Anything).Return(delExecErr).Once()
+	err := c.ctl.DeletePolicy(context.TODO(), 1)
+	c.ErrorIs(err, delExecErr)
+
+	// failed to delete policy due to slack executions deletion error
+	c.execMgr.On("DeleteByVendor", mock.Anything, "WEBHOOK", mock.Anything).Return(nil).Once()
+	c.execMgr.On("DeleteByVendor", mock.Anything, "SLACK", mock.Anything).Return(delExecErr).Once()
+	err = c.ctl.DeletePolicy(context.TODO(), 1)
+	c.ErrorIs(err, delExecErr)
+
+	// successfully deletion for all
+	c.execMgr.On("DeleteByVendor", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	c.policyMgr.On("Delete", mock.Anything, mock.Anything).Return(nil)
+	err = c.ctl.DeletePolicy(context.TODO(), 1)
+	c.NoError(err)
+}
+
+func (c *controllerTestSuite) TestGetRelatedPolices() {
+	c.policyMgr.On("GetRelatedPolices", mock.Anything, mock.Anything, mock.Anything).Return([]*model.Policy{{Name: "test-policy-1"}, {Name: "test-policy-2"}}, nil)
+	policies, err := c.ctl.GetRelatedPolices(context.TODO(), 1, "mock")
+	c.NoError(err)
+	c.Len(policies, 2)
+}
+
+func (c *controllerTestSuite) TestCountExecutions() {
+	c.execMgr.On("Count", mock.Anything, mock.Anything).Return(int64(1), nil)
+	cnt, err := c.ctl.CountExecutions(context.TODO(), 1, q.MustClone(nil))
+	c.NoError(err)
+	c.Equal(int64(1), cnt)
+}
+
+func (c *controllerTestSuite) TestListExecutions() {
+	c.execMgr.On("List", mock.Anything, mock.Anything).Return([]*task_model.Execution{{ID: 1, VendorType: "WEBHOOK", VendorID: 1}}, nil)
+	execs, err := c.ctl.ListExecutions(context.TODO(), 1, q.MustClone(nil))
+	c.NoError(err)
+	c.Len(execs, 1)
+}
+
+func (c *controllerTestSuite) TestCountTasks() {
+	c.taskMgr.On("Count", mock.Anything, mock.Anything).Return(int64(1), nil)
+	cnt, err := c.ctl.CountTasks(context.TODO(), 1, q.MustClone(nil))
+	c.NoError(err)
+	c.Equal(int64(1), cnt)
+}
+
+func (c *controllerTestSuite) TestListTasks() {
+	c.taskMgr.On("List", mock.Anything, mock.Anything).Return([]*task_model.Task{{ID: 1, ExecutionID: 1}}, nil)
+	tasks, err := c.ctl.ListTasks(context.TODO(), 1, q.MustClone(nil))
+	c.NoError(err)
+	c.Len(tasks, 1)
+}
+
+func (c *controllerTestSuite) TestGetTask() {
+	c.taskMgr.On("List", mock.Anything, mock.Anything).Return([]*task_model.Task{{ID: 1, ExecutionID: 1}}, nil)
+	task, err := c.ctl.GetTask(context.TODO(), 1)
+	c.NoError(err)
+	c.Equal(int64(1), task.ID)
+	c.Equal(int64(1), task.ExecutionID)
+}
+
+func (c *controllerTestSuite) TestGetTaskLog() {
+	c.taskMgr.On("List", mock.Anything, mock.Anything).Return([]*task_model.Task{{ID: 1, ExecutionID: 1}}, nil)
+	c.taskMgr.On("GetLog", mock.Anything, mock.Anything).Return([]byte("test logs"), nil)
+	logs, err := c.ctl.GetTaskLog(context.TODO(), 1)
+	c.NoError(err)
+	c.Equal("test logs", string(logs))
+}
+
+func (c *controllerTestSuite) TestGetLastTriggerTime() {
+	now := time.Now()
+	c.execMgr.On("List", mock.Anything, mock.Anything).Return([]*task_model.Execution{{ID: 1, StartTime: now}}, nil)
+	time, err := c.ctl.GetLastTriggerTime(context.TODO(), "mock", 1)
+	c.NoError(err)
+	c.Equal(now, time)
+}

src/core/middlewares/middlewares.go

@@ -70,10 +70,8 @@ var (
 		middleware.MethodAndPathSkipper(http.MethodPost, match("^/service/notifications/jobs/adminjob/"+numericRegexp.String())),
 		middleware.MethodAndPathSkipper(http.MethodPost, match("^/service/notifications/jobs/replication/"+numericRegexp.String())),
 		middleware.MethodAndPathSkipper(http.MethodPost, match("^/service/notifications/jobs/replication/task/"+numericRegexp.String())),
-		middleware.MethodAndPathSkipper(http.MethodPost, match("^/service/notifications/jobs/webhook/"+numericRegexp.String())),
 		middleware.MethodAndPathSkipper(http.MethodPost, match("^/service/notifications/jobs/retention/task/"+numericRegexp.String())),
 		middleware.MethodAndPathSkipper(http.MethodPost, match("^/service/notifications/jobs/schedules/"+numericRegexp.String())),
-		middleware.MethodAndPathSkipper(http.MethodPost, match("^/service/notifications/jobs/webhook/"+numericRegexp.String())),
 		pingSkipper,
 	}
 )

src/core/service/notifications/base.go

@@ -1,8 +0,0 @@
-package notifications
-
-import "github.com/goharbor/harbor/src/core/api"
-
-// BaseHandler extracts the common funcs, all notification handlers should shadow this struct
-type BaseHandler struct {
-	api.BaseController
-}

src/core/service/notifications/jobs/handler.go

@@ -1,101 +0,0 @@
-// Copyright 2018 Project Harbor Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//    http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package jobs
-
-import (
-	"encoding/json"
-	"time"
-
-	"github.com/goharbor/harbor/src/common/job"
-	"github.com/goharbor/harbor/src/common/models"
-	"github.com/goharbor/harbor/src/core/service/notifications"
-	jjob "github.com/goharbor/harbor/src/jobservice/job"
-	"github.com/goharbor/harbor/src/lib/log"
-	"github.com/goharbor/harbor/src/lib/orm"
-	"github.com/goharbor/harbor/src/pkg/notification"
-	"github.com/goharbor/harbor/src/pkg/notification/job/model"
-)
-
-var statusMap = map[string]string{
-	job.JobServiceStatusPending:   models.JobPending,
-	job.JobServiceStatusScheduled: models.JobScheduled,
-	job.JobServiceStatusRunning:   models.JobRunning,
-	job.JobServiceStatusStopped:   models.JobStopped,
-	job.JobServiceStatusError:     models.JobError,
-	job.JobServiceStatusSuccess:   models.JobFinished,
-}
-
-// Handler handles request on /service/notifications/jobs/*, which listens to the webhook of jobservice.
-type Handler struct {
-	notifications.BaseHandler
-	id        int64
-	status    string
-	rawStatus string
-	checkIn   string
-	revision  int64
-	trackID   string
-	change    *jjob.StatusChange
-}
-
-// Prepare ...
-func (h *Handler) Prepare() {
-	h.BaseHandler.Prepare()
-	h.trackID = h.GetStringFromPath(":uuid")
-	if len(h.trackID) == 0 {
-		id, err := h.GetInt64FromPath(":id")
-		if err != nil {
-			log.Errorf("Failed to get job ID, error: %v", err)
-			// Avoid job service from resending...
-			h.Abort("200")
-			return
-		}
-		h.id = id
-	}
-
-	var data jjob.StatusChange
-	err := json.Unmarshal(h.Ctx.Input.CopyBody(1<<32), &data)
-	if err != nil {
-		log.Errorf("Failed to decode job status change with error: %v", err)
-		h.Abort("200")
-		return
-	}
-	h.change = &data
-	h.rawStatus = data.Status
-	status, ok := statusMap[data.Status]
-	if !ok {
-		log.Debugf("drop the job status update event: job id-%d/track id-%s, status-%s", h.id, h.trackID, status)
-		h.Abort("200")
-		return
-	}
-	h.status = status
-	h.checkIn = data.CheckIn
-	if data.Metadata != nil {
-		h.revision = data.Metadata.Revision
-	}
-}
-
-// HandleNotificationJob handles the hook of notification job
-func (h *Handler) HandleNotificationJob() {
-	log.Debugf("received notification job status update event: job-%d, status-%s", h.id, h.status)
-	if err := notification.JobMgr.Update(orm.Context(), &model.Job{
-		ID:         h.id,
-		Status:     h.status,
-		UpdateTime: time.Now(),
-	}, "Status", "UpdateTime"); err != nil {
-		log.Errorf("Failed to update notification job status, id: %d, status: %s", h.id, h.status)
-		h.SendInternalServerError(err)
-		return
-	}
-}

src/pkg/notification/hook/hook.go

@@ -4,17 +4,13 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"time"
 
-	cJob "github.com/goharbor/harbor/src/common/job"
 	"github.com/goharbor/harbor/src/common/job/models"
-	cModels "github.com/goharbor/harbor/src/common/models"
-	"github.com/goharbor/harbor/src/core/utils"
-	"github.com/goharbor/harbor/src/lib/config"
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/lib/errors"
 	"github.com/goharbor/harbor/src/lib/log"
-	"github.com/goharbor/harbor/src/pkg/notification/job"
-	job_model "github.com/goharbor/harbor/src/pkg/notification/job/model"
 	"github.com/goharbor/harbor/src/pkg/notifier/model"
+	"github.com/goharbor/harbor/src/pkg/task"
 )
 
 // Manager send hook
@@ -24,63 +20,61 @@ type Manager interface {
 
 // DefaultManager ...
 type DefaultManager struct {
-	jobMgr job.Manager
-	client cJob.Client
+	execMgr task.ExecutionManager
+	taskMgr task.Manager
 }
 
 // NewHookManager ...
 func NewHookManager() *DefaultManager {
 	return &DefaultManager{
-		jobMgr: job.NewManager(),
-		client: utils.GetJobServiceClient(),
+		execMgr: task.ExecMgr,
+		taskMgr: task.Mgr,
 	}
 }
 
-// StartHook create a notification job record in database, and submit it to jobservice
+// StartHook create a webhook job record in database, and submit it to jobservice
 func (hm *DefaultManager) StartHook(ctx context.Context, event *model.HookEvent, data *models.JobData) error {
 	payload, err := json.Marshal(event.Payload)
 	if err != nil {
 		return err
 	}
 
-	t := time.Now()
-	id, err := hm.jobMgr.Create(ctx, &job_model.Job{
-		PolicyID:     event.PolicyID,
-		EventType:    event.EventType,
-		NotifyType:   event.Target.Type,
-		Status:       cModels.JobPending,
-		CreationTime: t,
-		UpdateTime:   t,
-		JobDetail:    string(payload),
-	})
-	if err != nil {
-		return fmt.Errorf("failed to create the job record for notification based on policy %d: %v", event.PolicyID, err)
-	}
-	statusHookURL := fmt.Sprintf("%s/service/notifications/jobs/webhook/%d", config.InternalCoreURL(), id)
-	data.StatusHook = statusHookURL
-
-	log.Debugf("created a notification job %d for the policy %d", id, event.PolicyID)
-
-	// submit hook job to jobservice
-	jobUUID, err := hm.client.SubmitJob(data)
-	if err != nil {
-		log.Errorf("failed to submit job with notification event: %v", err)
-		e := hm.jobMgr.Update(ctx, &job_model.Job{
-			ID:     id,
-			Status: cModels.JobError,
-		}, "Status")
-		if e != nil {
-			log.Errorf("failed to update the notification job status %d: %v", id, e)
-		}
+	extraAttrs := make(map[string]interface{})
+	if err = json.Unmarshal(payload, &extraAttrs); err != nil {
 		return err
 	}
 
-	if err = hm.jobMgr.Update(ctx, &job_model.Job{
-		ID:   id,
-		UUID: jobUUID,
-	}, "UUID"); err != nil {
-		log.Errorf("failed to update the notification job %d: %v", id, err)
-		return err
+	var vendorType string
+	switch event.Target.Type {
+	case model.NotifyTypeHTTP:
+		vendorType = job.WebhookJobVendorType
+	case model.NotifyTypeSlack:
+		vendorType = job.SlackJobVendorType
 	}
+
+	if len(vendorType) == 0 {
+		return errors.Errorf("invalid event target type: %s", event.Target.Type)
+	}
+
+	// create execution firstly, then create task.
+	execID, err := hm.execMgr.Create(ctx, vendorType, event.PolicyID, task.ExecutionTriggerEvent, extraAttrs)
+	if err != nil {
+		log.Errorf("failed to create execution for webhook based on policy %d: %v", event.PolicyID, err)
+		return nil
+	}
+
+	taskID, err := hm.taskMgr.Create(ctx, execID, &task.Job{
+		Name: data.Name,
+		Metadata: &job.Metadata{
+			JobKind: data.Metadata.JobKind,
+		},
+		Parameters: map[string]interface{}(data.Parameters),
+	}, extraAttrs)
+	if err != nil {
+		return fmt.Errorf("failed to create the task for webhook based on policy %d: %v", event.PolicyID, err)
+	}
+
+	log.Debugf("created a webhook job %d for the policy %d", taskID, event.PolicyID)
+
 	return nil
 }

src/pkg/notification/job/dao/dao.go

@@ -1,186 +0,0 @@
-package dao
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/goharbor/harbor/src/common/models"
-	"github.com/goharbor/harbor/src/lib/errors"
-	"github.com/goharbor/harbor/src/lib/log"
-	"github.com/goharbor/harbor/src/lib/orm"
-	"github.com/goharbor/harbor/src/lib/q"
-	"github.com/goharbor/harbor/src/pkg/notification/job/model"
-)
-
-// DAO defines the interface to access the robot data model
-type DAO interface {
-	// Create ...
-	Create(ctx context.Context, n *model.Job) (int64, error)
-
-	// Update ...
-	Update(ctx context.Context, n *model.Job, props ...string) error
-
-	// Get ...
-	Get(ctx context.Context, id int64) (*model.Job, error)
-
-	// Count ...
-	Count(ctx context.Context, query *q.Query) (total int64, err error)
-
-	// List ...
-	List(ctx context.Context, query *q.Query) ([]*model.Job, error)
-
-	// Delete ...
-	Delete(ctx context.Context, id int64) error
-
-	// GetLastTriggerJobsGroupByEventType ...
-	GetLastTriggerJobsGroupByEventType(ctx context.Context, policyID int64) ([]*model.Job, error)
-
-	// DeleteByPolicyID
-	DeleteByPolicyID(ctx context.Context, policyID int64) error
-}
-
-// New creates a default implementation for Dao
-func New() DAO {
-	return &dao{}
-}
-
-type dao struct{}
-
-// UpdateNotificationJob update notification job
-func (d *dao) Update(ctx context.Context, job *model.Job, props ...string) error {
-	if job == nil {
-		return errors.New("nil job")
-	}
-
-	if job.ID == 0 {
-		return fmt.Errorf("notification job ID is empty")
-	}
-
-	ormer, err := orm.FromContext(ctx)
-	if err != nil {
-		return err
-	}
-	n, err := ormer.Update(job, props...)
-	if n == 0 {
-		return errors.NotFoundError(nil).WithMessage("notification %d not found", job.ID)
-	}
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
-// Create insert new notification job to DB
-func (d *dao) Create(ctx context.Context, job *model.Job) (int64, error) {
-	if job == nil {
-		return 0, errors.New("nil job")
-	}
-	ormer, err := orm.FromContext(ctx)
-	if err != nil {
-		return 0, err
-	}
-	if len(job.Status) == 0 {
-		job.Status = models.JobPending
-	}
-	return ormer.Insert(job)
-}
-
-// Get ...
-func (d *dao) Get(ctx context.Context, id int64) (*model.Job, error) {
-	ormer, err := orm.FromContext(ctx)
-	if err != nil {
-		return nil, err
-	}
-	j := &model.Job{
-		ID: id,
-	}
-	if err := ormer.Read(j); err != nil {
-		if e := orm.AsNotFoundError(err, "notificationJob %d not found", id); e != nil {
-			err = e
-		}
-		return nil, err
-	}
-	return j, nil
-}
-
-// Count ...
-func (d *dao) Count(ctx context.Context, query *q.Query) (int64, error) {
-	qs, err := orm.QuerySetterForCount(ctx, &model.Job{}, query)
-	if err != nil {
-		return 0, err
-	}
-	return qs.Count()
-}
-
-// List ...
-func (d *dao) List(ctx context.Context, query *q.Query) ([]*model.Job, error) {
-	jobs := []*model.Job{}
-
-	qs, err := orm.QuerySetter(ctx, &model.Job{}, query)
-	if err != nil {
-		return nil, err
-	}
-	if _, err = qs.All(&jobs); err != nil {
-		return nil, err
-	}
-	return jobs, nil
-}
-
-// GetLastTriggerJobsGroupByEventType get notification jobs info of policy, including event type and last trigger time
-func (d *dao) GetLastTriggerJobsGroupByEventType(ctx context.Context, policyID int64) ([]*model.Job, error) {
-	ormer, err := orm.FromContext(ctx)
-	if err != nil {
-		return nil, err
-	}
-	// get jobs last triggered(created) group by event_type. postgres group by usage reference:
-	// https://stackoverflow.com/questions/13325583/postgresql-max-and-group-by
-	sql := `select distinct on (event_type) event_type, id, creation_time, status, notify_type, job_uuid, update_time, 
-			creation_time, job_detail from notification_job where policy_id = ? 
-			order by event_type, id desc, creation_time, status, notify_type, job_uuid, update_time, creation_time, job_detail`
-
-	jobs := []*model.Job{}
-	_, err = ormer.Raw(sql, policyID).QueryRows(&jobs)
-	if err != nil {
-		log.Errorf("query last trigger info group by event type failed: %v", err)
-		return nil, err
-	}
-
-	return jobs, nil
-}
-
-func (d *dao) Delete(ctx context.Context, id int64) error {
-	ormer, err := orm.FromContext(ctx)
-	if err != nil {
-		return err
-	}
-	n, err := ormer.Delete(&model.Job{
-		ID: id,
-	})
-	if err != nil {
-		return err
-	}
-	if n == 0 {
-		return errors.NotFoundError(nil).WithMessage("notificationJob %d not found", id)
-	}
-	return nil
-}
-
-// DeleteByPolicyID ...
-func (d *dao) DeleteByPolicyID(ctx context.Context, policyID int64) error {
-	qs, err := orm.QuerySetter(ctx, &model.Job{}, &q.Query{
-		Keywords: map[string]interface{}{
-			"policy_id": policyID,
-		},
-	})
-	if err != nil {
-		return err
-	}
-	n, err := qs.Delete()
-	if err != nil {
-		return err
-	}
-	if n == 0 {
-		return errors.NotFoundError(nil).WithMessage("notificationJob %d not found", policyID)
-	}
-	return nil
-}

src/pkg/notification/job/dao/dao_test.go

@@ -1,189 +0,0 @@
-package dao
-
-import (
-	"testing"
-
-	"github.com/stretchr/testify/suite"
-
-	"github.com/goharbor/harbor/src/lib/errors"
-	"github.com/goharbor/harbor/src/lib/orm"
-	"github.com/goharbor/harbor/src/lib/q"
-	"github.com/goharbor/harbor/src/pkg/notification/job/model"
-	htesting "github.com/goharbor/harbor/src/testing"
-)
-
-var (
-	testJob1 = &model.Job{
-		PolicyID:   1111,
-		EventType:  "pushImage",
-		NotifyType: "http",
-		Status:     "pending",
-		JobDetail:  "{\"type\":\"pushImage\",\"occur_at\":1563536782,\"event_data\":{\"resources\":[{\"digest\":\"sha256:bf1684a6e3676389ec861c602e97f27b03f14178e5bc3f70dce198f9f160cce9\",\"tag\":\"v1.0\",\"resource_url\":\"10.194.32.23/myproj/alpine:v1.0\"}],\"repository\":{\"date_created\":1563505587,\"name\":\"alpine\",\"namespace\":\"myproj\",\"repo_full_name\":\"myproj/alpine\",\"repo_type\":\"private\"}},\"operator\":\"admin\"}",
-		UUID:       "00000000",
-	}
-	testJob2 = &model.Job{
-		PolicyID:   111,
-		EventType:  "pullImage",
-		NotifyType: "http",
-		Status:     "",
-		JobDetail:  "{\"type\":\"pushImage\",\"occur_at\":1563537782,\"event_data\":{\"resources\":[{\"digest\":\"sha256:bf1684a6e3676389ec861c602e97f27b03f14178e5bc3f70dce198f9f160cce9\",\"tag\":\"v1.0\",\"resource_url\":\"10.194.32.23/myproj/alpine:v1.0\"}],\"repository\":{\"date_created\":1563505587,\"name\":\"alpine\",\"namespace\":\"myproj\",\"repo_full_name\":\"myproj/alpine\",\"repo_type\":\"private\"}},\"operator\":\"admin\"}",
-		UUID:       "00000000",
-	}
-	testJob3 = &model.Job{
-		PolicyID:   111,
-		EventType:  "deleteImage",
-		NotifyType: "http",
-		Status:     "pending",
-		JobDetail:  "{\"type\":\"pushImage\",\"occur_at\":1563538782,\"event_data\":{\"resources\":[{\"digest\":\"sha256:bf1684a6e3676389ec861c602e97f27b03f14178e5bc3f70dce198f9f160cce9\",\"tag\":\"v1.0\",\"resource_url\":\"10.194.32.23/myproj/alpine:v1.0\"}],\"repository\":{\"date_created\":1563505587,\"name\":\"alpine\",\"namespace\":\"myproj\",\"repo_full_name\":\"myproj/alpine\",\"repo_type\":\"private\"}},\"operator\":\"admin\"}",
-		UUID:       "00000000",
-	}
-)
-
-type DaoTestSuite struct {
-	htesting.Suite
-	dao DAO
-
-	jobID1 int64
-	jobID2 int64
-	jobID3 int64
-}
-
-func (suite *DaoTestSuite) SetupSuite() {
-	suite.Suite.SetupSuite()
-	suite.dao = New()
-	suite.Suite.ClearTables = []string{"notification_job"}
-	suite.jobs()
-}
-
-func (suite *DaoTestSuite) jobs() {
-	var err error
-	suite.jobID1, err = suite.dao.Create(orm.Context(), testJob1)
-	suite.Nil(err)
-
-	suite.jobID2, err = suite.dao.Create(orm.Context(), testJob2)
-	suite.Nil(err)
-
-	suite.jobID3, err = suite.dao.Create(orm.Context(), testJob3)
-	suite.Nil(err)
-}
-
-func (suite *DaoTestSuite) TestCreate() {
-	_, err := suite.dao.Create(orm.Context(), nil)
-	suite.NotNil(err)
-}
-
-func (suite *DaoTestSuite) TestDelete() {
-	err := suite.dao.Delete(orm.Context(), 1234)
-	suite.Require().NotNil(err)
-	suite.True(errors.IsErr(err, errors.NotFoundCode))
-
-	err = suite.dao.Delete(orm.Context(), suite.jobID2)
-	suite.Nil(err)
-}
-
-func (suite *DaoTestSuite) TestList() {
-	jobs, err := suite.dao.List(orm.Context(), &q.Query{
-		Keywords: map[string]interface{}{
-			"EventType": "pushImage",
-		},
-	})
-	suite.Require().Nil(err)
-	suite.Equal(len(jobs), 1)
-	suite.Equal(suite.jobID1, jobs[0].ID)
-}
-
-func (suite *DaoTestSuite) TestGet() {
-	_, err := suite.dao.Get(orm.Context(), 1234)
-	suite.Require().NotNil(err)
-	suite.True(errors.IsErr(err, errors.NotFoundCode))
-
-	id, err := suite.dao.Create(orm.Context(), &model.Job{
-		PolicyID:   2222,
-		EventType:  "pullImage",
-		NotifyType: "http",
-		Status:     "pending",
-		JobDetail:  "{\"type\":\"pushImage\",\"occur_at\":1563536782,\"event_data\":{\"resources\":[{\"digest\":\"sha256:bf1684a6e3676389ec861c602e97f27b03f14178e5bc3f70dce198f9f160cce9\",\"tag\":\"v1.0\",\"resource_url\":\"10.194.32.23/myproj/alpine:v1.0\"}],\"repository\":{\"date_created\":1563505587,\"name\":\"alpine\",\"namespace\":\"myproj\",\"repo_full_name\":\"myproj/alpine\",\"repo_type\":\"private\"}},\"operator\":\"admin\"}",
-		UUID:       "00000000",
-	})
-	suite.Nil(err)
-
-	r, err := suite.dao.Get(orm.Context(), id)
-	suite.Nil(err)
-	suite.Equal("pullImage", r.EventType)
-}
-
-func (suite *DaoTestSuite) TestUpdate() {
-	j := &model.Job{
-		ID:     suite.jobID1,
-		Status: "success",
-	}
-
-	err := suite.dao.Update(orm.Context(), j)
-	suite.Nil(err)
-
-	r1, err := suite.dao.Get(orm.Context(), j.ID)
-	suite.Equal("success", r1.Status)
-}
-
-func (suite *DaoTestSuite) TestCount() {
-	// nil query
-	total, err := suite.dao.Count(orm.Context(), nil)
-	suite.Nil(err)
-	suite.True(total > 0)
-
-	// query by name
-	total, err = suite.dao.Count(orm.Context(), &q.Query{
-		Keywords: map[string]interface{}{
-			"EventType": "deleteImage",
-		},
-	})
-	suite.Nil(err)
-	suite.Equal(int64(1), total)
-}
-
-func (suite *DaoTestSuite) TestDeleteByPolicyID() {
-	jobs, err := suite.dao.List(orm.Context(), &q.Query{
-		Keywords: map[string]interface{}{
-			"PolicyID": 111,
-		},
-	})
-	suite.True(len(jobs) > 0)
-
-	err = suite.dao.DeleteByPolicyID(orm.Context(), 111)
-	suite.Nil(err)
-
-	jobs, err = suite.dao.List(orm.Context(), &q.Query{
-		Keywords: map[string]interface{}{
-			"PolicyID": 111,
-		},
-	})
-	suite.Equal(0, len(jobs))
-}
-
-func (suite *DaoTestSuite) TestGetLastTriggerJobsGroupByEventType() {
-	_, err := suite.dao.Create(orm.Context(), &model.Job{
-		PolicyID:   3333,
-		EventType:  "replicateImage",
-		NotifyType: "http",
-		Status:     "pending",
-		JobDetail:  "{\"type\":\"pushImage\",\"occur_at\":1563536782,\"event_data\":{\"resources\":[{\"digest\":\"sha256:bf1684a6e3676389ec861c602e97f27b03f14178e5bc3f70dce198f9f160cce9\",\"tag\":\"v1.0\",\"resource_url\":\"10.194.32.23/myproj/alpine:v1.0\"}],\"repository\":{\"date_created\":1563505587,\"name\":\"alpine\",\"namespace\":\"myproj\",\"repo_full_name\":\"myproj/alpine\",\"repo_type\":\"private\"}},\"operator\":\"admin\"}",
-		UUID:       "00000000",
-	})
-	suite.Nil(err)
-	_, err = suite.dao.Create(orm.Context(), &model.Job{
-		PolicyID:   3333,
-		EventType:  "pullImage",
-		NotifyType: "http",
-		Status:     "pending",
-		JobDetail:  "{\"type\":\"pushImage\",\"occur_at\":1563536782,\"event_data\":{\"resources\":[{\"digest\":\"sha256:bf1684a6e3676389ec861c602e97f27b03f14178e5bc3f70dce198f9f160cce9\",\"tag\":\"v1.0\",\"resource_url\":\"10.194.32.23/myproj/alpine:v1.0\"}],\"repository\":{\"date_created\":1563505587,\"name\":\"alpine\",\"namespace\":\"myproj\",\"repo_full_name\":\"myproj/alpine\",\"repo_type\":\"private\"}},\"operator\":\"admin\"}",
-		UUID:       "00000000",
-	})
-	suite.Nil(err)
-	jobs, err := suite.dao.GetLastTriggerJobsGroupByEventType(orm.Context(), 3333)
-	suite.Nil(err)
-	suite.Equal(2, len(jobs))
-}
-
-func TestDaoTestSuite(t *testing.T) {
-	suite.Run(t, &DaoTestSuite{})
-}

src/pkg/notification/job/manager.go

@@ -1,70 +0,0 @@
-package job
-
-import (
-	"context"
-
-	"github.com/goharbor/harbor/src/lib/q"
-	"github.com/goharbor/harbor/src/pkg/notification/job/dao"
-	"github.com/goharbor/harbor/src/pkg/notification/job/model"
-)
-
-var (
-	// Mgr is a global variable for the default notification job
-	Mgr = NewManager()
-)
-
-// Manager manages notification jobs recorded in database
-type Manager interface {
-	// Create create a notification job
-	Create(ctx context.Context, job *model.Job) (int64, error)
-
-	// List list notification jobs
-	List(ctx context.Context, query *q.Query) ([]*model.Job, error)
-
-	// Update update notification job
-	Update(ctx context.Context, job *model.Job, props ...string) error
-
-	// ListJobsGroupByEventType lists last triggered jobs group by event type
-	ListJobsGroupByEventType(ctx context.Context, policyID int64) ([]*model.Job, error)
-
-	// Count ...
-	Count(ctx context.Context, query *q.Query) (total int64, err error)
-}
-
-var _ Manager = &manager{}
-
-type manager struct {
-	dao dao.DAO
-}
-
-// NewManager ...
-func NewManager() Manager {
-	return &manager{
-		dao: dao.New(),
-	}
-}
-
-// Create ...
-func (d *manager) Create(ctx context.Context, job *model.Job) (int64, error) {
-	return d.dao.Create(ctx, job)
-}
-
-// Count ...
-func (d *manager) Count(ctx context.Context, query *q.Query) (int64, error) {
-	return d.dao.Count(ctx, query)
-}
-
-// List ...
-func (d *manager) List(ctx context.Context, query *q.Query) ([]*model.Job, error) {
-	return d.dao.List(ctx, query)
-}
-
-// Update ...
-func (d *manager) Update(ctx context.Context, job *model.Job, props ...string) error {
-	return d.dao.Update(ctx, job, props...)
-}
-
-// ListJobsGroupByEventType lists last triggered jobs group by event type
-func (d *manager) ListJobsGroupByEventType(ctx context.Context, policyID int64) ([]*model.Job, error) {
-	return d.dao.GetLastTriggerJobsGroupByEventType(ctx, policyID)
-}

src/pkg/notification/job/manager_test.go

@@ -1,83 +0,0 @@
-package job
-
-import (
-	"context"
-	"testing"
-
-	"github.com/stretchr/testify/suite"
-
-	"github.com/goharbor/harbor/src/pkg/notification/job/model"
-	"github.com/goharbor/harbor/src/testing/mock"
-	"github.com/goharbor/harbor/src/testing/pkg/notification/job/dao"
-)
-
-type managerTestSuite struct {
-	suite.Suite
-	mgr *manager
-	dao *dao.DAO
-}
-
-func (m *managerTestSuite) SetupTest() {
-	m.dao = &dao.DAO{}
-	m.mgr = &manager{
-		dao: m.dao,
-	}
-}
-
-func (m *managerTestSuite) TestCreate() {
-	m.dao.On("Create", mock.Anything, mock.Anything).Return(int64(1), nil)
-	_, err := m.mgr.Create(context.Background(), &model.Job{})
-	m.Nil(err)
-	m.dao.AssertExpectations(m.T())
-}
-
-func (m *managerTestSuite) TestUpdate() {
-	m.dao.On("Update", mock.Anything, mock.Anything).Return(nil)
-	err := m.mgr.Update(context.Background(), &model.Job{})
-	m.Nil(err)
-	m.dao.AssertExpectations(m.T())
-}
-
-func (m *managerTestSuite) TestCount() {
-	m.dao.On("Count", mock.Anything, mock.Anything).Return(int64(1), nil)
-	n, err := m.mgr.Count(context.Background(), nil)
-	m.Nil(err)
-	m.Equal(int64(1), n)
-	m.dao.AssertExpectations(m.T())
-}
-
-func (m *managerTestSuite) TestList() {
-	m.dao.On("List", mock.Anything, mock.Anything).Return([]*model.Job{
-		{
-			ID:        1,
-			EventType: "test_job",
-		},
-	}, nil)
-	rpers, err := m.mgr.List(context.Background(), nil)
-	m.Nil(err)
-	m.Equal(1, len(rpers))
-	m.dao.AssertExpectations(m.T())
-}
-
-func (m *managerTestSuite) TestListJobsGroupByEventType() {
-	m.dao.On("GetLastTriggerJobsGroupByEventType", mock.Anything, mock.Anything).Return([]*model.Job{
-		{
-			ID:        1,
-			EventType: "test_job",
-			PolicyID:  1,
-		},
-		{
-			ID:        2,
-			EventType: "test_job",
-			PolicyID:  1,
-		},
-	}, nil)
-	rpers, err := m.mgr.ListJobsGroupByEventType(context.Background(), 1)
-	m.Nil(err)
-	m.Equal(2, len(rpers))
-	m.dao.AssertExpectations(m.T())
-}
-
-func TestManager(t *testing.T) {
-	suite.Run(t, &managerTestSuite{})
-}

src/pkg/notification/job/model/model.go

@@ -1,29 +0,0 @@
-package model
-
-import (
-	"time"
-
-	"github.com/beego/beego/v2/client/orm"
-)
-
-func init() {
-	orm.RegisterModel(&Job{})
-}
-
-// Job is the model for a notification job
-type Job struct {
-	ID           int64     `orm:"pk;auto;column(id)" json:"id"`
-	PolicyID     int64     `orm:"column(policy_id)" json:"policy_id"`
-	EventType    string    `orm:"column(event_type)" json:"event_type"`
-	NotifyType   string    `orm:"column(notify_type)" json:"notify_type"`
-	Status       string    `orm:"column(status)" json:"status"`
-	JobDetail    string    `orm:"column(job_detail)" json:"job_detail"`
-	UUID         string    `orm:"column(job_uuid)" json:"-"`
-	CreationTime time.Time `orm:"column(creation_time);auto_now_add" json:"creation_time"`
-	UpdateTime   time.Time `orm:"column(update_time);auto_now" json:"update_time" sort:"default:desc"`
-}
-
-// TableName set table name for ORM.
-func (j *Job) TableName() string {
-	return "notification_job"
-}

src/pkg/notification/notification.go

@@ -7,27 +7,38 @@ import (
 	"github.com/goharbor/harbor/src/controller/event"
 	"github.com/goharbor/harbor/src/lib/log"
 	"github.com/goharbor/harbor/src/pkg/notification/hook"
-	"github.com/goharbor/harbor/src/pkg/notification/job"
 	"github.com/goharbor/harbor/src/pkg/notification/policy"
 	n_event "github.com/goharbor/harbor/src/pkg/notifier/event"
 	notifier_model "github.com/goharbor/harbor/src/pkg/notifier/model"
 )
 
+type (
+	// EventType is the type of event
+	EventType string
+	// NotifyType is the type of notify
+	NotifyType string
+)
+
+func (e EventType) String() string {
+	return string(e)
+}
+
+func (n NotifyType) String() string {
+	return string(n)
+}
+
 var (
 	// PolicyMgr is a global notification policy manager
 	PolicyMgr policy.Manager
 
-	// JobMgr is a notification job controller
-	JobMgr job.Manager
-
 	// HookManager is a hook manager
 	HookManager hook.Manager
 
-	// SupportedEventTypes is a map to store supported event type, eg. pushImage, pullImage etc
-	SupportedEventTypes map[string]struct{}
+	// supportedEventTypes is a slice to store supported event type, eg. pushImage, pullImage etc
+	supportedEventTypes []EventType
 
-	// SupportedNotifyTypes is a map to store notification type, eg. HTTP, Email etc
-	SupportedNotifyTypes map[string]struct{}
+	// supportedNotifyTypes is a slice to store notification type, eg. HTTP, Email etc
+	supportedNotifyTypes []NotifyType
 )
 
 // Init ...
@@ -36,8 +47,6 @@ func Init() {
 	PolicyMgr = policy.Mgr
 	// init hook manager
 	HookManager = hook.NewHookManager()
-	// init notification job manager
-	JobMgr = job.Mgr
 
 	initSupportedNotifyType()
 
@@ -45,8 +54,8 @@ func Init() {
 }
 
 func initSupportedNotifyType() {
-	SupportedEventTypes = make(map[string]struct{}, 0)
-	SupportedNotifyTypes = make(map[string]struct{}, 0)
+	supportedEventTypes = make([]EventType, 0)
+	supportedNotifyTypes = make([]NotifyType, 0)
 
 	eventTypes := []string{
 		event.TopicPushArtifact,
@@ -61,12 +70,12 @@ func initSupportedNotifyType() {
 		event.TopicTagRetention,
 	}
 	for _, eventType := range eventTypes {
-		SupportedEventTypes[eventType] = struct{}{}
+		supportedEventTypes = append(supportedEventTypes, EventType(eventType))
 	}
 
 	notifyTypes := []string{notifier_model.NotifyTypeHTTP, notifier_model.NotifyTypeSlack}
 	for _, notifyType := range notifyTypes {
-		SupportedNotifyTypes[notifyType] = struct{}{}
+		supportedNotifyTypes = append(supportedNotifyTypes, NotifyType(notifyType))
 	}
 }
 
@@ -110,3 +119,11 @@ func AddEvent(ctx context.Context, m n_event.Metadata, notify ...bool) {
 	}
 	e.Events.PushBack(m)
 }
+
+func GetSupportedEventTypes() []EventType {
+	return supportedEventTypes
+}
+
+func GetSupportedNotifyTypes() []NotifyType {
+	return supportedNotifyTypes
+}

src/pkg/notification/policy/manager.go

@@ -3,16 +3,11 @@ package policy
 import (
 	"context"
 	"fmt"
-	"net/http"
 	"time"
 
-	commonhttp "github.com/goharbor/harbor/src/common/http"
-	"github.com/goharbor/harbor/src/lib/errors"
-	"github.com/goharbor/harbor/src/lib/log"
 	"github.com/goharbor/harbor/src/lib/q"
 	"github.com/goharbor/harbor/src/pkg/notification/policy/dao"
 	"github.com/goharbor/harbor/src/pkg/notification/policy/model"
-	notifier_model "github.com/goharbor/harbor/src/pkg/notifier/model"
 )
 
 var (
@@ -30,14 +25,10 @@ type Manager interface {
 	Count(ctx context.Context, query *q.Query) (int64, error)
 	// Get policy with specified ID
 	Get(ctx context.Context, id int64) (*model.Policy, error)
-	// GetByNameAndProjectID get policy by the name and projectID
-	GetByNameAndProjectID(ctx context.Context, name string, projectID int64) (*model.Policy, error)
 	// Update the specified policy
 	Update(ctx context.Context, policy *model.Policy) error
 	// Delete the specified policy
 	Delete(ctx context.Context, policyID int64) error
-	// Test the specified policy
-	Test(policy *model.Policy) error
 	// GetRelatedPolices get event type related policies in project
 	GetRelatedPolices(ctx context.Context, projectID int64, eventType string) ([]*model.Policy, error)
 }
@@ -107,23 +98,6 @@ func (m *manager) Get(ctx context.Context, id int64) (*model.Policy, error) {
 	return policy, err
 }
 
-// GetByNameAndProjectID notification policy by the name and projectID
-func (m *manager) GetByNameAndProjectID(ctx context.Context, name string, projectID int64) (*model.Policy, error) {
-	query := q.New(q.KeyWords{"name": name, "project_id": projectID})
-	policies, err := m.dao.List(ctx, query)
-	if err != nil {
-		return nil, err
-	}
-	if len(policies) == 0 {
-		return nil, errors.New(nil).WithCode(errors.NotFoundCode).WithMessage("no notification policy found")
-	}
-	policy := policies[0]
-	if err := policy.ConvertFromDBModel(); err != nil {
-		return nil, err
-	}
-	return policy, err
-}
-
 // Update the specified notification policy
 func (m *manager) Update(ctx context.Context, policy *model.Policy) error {
 	policy.UpdateTime = time.Now()
@@ -139,40 +113,6 @@ func (m *manager) Delete(ctx context.Context, policyID int64) error {
 	return m.dao.Delete(ctx, policyID)
 }
 
-// Test the specified notification policy, just test for network connection without request body
-func (m *manager) Test(policy *model.Policy) error {
-	for _, target := range policy.Targets {
-		switch target.Type {
-		case notifier_model.NotifyTypeHTTP, notifier_model.NotifyTypeSlack:
-			return m.policyHTTPTest(target.Address, target.SkipCertVerify)
-		default:
-			return fmt.Errorf("invalid policy target type: %s", target.Type)
-		}
-	}
-	return nil
-}
-
-func (m *manager) policyHTTPTest(address string, skipCertVerify bool) error {
-	req, err := http.NewRequest(http.MethodPost, address, nil)
-	if err != nil {
-		return err
-	}
-
-	req.Header.Set("Content-Type", "application/json")
-	client := http.Client{
-		Transport: commonhttp.GetHTTPTransport(commonhttp.WithInsecure(skipCertVerify)),
-	}
-
-	resp, err := client.Do(req)
-	if err != nil {
-		return err
-	}
-	defer resp.Body.Close()
-	log.Debugf("policy test success with address %s, skip cert verify :%v", address, skipCertVerify)
-
-	return nil
-}
-
 // GetRelatedPolices get policies including event type in project
 func (m *manager) GetRelatedPolices(ctx context.Context, projectID int64, eventType string) ([]*model.Policy, error) {
 	policies, err := m.List(ctx, q.New(q.KeyWords{"project_id": projectID}))

src/pkg/notification/policy/manager_test.go

@@ -6,7 +6,6 @@ import (
 
 	"github.com/stretchr/testify/suite"
 
-	"github.com/goharbor/harbor/src/lib/errors"
 	"github.com/goharbor/harbor/src/pkg/notification/policy/model"
 	"github.com/goharbor/harbor/src/testing/mock"
 	"github.com/goharbor/harbor/src/testing/pkg/notification/policy/dao"
@@ -68,28 +67,6 @@ func (m *managerTestSuite) TestList() {
 	m.Equal(1, len(rpers))
 	m.dao.AssertExpectations(m.T())
 }
-
-func (m *managerTestSuite) TestGetByNameAndProjectID404() {
-	m.dao.On("List", mock.Anything, mock.Anything).Return([]*model.Policy{}, nil)
-	_, err := m.mgr.GetByNameAndProjectID(context.Background(), "test_policy", 1)
-	m.NotNil(err)
-	m.True(errors.IsNotFoundErr(err))
-}
-
-func (m *managerTestSuite) TestGetByNameAndProjectID() {
-	m.dao.On("List", mock.Anything, mock.Anything).Return([]*model.Policy{
-		{
-			ID:        1,
-			Name:      "test_policy",
-			ProjectID: 1,
-		},
-	}, nil)
-	policy, err := m.mgr.GetByNameAndProjectID(context.Background(), "test_policy", 1)
-	m.Nil(err)
-	m.Equal("test_policy", policy.Name)
-	m.dao.AssertExpectations(m.T())
-}
-
 func (m *managerTestSuite) TestGetRelatedPolices() {
 	m.dao.On("List", mock.Anything, mock.Anything).Return([]*model.Policy{
 		{

src/server/route.go

@@ -22,7 +22,6 @@ import (
 	"github.com/goharbor/harbor/src/common"
 	"github.com/goharbor/harbor/src/core/api"
 	"github.com/goharbor/harbor/src/core/controllers"
-	"github.com/goharbor/harbor/src/core/service/notifications/jobs"
 	"github.com/goharbor/harbor/src/core/service/token"
 	"github.com/goharbor/harbor/src/server/handler"
 	"github.com/goharbor/harbor/src/server/router"
@@ -48,7 +47,6 @@ func registerRoutes() {
 	web.Router("/api/internal/renameadmin", &api.InternalAPI{}, "post:RenameAdmin")
 	web.Router("/api/internal/syncquota", &api.InternalAPI{}, "post:SyncQuota")
 
-	web.Router("/service/notifications/jobs/webhook/:id([0-9]+)", &jobs.Handler{}, "post:HandleNotificationJob")
 	router.NewRoute().Method(http.MethodPost).Path("/service/notifications/jobs/adminjob/:id([0-9]+)").Handler(handler.NewJobStatusHandler())         // legacy job status hook endpoint for adminjob
 	router.NewRoute().Method(http.MethodPost).Path("/service/notifications/jobs/scan/:uuid").HandlerFunc(ignoreNotification)                          // ignore legacy scan job notifaction
 	router.NewRoute().Method(http.MethodPost).Path("/service/notifications/schedules/:id([0-9]+)").Handler(handler.NewJobStatusHandler())             // legacy job status hook endpoint for scheduler

src/server/v2.0/handler/handler.go

@@ -54,8 +54,8 @@ func New() http.Handler {
 		GCAPI:                 newGCAPI(),
 		QuotaAPI:              newQuotaAPI(),
 		RetentionAPI:          newRetentionAPI(),
-		WebhookAPI:            newNotificationPolicyAPI(),
-		WebhookjobAPI:         newNotificationJobAPI(),
+		WebhookAPI:            newWebhookAPI(),
+		WebhookjobAPI:         newWebhookJobAPI(),
 		ImmutableAPI:          newImmutableAPI(),
 		OIDCAPI:               newOIDCAPI(),
 		SystemCVEAllowlistAPI: newSystemCVEAllowListAPI(),

src/server/v2.0/handler/model/notification_job.go

@@ -1,34 +0,0 @@
-package model
-
-import (
-	"github.com/go-openapi/strfmt"
-
-	"github.com/goharbor/harbor/src/pkg/notification/job/model"
-	"github.com/goharbor/harbor/src/server/v2.0/models"
-)
-
-// NotificationJob ...
-type NotificationJob struct {
-	*model.Job
-}
-
-// ToSwagger ...
-func (n *NotificationJob) ToSwagger() *models.WebhookJob {
-	return &models.WebhookJob{
-		ID:           n.ID,
-		EventType:    n.EventType,
-		JobDetail:    n.JobDetail,
-		NotifyType:   n.NotifyType,
-		PolicyID:     n.PolicyID,
-		Status:       n.Status,
-		CreationTime: strfmt.DateTime(n.CreationTime),
-		UpdateTime:   strfmt.DateTime(n.UpdateTime),
-	}
-}
-
-// NewNotificationJob ...
-func NewNotificationJob(j *model.Job) *NotificationJob {
-	return &NotificationJob{
-		Job: j,
-	}
-}

src/server/v2.0/handler/model/webhook_job.go

@@ -0,0 +1,58 @@
+package model
+
+import (
+	"encoding/json"
+
+	"github.com/go-openapi/strfmt"
+
+	"github.com/goharbor/harbor/src/jobservice/job"
+	"github.com/goharbor/harbor/src/lib/log"
+	"github.com/goharbor/harbor/src/pkg/task"
+	"github.com/goharbor/harbor/src/server/v2.0/models"
+)
+
+// WebhookJob ...
+type WebhookJob struct {
+	*task.Execution
+}
+
+// ToSwagger ...
+func (n *WebhookJob) ToSwagger() *models.WebhookJob {
+	webhookJob := &models.WebhookJob{
+		ID:           n.ID,
+		PolicyID:     n.VendorID,
+		Status:       n.Status,
+		CreationTime: strfmt.DateTime(n.StartTime),
+		UpdateTime:   strfmt.DateTime(n.UpdateTime),
+	}
+
+	var notifyType string
+	// do the conversion for compatible with old API
+	if n.VendorType == job.WebhookJobVendorType {
+		notifyType = "http"
+	} else if n.VendorType == job.SlackJobVendorType {
+		notifyType = "slack"
+	}
+	webhookJob.NotifyType = notifyType
+
+	if n.ExtraAttrs != nil {
+		if eventType, ok := n.ExtraAttrs["type"].(string); ok {
+			webhookJob.EventType = eventType
+		}
+		detail, err := json.Marshal(n.ExtraAttrs)
+		if err == nil {
+			webhookJob.JobDetail = string(detail)
+		} else {
+			log.Errorf("failed to marshal exec.ExtraAttrs, error: %v", err)
+		}
+	}
+
+	return webhookJob
+}
+
+// NewWebhookJob ...
+func NewWebhookJob(exec *task.Execution) *WebhookJob {
+	return &WebhookJob{
+		Execution: exec,
+	}
+}

src/server/v2.0/handler/model/webhook_policy.go

@@ -7,13 +7,13 @@ import (
 	"github.com/goharbor/harbor/src/server/v2.0/models"
 )
 
-// NotifiactionPolicy ...
-type NotifiactionPolicy struct {
+// WebhookPolicy ...
+type WebhookPolicy struct {
 	*model.Policy
 }
 
 // ToSwagger ...
-func (n *NotifiactionPolicy) ToSwagger() *models.WebhookPolicy {
+func (n *WebhookPolicy) ToSwagger() *models.WebhookPolicy {
 	return &models.WebhookPolicy{
 		ID:           n.ID,
 		CreationTime: strfmt.DateTime(n.CreationTime),
@@ -29,7 +29,7 @@ func (n *NotifiactionPolicy) ToSwagger() *models.WebhookPolicy {
 }
 
 // ToTargets ...
-func (n *NotifiactionPolicy) ToTargets() []*models.WebhookTargetObject {
+func (n *WebhookPolicy) ToTargets() []*models.WebhookTargetObject {
 	var results []*models.WebhookTargetObject
 	for _, t := range n.Targets {
 		results = append(results, &models.WebhookTargetObject{
@@ -43,8 +43,8 @@ func (n *NotifiactionPolicy) ToTargets() []*models.WebhookTargetObject {
 }
 
 // NewNotifiactionPolicy ...
-func NewNotifiactionPolicy(p *model.Policy) *NotifiactionPolicy {
-	return &NotifiactionPolicy{
+func NewWebhookPolicy(p *model.Policy) *WebhookPolicy {
+	return &WebhookPolicy{
 		Policy: p,
 	}
 }

src/server/v2.0/handler/webhook.go

@@ -18,59 +18,89 @@ import (
 	"context"
 	"fmt"
 	"strings"
-	"time"
 
 	"github.com/go-openapi/runtime/middleware"
 	"github.com/go-openapi/strfmt"
 
 	"github.com/goharbor/harbor/src/common/rbac"
 	"github.com/goharbor/harbor/src/common/utils"
+	"github.com/goharbor/harbor/src/controller/task"
+	webhook_ctl "github.com/goharbor/harbor/src/controller/webhook"
+	"github.com/goharbor/harbor/src/jobservice/job"
 	"github.com/goharbor/harbor/src/lib"
 	"github.com/goharbor/harbor/src/lib/errors"
 	"github.com/goharbor/harbor/src/lib/log"
 	"github.com/goharbor/harbor/src/lib/q"
 	"github.com/goharbor/harbor/src/pkg/notification"
-	"github.com/goharbor/harbor/src/pkg/notification/job"
-	"github.com/goharbor/harbor/src/pkg/notification/policy"
 	policy_model "github.com/goharbor/harbor/src/pkg/notification/policy/model"
 	"github.com/goharbor/harbor/src/server/v2.0/handler/model"
 	"github.com/goharbor/harbor/src/server/v2.0/models"
 	"github.com/goharbor/harbor/src/server/v2.0/restapi/operations/webhook"
 )
 
-func newNotificationPolicyAPI() *notificationPolicyAPI {
-	return &notificationPolicyAPI{
-		webhookjobMgr:    job.Mgr,
-		webhookPolicyMgr: policy.Mgr,
+func newWebhookAPI() *webhookAPI {
+	return &webhookAPI{
+		execCtl:    task.ExecutionCtl,
+		taskCtl:    task.Ctl,
+		webhookCtl: webhook_ctl.Ctl,
 	}
 }
 
-type notificationPolicyAPI struct {
+type webhookAPI struct {
 	BaseAPI
-	webhookjobMgr    job.Manager
-	webhookPolicyMgr policy.Manager
+	execCtl    task.ExecutionController
+	taskCtl    task.Controller
+	webhookCtl webhook_ctl.Controller
 }
 
-func (n *notificationPolicyAPI) Prepare(ctx context.Context, operation string, params interface{}) middleware.Responder {
+func (n *webhookAPI) Prepare(ctx context.Context, operation string, params interface{}) middleware.Responder {
 	return nil
 }
 
-func (n *notificationPolicyAPI) requirePolicyInProject(ctx context.Context, projectIDOrName interface{}, policyID int64) error {
+func (n *webhookAPI) requirePolicyInProject(ctx context.Context, projectIDOrName interface{}, policyID int64) error {
 	projectID, err := getProjectID(ctx, projectIDOrName)
 	if err != nil {
 		return err
 	}
-	l, err := n.webhookPolicyMgr.Get(ctx, policyID)
+
+	l, err := n.webhookCtl.GetPolicy(ctx, policyID)
 	if err != nil {
 		return err
 	}
+
 	if projectID != l.ProjectID {
 		return errors.NotFoundError(fmt.Errorf("project id:%d, webhook policy id: %d not found", projectID, policyID))
 	}
 	return nil
 }
 
-func (n *notificationPolicyAPI) ListWebhookPoliciesOfProject(ctx context.Context, params webhook.ListWebhookPoliciesOfProjectParams) middleware.Responder {
+func (n *webhookAPI) requireExecutionInPolicy(ctx context.Context, execID, policyID int64) error {
+	exec, err := n.execCtl.Get(ctx, execID)
+	if err != nil {
+		return err
+	}
+
+	if exec.VendorID == policyID && (exec.VendorType == job.WebhookJobVendorType || exec.VendorType == job.SlackJobVendorType) {
+		return nil
+	}
+
+	return errors.NotFoundError(fmt.Errorf("execution %d not found in policy %d", execID, policyID))
+}
+
+func (n *webhookAPI) requireTaskInExecution(ctx context.Context, taskID, execID int64) error {
+	task, err := n.taskCtl.Get(ctx, taskID)
+	if err != nil {
+		return err
+	}
+
+	if task.ExecutionID == execID {
+		return nil
+	}
+
+	return errors.NotFoundError(fmt.Errorf("task %d not found in execution %d", taskID, execID))
+}
+
+func (n *webhookAPI) ListWebhookPoliciesOfProject(ctx context.Context, params webhook.ListWebhookPoliciesOfProjectParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	if err := n.RequireProjectAccess(ctx, projectNameOrID, rbac.ActionList, rbac.ResourceNotificationPolicy); err != nil {
 		return n.SendError(ctx, err)
@@ -87,18 +117,18 @@ func (n *notificationPolicyAPI) ListWebhookPoliciesOfProject(ctx context.Context
 	}
 	query.Keywords["ProjectID"] = projectID
 
-	total, err := n.webhookPolicyMgr.Count(ctx, query)
+	total, err := n.webhookCtl.CountPolicies(ctx, query)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
 
-	policies, err := n.webhookPolicyMgr.List(ctx, query)
+	policies, err := n.webhookCtl.ListPolicies(ctx, query)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
 	var results []*models.WebhookPolicy
 	for _, p := range policies {
-		results = append(results, model.NewNotifiactionPolicy(p).ToSwagger())
+		results = append(results, model.NewWebhookPolicy(p).ToSwagger())
 	}
 
 	return webhook.NewListWebhookPoliciesOfProjectOK().
@@ -107,7 +137,7 @@ func (n *notificationPolicyAPI) ListWebhookPoliciesOfProject(ctx context.Context
 		WithPayload(results)
 }
 
-func (n *notificationPolicyAPI) CreateWebhookPolicyOfProject(ctx context.Context, params webhook.CreateWebhookPolicyOfProjectParams) middleware.Responder {
+func (n *webhookAPI) CreateWebhookPolicyOfProject(ctx context.Context, params webhook.CreateWebhookPolicyOfProjectParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	if err := n.RequireProjectAccess(ctx, projectNameOrID, rbac.ActionCreate, rbac.ResourceNotificationPolicy); err != nil {
 		return n.SendError(ctx, err)
@@ -130,7 +160,7 @@ func (n *notificationPolicyAPI) CreateWebhookPolicyOfProject(ctx context.Context
 		return n.SendError(ctx, err)
 	}
 	policy.ProjectID = projectID
-	id, err := n.webhookPolicyMgr.Create(ctx, policy)
+	id, err := n.webhookCtl.CreatePolicy(ctx, policy)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
@@ -139,7 +169,7 @@ func (n *notificationPolicyAPI) CreateWebhookPolicyOfProject(ctx context.Context
 	return webhook.NewCreateWebhookPolicyOfProjectCreated().WithLocation(location)
 }
 
-func (n *notificationPolicyAPI) UpdateWebhookPolicyOfProject(ctx context.Context, params webhook.UpdateWebhookPolicyOfProjectParams) middleware.Responder {
+func (n *webhookAPI) UpdateWebhookPolicyOfProject(ctx context.Context, params webhook.UpdateWebhookPolicyOfProjectParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	if err := n.RequireProjectAccess(ctx, projectNameOrID, rbac.ActionUpdate, rbac.ResourceNotificationPolicy); err != nil {
 		return n.SendError(ctx, err)
@@ -166,14 +196,14 @@ func (n *notificationPolicyAPI) UpdateWebhookPolicyOfProject(ctx context.Context
 
 	policy.ID = policyID
 	policy.ProjectID = projectID
-	if err := n.webhookPolicyMgr.Update(ctx, policy); err != nil {
+	if err := n.webhookCtl.UpdatePolicy(ctx, policy); err != nil {
 		return n.SendError(ctx, err)
 	}
 
 	return webhook.NewUpdateWebhookPolicyOfProjectOK()
 }
 
-func (n *notificationPolicyAPI) DeleteWebhookPolicyOfProject(ctx context.Context, params webhook.DeleteWebhookPolicyOfProjectParams) middleware.Responder {
+func (n *webhookAPI) DeleteWebhookPolicyOfProject(ctx context.Context, params webhook.DeleteWebhookPolicyOfProjectParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	if err := n.RequireProjectAccess(ctx, projectNameOrID, rbac.ActionDelete, rbac.ResourceNotificationPolicy); err != nil {
 		return n.SendError(ctx, err)
@@ -181,13 +211,13 @@ func (n *notificationPolicyAPI) DeleteWebhookPolicyOfProject(ctx context.Context
 	if err := n.requirePolicyInProject(ctx, projectNameOrID, params.WebhookPolicyID); err != nil {
 		return n.SendError(ctx, err)
 	}
-	if err := n.webhookPolicyMgr.Delete(ctx, params.WebhookPolicyID); err != nil {
+	if err := n.webhookCtl.DeletePolicy(ctx, params.WebhookPolicyID); err != nil {
 		return n.SendError(ctx, err)
 	}
 	return webhook.NewDeleteWebhookPolicyOfProjectOK()
 }
 
-func (n *notificationPolicyAPI) GetWebhookPolicyOfProject(ctx context.Context, params webhook.GetWebhookPolicyOfProjectParams) middleware.Responder {
+func (n *webhookAPI) GetWebhookPolicyOfProject(ctx context.Context, params webhook.GetWebhookPolicyOfProjectParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	projectID, err := getProjectID(ctx, projectNameOrID)
 	if err != nil {
@@ -200,15 +230,127 @@ func (n *notificationPolicyAPI) GetWebhookPolicyOfProject(ctx context.Context, p
 		return n.SendError(ctx, err)
 	}
 
-	policy, err := n.webhookPolicyMgr.Get(ctx, params.WebhookPolicyID)
+	policy, err := n.webhookCtl.GetPolicy(ctx, params.WebhookPolicyID)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
 
-	return webhook.NewGetWebhookPolicyOfProjectOK().WithPayload(model.NewNotifiactionPolicy(policy).ToSwagger())
+	return webhook.NewGetWebhookPolicyOfProjectOK().WithPayload(model.NewWebhookPolicy(policy).ToSwagger())
 }
 
-func (n *notificationPolicyAPI) LastTrigger(ctx context.Context, params webhook.LastTriggerParams) middleware.Responder {
+func (n *webhookAPI) ListExecutionsOfWebhookPolicy(ctx context.Context, params webhook.ListExecutionsOfWebhookPolicyParams) middleware.Responder {
+	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
+	projectID, err := getProjectID(ctx, projectNameOrID)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.RequireProjectAccess(ctx, projectID, rbac.ActionRead, rbac.ResourceNotificationPolicy); err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.requirePolicyInProject(ctx, projectID, params.WebhookPolicyID); err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	query, err := n.BuildQuery(ctx, params.Q, params.Sort, params.Page, params.PageSize)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	total, err := n.webhookCtl.CountExecutions(ctx, params.WebhookPolicyID, query)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	executions, err := n.webhookCtl.ListExecutions(ctx, params.WebhookPolicyID, query)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	var payloads []*models.Execution
+	for _, exec := range executions {
+		p, err := convertExecutionToPayload(exec)
+		if err != nil {
+			return n.SendError(ctx, err)
+		}
+		payloads = append(payloads, p)
+	}
+
+	return webhook.NewListExecutionsOfWebhookPolicyOK().WithPayload(payloads).WithXTotalCount(total).
+		WithLink(n.Links(ctx, params.HTTPRequest.URL, total, query.PageNumber, query.PageSize).String())
+}
+
+func (n *webhookAPI) ListTasksOfWebhookExecution(ctx context.Context, params webhook.ListTasksOfWebhookExecutionParams) middleware.Responder {
+	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
+	projectID, err := getProjectID(ctx, projectNameOrID)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.RequireProjectAccess(ctx, projectID, rbac.ActionRead, rbac.ResourceNotificationPolicy); err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.requirePolicyInProject(ctx, projectID, params.WebhookPolicyID); err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.requireExecutionInPolicy(ctx, params.ExecutionID, params.WebhookPolicyID); err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	query, err := n.BuildQuery(ctx, params.Q, params.Sort, params.Page, params.PageSize)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	total, err := n.webhookCtl.CountTasks(ctx, params.ExecutionID, query)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	tasks, err := n.webhookCtl.ListTasks(ctx, params.ExecutionID, query)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	var payloads []*models.Task
+	for _, task := range tasks {
+		p, err := convertTaskToPayload(task)
+		if err != nil {
+			return n.SendError(ctx, err)
+		}
+		payloads = append(payloads, p)
+	}
+
+	return webhook.NewListTasksOfWebhookExecutionOK().WithPayload(payloads).WithXTotalCount(total).
+		WithLink(n.Links(ctx, params.HTTPRequest.URL, total, query.PageNumber, query.PageSize).String())
+}
+
+func (n *webhookAPI) GetLogsOfWebhookTask(ctx context.Context, params webhook.GetLogsOfWebhookTaskParams) middleware.Responder {
+	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
+	projectID, err := getProjectID(ctx, projectNameOrID)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.RequireProjectAccess(ctx, projectID, rbac.ActionRead, rbac.ResourceNotificationPolicy); err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.requirePolicyInProject(ctx, projectID, params.WebhookPolicyID); err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.requireExecutionInPolicy(ctx, params.ExecutionID, params.WebhookPolicyID); err != nil {
+		return n.SendError(ctx, err)
+	}
+	if err := n.requireTaskInExecution(ctx, params.TaskID, params.ExecutionID); err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	l, err := n.webhookCtl.GetTaskLog(ctx, params.TaskID)
+	if err != nil {
+		return n.SendError(ctx, err)
+	}
+
+	return webhook.NewGetLogsOfWebhookTaskOK().WithPayload(string(l))
+}
+
+func (n *webhookAPI) LastTrigger(ctx context.Context, params webhook.LastTriggerParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	if err := n.RequireProjectAccess(ctx, projectNameOrID, rbac.ActionRead, rbac.ResourceNotificationPolicy); err != nil {
 		return n.SendError(ctx, err)
@@ -224,7 +366,7 @@ func (n *notificationPolicyAPI) LastTrigger(ctx context.Context, params webhook.
 			"ProjectID": projectID,
 		},
 	}
-	policies, err := n.webhookPolicyMgr.List(ctx, query)
+	policies, err := n.webhookCtl.ListPolicies(ctx, query)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
@@ -236,39 +378,25 @@ func (n *notificationPolicyAPI) LastTrigger(ctx context.Context, params webhook.
 	return webhook.NewLastTriggerOK().WithPayload(triggers)
 }
 
-func (n *notificationPolicyAPI) GetSupportedEventTypes(ctx context.Context, params webhook.GetSupportedEventTypesParams) middleware.Responder {
+func (n *webhookAPI) GetSupportedEventTypes(ctx context.Context, params webhook.GetSupportedEventTypesParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	if err := n.RequireProjectAccess(ctx, projectNameOrID, rbac.ActionRead, rbac.ResourceNotificationPolicy); err != nil {
 		return n.SendError(ctx, err)
 	}
 
 	var notificationTypes = &models.SupportedWebhookEventTypes{}
-	for key := range notification.SupportedNotifyTypes {
-		notificationTypes.NotifyType = append(notificationTypes.NotifyType, models.NotifyType(key))
+	for _, notifyType := range notification.GetSupportedNotifyTypes() {
+		notificationTypes.NotifyType = append(notificationTypes.NotifyType, models.NotifyType(notifyType))
 	}
 
-	for key := range notification.SupportedEventTypes {
-		notificationTypes.EventType = append(notificationTypes.EventType, models.EventType(key))
+	for _, eventType := range notification.GetSupportedEventTypes() {
+		notificationTypes.EventType = append(notificationTypes.EventType, models.EventType(eventType))
 	}
 
 	return webhook.NewGetSupportedEventTypesOK().WithPayload(notificationTypes)
 }
 
-func (n *notificationPolicyAPI) getLastTriggerTimeGroupByEventType(ctx context.Context, eventType string, policyID int64) (time.Time, error) {
-	jobs, err := n.webhookjobMgr.ListJobsGroupByEventType(ctx, policyID)
-	if err != nil {
-		return time.Time{}, err
-	}
-
-	for _, job := range jobs {
-		if eventType == job.EventType {
-			return job.CreationTime, nil
-		}
-	}
-	return time.Time{}, nil
-}
-
-func (n *notificationPolicyAPI) validateTargets(policy *policy_model.Policy) (bool, error) {
+func (n *webhookAPI) validateTargets(policy *policy_model.Policy) (bool, error) {
 	if len(policy.Targets) == 0 {
 		return false, errors.New(nil).WithMessage("empty notification target with policy %s", policy.Name).WithCode(errors.BadRequestCode)
 	}
@@ -280,21 +408,19 @@ func (n *notificationPolicyAPI) validateTargets(policy *policy_model.Policy) (bo
 		// Prevent SSRF security issue #3755
 		target.Address = url.Scheme + "://" + url.Host + url.Path
 
-		_, ok := notification.SupportedNotifyTypes[target.Type]
-		if !ok {
+		if !isNotifyTypeSupported(target.Type) {
 			return false, errors.New(nil).WithMessage("unsupported target type %s with policy %s", target.Type, policy.Name).WithCode(errors.BadRequestCode)
 		}
 	}
 	return true, nil
 }
 
-func (n *notificationPolicyAPI) validateEventTypes(policy *policy_model.Policy) (bool, error) {
+func (n *webhookAPI) validateEventTypes(policy *policy_model.Policy) (bool, error) {
 	if len(policy.EventTypes) == 0 {
 		return false, errors.New(nil).WithMessage("empty event type").WithCode(errors.BadRequestCode)
 	}
 	for _, eventType := range policy.EventTypes {
-		_, ok := notification.SupportedEventTypes[eventType]
-		if !ok {
+		if !isEventTypeSupported(eventType) {
 			return false, errors.New(nil).WithMessage("unsupported event type %s", eventType).WithCode(errors.BadRequestCode)
 		}
 	}
@@ -303,7 +429,7 @@ func (n *notificationPolicyAPI) validateEventTypes(policy *policy_model.Policy)
 
 // constructPolicyWithTriggerTime construct notification policy information displayed in UI
 // including event type, enabled, creation time, last trigger time
-func (n *notificationPolicyAPI) constructPolicyWithTriggerTime(ctx context.Context, policies []*policy_model.Policy) ([]*models.WebhookLastTrigger, error) {
+func (n *webhookAPI) constructPolicyWithTriggerTime(ctx context.Context, policies []*policy_model.Policy) ([]*models.WebhookLastTrigger, error) {
 	res := []*models.WebhookLastTrigger{}
 	for _, policy := range policies {
 		for _, t := range policy.EventTypes {
@@ -317,7 +443,7 @@ func (n *notificationPolicyAPI) constructPolicyWithTriggerTime(ctx context.Conte
 				ply.CreationTime = strfmt.DateTime(policy.CreationTime)
 			}
 
-			ltTime, err := n.getLastTriggerTimeGroupByEventType(ctx, t, policy.ID)
+			ltTime, err := n.webhookCtl.GetLastTriggerTime(ctx, t, policy.ID)
 			if err != nil {
 				return nil, err
 			}
@@ -329,3 +455,23 @@ func (n *notificationPolicyAPI) constructPolicyWithTriggerTime(ctx context.Conte
 	}
 	return res, nil
 }
+
+func isEventTypeSupported(eventType string) bool {
+	for _, t := range notification.GetSupportedEventTypes() {
+		if t.String() == eventType {
+			return true
+		}
+	}
+
+	return false
+}
+
+func isNotifyTypeSupported(notifyType string) bool {
+	for _, t := range notification.GetSupportedNotifyTypes() {
+		if t.String() == notifyType {
+			return true
+		}
+	}
+
+	return false
+}

src/server/v2.0/handler/webhook_job.go

@@ -1,3 +1,17 @@
+//  Copyright Project Harbor Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
 package handler
 
 import (
@@ -6,10 +20,9 @@ import (
 	"github.com/go-openapi/runtime/middleware"
 
 	"github.com/goharbor/harbor/src/common/rbac"
+	webhook_ctl "github.com/goharbor/harbor/src/controller/webhook"
 	"github.com/goharbor/harbor/src/lib/errors"
 	"github.com/goharbor/harbor/src/pkg"
-	"github.com/goharbor/harbor/src/pkg/notification/job"
-	"github.com/goharbor/harbor/src/pkg/notification/policy"
 	policyModel "github.com/goharbor/harbor/src/pkg/notification/policy/model"
 	"github.com/goharbor/harbor/src/pkg/project"
 	"github.com/goharbor/harbor/src/server/v2.0/handler/model"
@@ -17,28 +30,26 @@ import (
 	"github.com/goharbor/harbor/src/server/v2.0/restapi/operations/webhookjob"
 )
 
-func newNotificationJobAPI() *notificationJobAPI {
-	return &notificationJobAPI{
-		webhookjobMgr:    job.Mgr,
-		webhookPolicyMgr: policy.Mgr,
-		projectMgr:       pkg.ProjectMgr,
+func newWebhookJobAPI() *webhookJobAPI {
+	return &webhookJobAPI{
+		webhookCtl: webhook_ctl.Ctl,
+		projectMgr: pkg.ProjectMgr,
 	}
 }
 
-type notificationJobAPI struct {
+type webhookJobAPI struct {
 	BaseAPI
-	webhookjobMgr    job.Manager
-	webhookPolicyMgr policy.Manager
-	projectMgr       project.Manager
+	webhookCtl webhook_ctl.Controller
+	projectMgr project.Manager
 }
 
-func (n *notificationJobAPI) ListWebhookJobs(ctx context.Context, params webhookjob.ListWebhookJobsParams) middleware.Responder {
+func (n *webhookJobAPI) ListWebhookJobs(ctx context.Context, params webhookjob.ListWebhookJobsParams) middleware.Responder {
 	projectNameOrID := parseProjectNameOrID(params.ProjectNameOrID, params.XIsResourceName)
 	if err := n.RequireProjectAccess(ctx, projectNameOrID, rbac.ActionList, rbac.ResourceNotificationPolicy); err != nil {
 		return n.SendError(ctx, err)
 	}
 
-	policy, err := n.webhookPolicyMgr.Get(ctx, params.PolicyID)
+	policy, err := n.webhookCtl.GetPolicy(ctx, params.PolicyID)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
@@ -51,24 +62,24 @@ func (n *notificationJobAPI) ListWebhookJobs(ctx context.Context, params webhook
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
-	query.Keywords["PolicyID"] = policy.ID
+
 	if len(params.Status) != 0 {
-		query.Keywords["Status"] = params.Status
+		query.Keywords["status"] = params.Status
 	}
 
-	total, err := n.webhookjobMgr.Count(ctx, query)
+	total, err := n.webhookCtl.CountExecutions(ctx, params.PolicyID, query)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
-
-	jobs, err := n.webhookjobMgr.List(ctx, query)
+	// the relationship of webhook execution and task is 1:1, so we can think the execution is the job as before.
+	jobs, err := n.webhookCtl.ListExecutions(ctx, params.PolicyID, query)
 	if err != nil {
 		return n.SendError(ctx, err)
 	}
 
 	var results []*models.WebhookJob
 	for _, j := range jobs {
-		results = append(results, model.NewNotificationJob(j).ToSwagger())
+		results = append(results, model.NewWebhookJob(j).ToSwagger())
 	}
 
 	return webhookjob.NewListWebhookJobsOK().
@@ -78,7 +89,7 @@ func (n *notificationJobAPI) ListWebhookJobs(ctx context.Context, params webhook
 }
 
 // requirePolicyAccess checks whether the project has the permission to the policy.
-func (n *notificationJobAPI) requirePolicyAccess(ctx context.Context, projectNameIrID interface{}, policy *policyModel.Policy) error {
+func (n *webhookJobAPI) requirePolicyAccess(ctx context.Context, projectNameIrID interface{}, policy *policyModel.Policy) error {
 	p, err := n.projectMgr.Get(ctx, projectNameIrID)
 	if err != nil {
 		return err

src/server/v2.0/handler/webhook_job_test.go

@@ -0,0 +1,102 @@
+//  Copyright Project Harbor Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+package handler
+
+import (
+	"fmt"
+	"testing"
+
+	policyModel "github.com/goharbor/harbor/src/pkg/notification/policy/model"
+	projectModel "github.com/goharbor/harbor/src/pkg/project/models"
+	"github.com/goharbor/harbor/src/pkg/task"
+	"github.com/goharbor/harbor/src/server/v2.0/models"
+	"github.com/goharbor/harbor/src/server/v2.0/restapi"
+	"github.com/goharbor/harbor/src/testing/controller/webhook"
+	"github.com/goharbor/harbor/src/testing/mock"
+	"github.com/goharbor/harbor/src/testing/pkg/project"
+	htesting "github.com/goharbor/harbor/src/testing/server/v2.0/handler"
+	"github.com/stretchr/testify/suite"
+)
+
+type WebhookJobTestSuite struct {
+	htesting.Suite
+	webhookCtl *webhook.Controller
+	projectMgr *project.Manager
+}
+
+func (suite *WebhookJobTestSuite) SetupSuite() {
+	suite.webhookCtl = &webhook.Controller{}
+	suite.projectMgr = &project.Manager{}
+	suite.Config = &restapi.Config{
+		WebhookjobAPI: &webhookJobAPI{
+			webhookCtl: suite.webhookCtl,
+			projectMgr: suite.projectMgr,
+		},
+	}
+	suite.Suite.SetupSuite()
+}
+
+func (suite *WebhookJobTestSuite) TestListWebhookJobs() {
+	projectID := int64(1)
+	policyID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.projectMgr.On("Get", mock.Anything, projectID).Return(&projectModel.Project{ProjectID: projectID}, nil)
+	suite.webhookCtl.On("GetPolicy", mock.Anything, policyID).Return(&policyModel.Policy{ID: policyID, Name: "test-policy"}, nil).Once()
+	suite.webhookCtl.On("CountExecutions", mock.Anything, policyID, mock.Anything).Return(int64(2), nil)
+	t1 := &task.Execution{ID: 1, VendorType: "WEBHOOK", VendorID: policyID, Status: "Success"}
+	t2 := &task.Execution{ID: 2, VendorType: "SLACK", VendorID: policyID, Status: "Stopped"}
+	suite.webhookCtl.On("ListExecutions", mock.Anything, policyID, mock.Anything).Return([]*task.Execution{t1, t2}, nil)
+
+	{
+		// query has no policy id should got 422
+		url := fmt.Sprintf("/projects/%d/webhook/jobs", projectID)
+		var body []*models.WebhookJob
+		resp, err := suite.GetJSON(url, &body)
+		suite.NoError(err)
+		suite.Equal(422, resp.StatusCode)
+	}
+
+	{
+		// unmatched project id should got 404
+		url := fmt.Sprintf("/projects/%d/webhook/jobs?policy_id=%d", projectID, policyID)
+		var body []*models.WebhookJob
+		resp, err := suite.GetJSON(url, &body)
+		suite.NoError(err)
+		suite.Equal(404, resp.StatusCode)
+	}
+
+	{
+		// normal requests should got 200
+		suite.webhookCtl.On("GetPolicy", mock.Anything, policyID, mock.Anything).Return(&policyModel.Policy{ID: policyID, Name: "test-policy", ProjectID: projectID}, nil)
+		url := fmt.Sprintf("/projects/%d/webhook/jobs?policy_id=%d", projectID, policyID)
+		var body []*models.WebhookJob
+		resp, err := suite.GetJSON(url, &body)
+		suite.NoError(err)
+		suite.Equal(200, resp.StatusCode)
+		suite.Len(body, 2)
+		// verify backward compatible
+		suite.Equal(body[0].ID, int64(1))
+		suite.Equal(body[0].NotifyType, "http")
+		suite.Equal(body[0].Status, "Success")
+		suite.Equal(body[1].ID, int64(2))
+		suite.Equal(body[1].NotifyType, "slack")
+		suite.Equal(body[1].Status, "Stopped")
+	}
+
+}
+
+func TestWebhookJobTestSuite(t *testing.T) {
+	suite.Run(t, &WebhookJobTestSuite{})
+}

src/server/v2.0/handler/webhook_test.go

@@ -0,0 +1,237 @@
+//  Copyright Project Harbor Authors
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+package handler
+
+import (
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/go-openapi/strfmt"
+	"github.com/goharbor/harbor/src/pkg/notification"
+	policyModel "github.com/goharbor/harbor/src/pkg/notification/policy/model"
+	taskModel "github.com/goharbor/harbor/src/pkg/task"
+	"github.com/goharbor/harbor/src/server/v2.0/models"
+	"github.com/goharbor/harbor/src/server/v2.0/restapi"
+	"github.com/goharbor/harbor/src/testing/controller/task"
+	"github.com/goharbor/harbor/src/testing/controller/webhook"
+	"github.com/goharbor/harbor/src/testing/mock"
+	htesting "github.com/goharbor/harbor/src/testing/server/v2.0/handler"
+	"github.com/stretchr/testify/suite"
+)
+
+type WebhookTestSuite struct {
+	htesting.Suite
+	webhookCtl *webhook.Controller
+	execCtl    *task.ExecutionController
+	taskCtl    *task.Controller
+}
+
+func (suite *WebhookTestSuite) SetupSuite() {
+	suite.webhookCtl = &webhook.Controller{}
+	suite.execCtl = &task.ExecutionController{}
+	suite.taskCtl = &task.Controller{}
+	suite.Config = &restapi.Config{
+		WebhookAPI: &webhookAPI{
+			webhookCtl: suite.webhookCtl,
+			execCtl:    suite.execCtl,
+			taskCtl:    suite.taskCtl,
+		},
+	}
+
+	suite.Suite.SetupSuite()
+	notification.Init()
+}
+
+func (suite *WebhookTestSuite) TestListWebhookPoliciesOfProject() {
+	projectID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("CountPolicies", mock.Anything, mock.Anything).Return(int64(1), nil)
+	suite.webhookCtl.On("ListPolicies", mock.Anything, mock.Anything).Return([]*policyModel.Policy{{ID: 1, ProjectID: projectID}}, nil)
+
+	url := fmt.Sprintf("/projects/%d/webhook/policies", projectID)
+	var body []*policyModel.Policy
+	resp, err := suite.GetJSON(url, &body)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+	suite.Len(body, 1)
+	suite.Equal(projectID, body[0].ProjectID)
+}
+
+func (suite *WebhookTestSuite) TestCreateWebhookPolicyOfProject() {
+	projectID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("CreatePolicy", mock.Anything, mock.Anything).Return(int64(1), nil)
+	url := fmt.Sprintf("/projects/%d/webhook/policies", projectID)
+	{
+		// invalid event type should got 400
+		resp, err := suite.PostJSON(url, &models.WebhookPolicy{EventTypes: []string{"INVALID"}})
+		suite.NoError(err)
+		suite.Equal(400, resp.StatusCode)
+	}
+
+	{
+		// invalid target type should got 400
+		resp, err := suite.PostJSON(url, &models.WebhookPolicy{EventTypes: []string{"PUSH_ARTIFACT"}, Targets: []*models.WebhookTargetObject{{Type: "invalid"}}})
+		suite.NoError(err)
+		suite.Equal(400, resp.StatusCode)
+	}
+
+	{
+		// valid policy should got 200
+		resp, err := suite.PostJSON(url, &models.WebhookPolicy{EventTypes: []string{"PUSH_ARTIFACT"}, Targets: []*models.WebhookTargetObject{{Type: "http", Address: "http://127.0.0.1"}}})
+		suite.NoError(err)
+		suite.Equal(201, resp.StatusCode)
+	}
+}
+
+func (suite *WebhookTestSuite) TestUpdateWebhookPolicyOfProject() {
+	projectID := int64(1)
+	policyID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("GetPolicy", mock.Anything, mock.Anything).Return(&policyModel.Policy{ID: policyID, ProjectID: projectID}, nil)
+	suite.webhookCtl.On("UpdatePolicy", mock.Anything, mock.Anything).Return(nil)
+	url := fmt.Sprintf("/projects/%d/webhook/policies/%d", projectID, policyID)
+	{
+		// invalid event type should got 400
+		resp, err := suite.PutJSON(url, &models.WebhookPolicy{EventTypes: []string{"INVALID"}})
+		suite.NoError(err)
+		suite.Equal(400, resp.StatusCode)
+	}
+
+	{
+		// invalid target type should got 400
+		resp, err := suite.PutJSON(url, &models.WebhookPolicy{EventTypes: []string{"PUSH_ARTIFACT"}, Targets: []*models.WebhookTargetObject{{Type: "invalid"}}})
+		suite.NoError(err)
+		suite.Equal(400, resp.StatusCode)
+	}
+
+	{
+		// valid policy should got 200
+		resp, err := suite.PutJSON(url, &models.WebhookPolicy{EventTypes: []string{"PUSH_ARTIFACT"}, Targets: []*models.WebhookTargetObject{{Type: "http", Address: "http://127.0.0.1"}}})
+		suite.NoError(err)
+		suite.Equal(200, resp.StatusCode)
+	}
+}
+
+func (suite *WebhookTestSuite) TestDeleteWebhookPolicyOfProject() {
+	projectID := int64(1)
+	policyID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("GetPolicy", mock.Anything, mock.Anything).Return(&policyModel.Policy{ID: policyID, ProjectID: projectID}, nil)
+	suite.webhookCtl.On("DeletePolicy", mock.Anything, mock.Anything).Return(nil)
+	url := fmt.Sprintf("/projects/%d/webhook/policies/%d", projectID, policyID)
+	resp, err := suite.Delete(url)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+}
+
+func (suite *WebhookTestSuite) TestGetWebhookPolicyOfProject() {
+	projectID := int64(1)
+	policyID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("GetPolicy", mock.Anything, mock.Anything).Return(&policyModel.Policy{ID: policyID, ProjectID: projectID}, nil)
+	url := fmt.Sprintf("/projects/%d/webhook/policies/%d", projectID, policyID)
+	var body *models.WebhookPolicy
+	resp, err := suite.GetJSON(url, &body)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+	suite.Equal(projectID, body.ProjectID)
+}
+
+func (suite *WebhookTestSuite) TestListExecutionsOfWebhookPolicy() {
+	projectID := int64(1)
+	policyID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("GetPolicy", mock.Anything, mock.Anything).Return(&policyModel.Policy{ID: policyID, ProjectID: projectID}, nil)
+	suite.webhookCtl.On("CountExecutions", mock.Anything, policyID, mock.Anything).Return(int64(1), nil)
+	suite.webhookCtl.On("ListExecutions", mock.Anything, policyID, mock.Anything).Return([]*taskModel.Execution{{ID: 1, VendorID: policyID}}, nil)
+	url := fmt.Sprintf("/projects/%d/webhook/policies/%d/executions", projectID, policyID)
+	var body []*taskModel.Execution
+	resp, err := suite.GetJSON(url, &body)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+	suite.Equal(policyID, body[0].VendorID)
+}
+
+func (suite *WebhookTestSuite) TestListTasksOfWebhookExecution() {
+	projectID := int64(1)
+	policyID := int64(1)
+	execID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("GetPolicy", mock.Anything, mock.Anything).Return(&policyModel.Policy{ID: policyID, ProjectID: projectID}, nil)
+	suite.execCtl.On("Get", mock.Anything, mock.Anything).Return(&taskModel.Execution{ID: execID, VendorID: projectID, VendorType: "WEBHOOK"}, nil)
+	suite.webhookCtl.On("CountTasks", mock.Anything, policyID, mock.Anything).Return(int64(1), nil)
+	suite.webhookCtl.On("ListTasks", mock.Anything, policyID, mock.Anything).Return([]*taskModel.Task{{ID: 1, ExecutionID: execID}}, nil)
+	url := fmt.Sprintf("/projects/%d/webhook/policies/%d/executions/%d/tasks", projectID, policyID, execID)
+	var body []*taskModel.Task
+	resp, err := suite.GetJSON(url, &body)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+	suite.Equal(execID, body[0].ExecutionID)
+}
+
+func (suite *WebhookTestSuite) TestGetLogsOfWebhookTask() {
+	projectID := int64(1)
+	policyID := int64(1)
+	execID := int64(1)
+	taskID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("GetPolicy", mock.Anything, mock.Anything).Return(&policyModel.Policy{ID: policyID, ProjectID: projectID}, nil)
+	suite.execCtl.On("Get", mock.Anything, mock.Anything).Return(&taskModel.Execution{ID: execID, VendorID: projectID, VendorType: "WEBHOOK"}, nil)
+	suite.taskCtl.On("Get", mock.Anything, mock.Anything).Return(&taskModel.Task{ID: taskID, ExecutionID: execID}, nil)
+	suite.webhookCtl.On("GetTaskLog", mock.Anything, taskID).Return([]byte("logs..."), nil)
+	url := fmt.Sprintf("/projects/%d/webhook/policies/%d/executions/%d/tasks/%d/log", projectID, policyID, execID, taskID)
+	resp, err := suite.Get(url)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+	data, err := io.ReadAll(resp.Body)
+	defer resp.Body.Close()
+	suite.NoError(err)
+	suite.Equal("logs...", string(data))
+}
+
+func (suite *WebhookTestSuite) TestLastTrigger() {
+	projectID := int64(1)
+	policyID := int64(1)
+	now := time.Now()
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	suite.webhookCtl.On("ListPolicies", mock.Anything, mock.Anything).Return([]*policyModel.Policy{{ID: policyID, ProjectID: projectID, EventTypes: []string{"PUSH_ARTIFACT"}}}, nil)
+	suite.webhookCtl.On("GetLastTriggerTime", mock.Anything, mock.Anything, policyID).Return(now, nil)
+	url := fmt.Sprintf("/projects/%d/webhook/lasttrigger", projectID)
+	var body []*models.WebhookLastTrigger
+	resp, err := suite.GetJSON(url, &body)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+	suite.Len(body, 1)
+	suite.Equal(strfmt.DateTime(now).String(), body[0].LastTriggerTime.String())
+}
+
+func (suite *WebhookTestSuite) TestGetSupportedEventTypes() {
+	projectID := int64(1)
+	suite.Security.On("Can", mock.Anything, mock.Anything, mock.Anything).Return(true)
+	url := fmt.Sprintf("/projects/%d/webhook/events", projectID)
+	var body *models.SupportedWebhookEventTypes
+	resp, err := suite.GetJSON(url, &body)
+	suite.NoError(err)
+	suite.Equal(200, resp.StatusCode)
+	suite.Len(body.EventType, len(notification.GetSupportedEventTypes()))
+	suite.Len(body.NotifyType, len(notification.GetSupportedNotifyTypes()))
+}
+
+func TestWebhookTestSuite(t *testing.T) {
+	suite.Run(t, &WebhookTestSuite{})
+}

src/testing/controller/controller.go

@@ -32,3 +32,6 @@ package controller
 //go:generate mockery --case snake --dir ../../controller/jobservice --name SchedulerController --output ./jobservice --outpkg jobservice
 //go:generate mockery --case snake --dir ../../controller/systemartifact --name Controller --output ./systemartifact --outpkg systemartifact
 //go:generate mockery --case snake --dir ../../controller/scandataexport --name Controller --output ./scandataexport --outpkg scandataexport
+//go:generate mockery --case snake --dir ../../controller/task --name Controller --output ./task --outpkg task
+//go:generate mockery --case snake --dir ../../controller/task --name ExecutionController --output ./task --outpkg task
+//go:generate mockery --case snake --dir ../../controller/webhook --name Controller --output ./webhook --outpkg webhook

src/testing/controller/task/controller.go

@@ -0,0 +1,136 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package task
+
+import (
+	context "context"
+
+	pkgtask "github.com/goharbor/harbor/src/pkg/task"
+	mock "github.com/stretchr/testify/mock"
+
+	q "github.com/goharbor/harbor/src/lib/q"
+)
+
+// Controller is an autogenerated mock type for the Controller type
+type Controller struct {
+	mock.Mock
+}
+
+// Count provides a mock function with given fields: ctx, query
+func (_m *Controller) Count(ctx context.Context, query *q.Query) (int64, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) int64); ok {
+		r0 = rf(ctx, query)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Get provides a mock function with given fields: ctx, id
+func (_m *Controller) Get(ctx context.Context, id int64) (*pkgtask.Task, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *pkgtask.Task
+	if rf, ok := ret.Get(0).(func(context.Context, int64) *pkgtask.Task); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*pkgtask.Task)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetLog provides a mock function with given fields: ctx, id
+func (_m *Controller) GetLog(ctx context.Context, id int64) ([]byte, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 []byte
+	if rf, ok := ret.Get(0).(func(context.Context, int64) []byte); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]byte)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// List provides a mock function with given fields: ctx, query
+func (_m *Controller) List(ctx context.Context, query *q.Query) ([]*pkgtask.Task, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 []*pkgtask.Task
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) []*pkgtask.Task); ok {
+		r0 = rf(ctx, query)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*pkgtask.Task)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Stop provides a mock function with given fields: ctx, id
+func (_m *Controller) Stop(ctx context.Context, id int64) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewController interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewController creates a new instance of Controller. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewController(t mockConstructorTestingTNewController) *Controller {
+	mock := &Controller{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

src/testing/controller/task/execution_controller.go

@@ -0,0 +1,127 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package task
+
+import (
+	context "context"
+
+	pkgtask "github.com/goharbor/harbor/src/pkg/task"
+	mock "github.com/stretchr/testify/mock"
+
+	q "github.com/goharbor/harbor/src/lib/q"
+)
+
+// ExecutionController is an autogenerated mock type for the ExecutionController type
+type ExecutionController struct {
+	mock.Mock
+}
+
+// Count provides a mock function with given fields: ctx, query
+func (_m *ExecutionController) Count(ctx context.Context, query *q.Query) (int64, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) int64); ok {
+		r0 = rf(ctx, query)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Delete provides a mock function with given fields: ctx, id
+func (_m *ExecutionController) Delete(ctx context.Context, id int64) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Get provides a mock function with given fields: ctx, id
+func (_m *ExecutionController) Get(ctx context.Context, id int64) (*pkgtask.Execution, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *pkgtask.Execution
+	if rf, ok := ret.Get(0).(func(context.Context, int64) *pkgtask.Execution); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*pkgtask.Execution)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// List provides a mock function with given fields: ctx, query
+func (_m *ExecutionController) List(ctx context.Context, query *q.Query) ([]*pkgtask.Execution, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 []*pkgtask.Execution
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) []*pkgtask.Execution); ok {
+		r0 = rf(ctx, query)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*pkgtask.Execution)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Stop provides a mock function with given fields: ctx, id
+func (_m *ExecutionController) Stop(ctx context.Context, id int64) error {
+	ret := _m.Called(ctx, id)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewExecutionController interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewExecutionController creates a new instance of ExecutionController. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewExecutionController(t mockConstructorTestingTNewExecutionController) *ExecutionController {
+	mock := &ExecutionController{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

src/testing/controller/webhook/controller.go

@@ -0,0 +1,330 @@
+// Code generated by mockery v2.14.0. DO NOT EDIT.
+
+package webhook
+
+import (
+	context "context"
+
+	model "github.com/goharbor/harbor/src/pkg/notification/policy/model"
+	mock "github.com/stretchr/testify/mock"
+
+	q "github.com/goharbor/harbor/src/lib/q"
+
+	task "github.com/goharbor/harbor/src/pkg/task"
+
+	time "time"
+)
+
+// Controller is an autogenerated mock type for the Controller type
+type Controller struct {
+	mock.Mock
+}
+
+// CountExecutions provides a mock function with given fields: ctx, policyID, query
+func (_m *Controller) CountExecutions(ctx context.Context, policyID int64, query *q.Query) (int64, error) {
+	ret := _m.Called(ctx, policyID, query)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, int64, *q.Query) int64); ok {
+		r0 = rf(ctx, policyID, query)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64, *q.Query) error); ok {
+		r1 = rf(ctx, policyID, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// CountPolicies provides a mock function with given fields: ctx, query
+func (_m *Controller) CountPolicies(ctx context.Context, query *q.Query) (int64, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) int64); ok {
+		r0 = rf(ctx, query)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// CountTasks provides a mock function with given fields: ctx, execID, query
+func (_m *Controller) CountTasks(ctx context.Context, execID int64, query *q.Query) (int64, error) {
+	ret := _m.Called(ctx, execID, query)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, int64, *q.Query) int64); ok {
+		r0 = rf(ctx, execID, query)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64, *q.Query) error); ok {
+		r1 = rf(ctx, execID, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// CreatePolicy provides a mock function with given fields: ctx, policy
+func (_m *Controller) CreatePolicy(ctx context.Context, policy *model.Policy) (int64, error) {
+	ret := _m.Called(ctx, policy)
+
+	var r0 int64
+	if rf, ok := ret.Get(0).(func(context.Context, *model.Policy) int64); ok {
+		r0 = rf(ctx, policy)
+	} else {
+		r0 = ret.Get(0).(int64)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *model.Policy) error); ok {
+		r1 = rf(ctx, policy)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// DeletePolicy provides a mock function with given fields: ctx, policyID
+func (_m *Controller) DeletePolicy(ctx context.Context, policyID int64) error {
+	ret := _m.Called(ctx, policyID)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
+		r0 = rf(ctx, policyID)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// GetLastTriggerTime provides a mock function with given fields: ctx, eventType, policyID
+func (_m *Controller) GetLastTriggerTime(ctx context.Context, eventType string, policyID int64) (time.Time, error) {
+	ret := _m.Called(ctx, eventType, policyID)
+
+	var r0 time.Time
+	if rf, ok := ret.Get(0).(func(context.Context, string, int64) time.Time); ok {
+		r0 = rf(ctx, eventType, policyID)
+	} else {
+		r0 = ret.Get(0).(time.Time)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, int64) error); ok {
+		r1 = rf(ctx, eventType, policyID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetPolicy provides a mock function with given fields: ctx, id
+func (_m *Controller) GetPolicy(ctx context.Context, id int64) (*model.Policy, error) {
+	ret := _m.Called(ctx, id)
+
+	var r0 *model.Policy
+	if rf, ok := ret.Get(0).(func(context.Context, int64) *model.Policy); ok {
+		r0 = rf(ctx, id)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*model.Policy)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, id)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetRelatedPolices provides a mock function with given fields: ctx, projectID, eventType
+func (_m *Controller) GetRelatedPolices(ctx context.Context, projectID int64, eventType string) ([]*model.Policy, error) {
+	ret := _m.Called(ctx, projectID, eventType)
+
+	var r0 []*model.Policy
+	if rf, ok := ret.Get(0).(func(context.Context, int64, string) []*model.Policy); ok {
+		r0 = rf(ctx, projectID, eventType)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*model.Policy)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64, string) error); ok {
+		r1 = rf(ctx, projectID, eventType)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetTask provides a mock function with given fields: ctx, taskID
+func (_m *Controller) GetTask(ctx context.Context, taskID int64) (*task.Task, error) {
+	ret := _m.Called(ctx, taskID)
+
+	var r0 *task.Task
+	if rf, ok := ret.Get(0).(func(context.Context, int64) *task.Task); ok {
+		r0 = rf(ctx, taskID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*task.Task)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, taskID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// GetTaskLog provides a mock function with given fields: ctx, taskID
+func (_m *Controller) GetTaskLog(ctx context.Context, taskID int64) ([]byte, error) {
+	ret := _m.Called(ctx, taskID)
+
+	var r0 []byte
+	if rf, ok := ret.Get(0).(func(context.Context, int64) []byte); ok {
+		r0 = rf(ctx, taskID)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]byte)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, taskID)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ListExecutions provides a mock function with given fields: ctx, policyID, query
+func (_m *Controller) ListExecutions(ctx context.Context, policyID int64, query *q.Query) ([]*task.Execution, error) {
+	ret := _m.Called(ctx, policyID, query)
+
+	var r0 []*task.Execution
+	if rf, ok := ret.Get(0).(func(context.Context, int64, *q.Query) []*task.Execution); ok {
+		r0 = rf(ctx, policyID, query)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*task.Execution)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64, *q.Query) error); ok {
+		r1 = rf(ctx, policyID, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ListPolicies provides a mock function with given fields: ctx, query
+func (_m *Controller) ListPolicies(ctx context.Context, query *q.Query) ([]*model.Policy, error) {
+	ret := _m.Called(ctx, query)
+
+	var r0 []*model.Policy
+	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) []*model.Policy); ok {
+		r0 = rf(ctx, query)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*model.Policy)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
+		r1 = rf(ctx, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// ListTasks provides a mock function with given fields: ctx, execID, query
+func (_m *Controller) ListTasks(ctx context.Context, execID int64, query *q.Query) ([]*task.Task, error) {
+	ret := _m.Called(ctx, execID, query)
+
+	var r0 []*task.Task
+	if rf, ok := ret.Get(0).(func(context.Context, int64, *q.Query) []*task.Task); ok {
+		r0 = rf(ctx, execID, query)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*task.Task)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, int64, *q.Query) error); ok {
+		r1 = rf(ctx, execID, query)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// UpdatePolicy provides a mock function with given fields: ctx, policy
+func (_m *Controller) UpdatePolicy(ctx context.Context, policy *model.Policy) error {
+	ret := _m.Called(ctx, policy)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context, *model.Policy) error); ok {
+		r0 = rf(ctx, policy)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewController interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewController creates a new instance of Controller. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewController(t mockConstructorTestingTNewController) *Controller {
+	mock := &Controller{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

src/testing/pkg/notification/job/dao/dao.go

@@ -1,193 +0,0 @@
-// Code generated by mockery v2.14.0. DO NOT EDIT.
-
-package dao
-
-import (
-	context "context"
-
-	mock "github.com/stretchr/testify/mock"
-
-	model "github.com/goharbor/harbor/src/pkg/notification/job/model"
-
-	q "github.com/goharbor/harbor/src/lib/q"
-)
-
-// DAO is an autogenerated mock type for the DAO type
-type DAO struct {
-	mock.Mock
-}
-
-// Count provides a mock function with given fields: ctx, query
-func (_m *DAO) Count(ctx context.Context, query *q.Query) (int64, error) {
-	ret := _m.Called(ctx, query)
-
-	var r0 int64
-	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) int64); ok {
-		r0 = rf(ctx, query)
-	} else {
-		r0 = ret.Get(0).(int64)
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
-		r1 = rf(ctx, query)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// Create provides a mock function with given fields: ctx, n
-func (_m *DAO) Create(ctx context.Context, n *model.Job) (int64, error) {
-	ret := _m.Called(ctx, n)
-
-	var r0 int64
-	if rf, ok := ret.Get(0).(func(context.Context, *model.Job) int64); ok {
-		r0 = rf(ctx, n)
-	} else {
-		r0 = ret.Get(0).(int64)
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *model.Job) error); ok {
-		r1 = rf(ctx, n)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// Delete provides a mock function with given fields: ctx, id
-func (_m *DAO) Delete(ctx context.Context, id int64) error {
-	ret := _m.Called(ctx, id)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
-		r0 = rf(ctx, id)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// DeleteByPolicyID provides a mock function with given fields: ctx, policyID
-func (_m *DAO) DeleteByPolicyID(ctx context.Context, policyID int64) error {
-	ret := _m.Called(ctx, policyID)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
-		r0 = rf(ctx, policyID)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// Get provides a mock function with given fields: ctx, id
-func (_m *DAO) Get(ctx context.Context, id int64) (*model.Job, error) {
-	ret := _m.Called(ctx, id)
-
-	var r0 *model.Job
-	if rf, ok := ret.Get(0).(func(context.Context, int64) *model.Job); ok {
-		r0 = rf(ctx, id)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*model.Job)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
-		r1 = rf(ctx, id)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// GetLastTriggerJobsGroupByEventType provides a mock function with given fields: ctx, policyID
-func (_m *DAO) GetLastTriggerJobsGroupByEventType(ctx context.Context, policyID int64) ([]*model.Job, error) {
-	ret := _m.Called(ctx, policyID)
-
-	var r0 []*model.Job
-	if rf, ok := ret.Get(0).(func(context.Context, int64) []*model.Job); ok {
-		r0 = rf(ctx, policyID)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]*model.Job)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
-		r1 = rf(ctx, policyID)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// List provides a mock function with given fields: ctx, query
-func (_m *DAO) List(ctx context.Context, query *q.Query) ([]*model.Job, error) {
-	ret := _m.Called(ctx, query)
-
-	var r0 []*model.Job
-	if rf, ok := ret.Get(0).(func(context.Context, *q.Query) []*model.Job); ok {
-		r0 = rf(ctx, query)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]*model.Job)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *q.Query) error); ok {
-		r1 = rf(ctx, query)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// Update provides a mock function with given fields: ctx, n, props
-func (_m *DAO) Update(ctx context.Context, n *model.Job, props ...string) error {
-	_va := make([]interface{}, len(props))
-	for _i := range props {
-		_va[_i] = props[_i]
-	}
-	var _ca []interface{}
-	_ca = append(_ca, ctx, n)
-	_ca = append(_ca, _va...)
-	ret := _m.Called(_ca...)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *model.Job, ...string) error); ok {
-		r0 = rf(ctx, n, props...)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-type mockConstructorTestingTNewDAO interface {
-	mock.TestingT
-	Cleanup(func())
-}
-
-// NewDAO creates a new instance of DAO. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
-func NewDAO(t mockConstructorTestingTNewDAO) *DAO {
-	mock := &DAO{}
-	mock.Mock.Test(t)
-
-	t.Cleanup(func() { mock.AssertExpectations(t) })
-
-	return mock
-}

src/testing/pkg/notification/policy/manager.go

@@ -1,6 +1,6 @@
 // Code generated by mockery v2.14.0. DO NOT EDIT.
 
-package notification
+package policy
 
 import (
 	context "context"
@@ -95,29 +95,6 @@ func (_m *Manager) Get(ctx context.Context, id int64) (*model.Policy, error) {
 	return r0, r1
 }
 
-// GetByNameAndProjectID provides a mock function with given fields: ctx, name, projectID
-func (_m *Manager) GetByNameAndProjectID(ctx context.Context, name string, projectID int64) (*model.Policy, error) {
-	ret := _m.Called(ctx, name, projectID)
-
-	var r0 *model.Policy
-	if rf, ok := ret.Get(0).(func(context.Context, string, int64) *model.Policy); ok {
-		r0 = rf(ctx, name, projectID)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*model.Policy)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, string, int64) error); ok {
-		r1 = rf(ctx, name, projectID)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
 // GetRelatedPolices provides a mock function with given fields: ctx, projectID, eventType
 func (_m *Manager) GetRelatedPolices(ctx context.Context, projectID int64, eventType string) ([]*model.Policy, error) {
 	ret := _m.Called(ctx, projectID, eventType)
@@ -164,20 +141,6 @@ func (_m *Manager) List(ctx context.Context, query *q.Query) ([]*model.Policy, e
 	return r0, r1
 }
 
-// Test provides a mock function with given fields: _a0
-func (_m *Manager) Test(_a0 *model.Policy) error {
-	ret := _m.Called(_a0)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(*model.Policy) error); ok {
-		r0 = rf(_a0)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // Update provides a mock function with given fields: ctx, _a1
 func (_m *Manager) Update(ctx context.Context, _a1 *model.Policy) error {
 	ret := _m.Called(ctx, _a1)

src/testing/pkg/pkg.go

@@ -37,9 +37,8 @@ package pkg
 //go:generate mockery --case snake --dir ../../pkg/robot/dao --name DAO --output ./robot/dao --outpkg dao
 //go:generate mockery --case snake --dir ../../pkg/repository --name Manager --output ./repository --outpkg repository
 //go:generate mockery --case snake --dir ../../pkg/repository/dao --name DAO --output ./repository/dao --outpkg dao
-//go:generate mockery --case snake --dir ../../pkg/notification/job/dao --name DAO --output ./notification/job/dao --outpkg dao
 //go:generate mockery --case snake --dir ../../pkg/notification/policy/dao --name DAO --output ./notification/policy/dao --outpkg dao
-//go:generate mockery --case snake --dir ../../pkg/notification/policy --name Manager --output ./notification/policy --outpkg notification
+//go:generate mockery --case snake --dir ../../pkg/notification/policy --name Manager --output ./notification/policy --outpkg policy
 //go:generate mockery --case snake --dir ../../pkg/immutable/dao --name DAO --output ./immutable/dao --outpkg dao
 //go:generate mockery --case snake --dir ../../pkg/ldap --name Manager --output ./ldap --outpkg ldap
 //go:generate mockery --case snake --dir ../../pkg/allowlist --name Manager --output ./allowlist --outpkg robot