Merge pull request #472 from alessandro-sorint/commitstatus-retry

gateway: add api to redeliver a commit status delivery

internal/services/gateway/action/commitstatusdelivery.go

@@ -85,3 +85,29 @@ func (h *ActionHandler) GetProjectCommitStatusDeliveries(ctx context.Context, re
 
 	return res, nil
 }
+
+type ProjectCommitStatusRedeliveryRequest struct {
+	ProjectRef             string
+	CommitStatusDeliveryID string
+}
+
+func (h *ActionHandler) ProjectCommitStatusRedelivery(ctx context.Context, req *ProjectCommitStatusRedeliveryRequest) error {
+	project, _, err := h.configstoreClient.GetProject(ctx, req.ProjectRef)
+	if err != nil {
+		return util.NewAPIError(util.KindFromRemoteError(err), err)
+	}
+	isUserOwner, err := h.IsAuthUserProjectOwner(ctx, project.OwnerType, project.OwnerID)
+	if err != nil {
+		return errors.Wrapf(err, "failed to determine permissions")
+	}
+	if !isUserOwner {
+		return util.NewAPIError(util.ErrForbidden, errors.Errorf("user not authorized"))
+	}
+
+	_, err = h.notificationClient.CommitStatusRedelivery(ctx, project.ID, req.CommitStatusDeliveryID)
+	if err != nil {
+		return util.NewAPIError(util.KindFromRemoteError(err), err)
+	}
+
+	return nil
+}

internal/services/gateway/api/commitstatusdelivery.go

@@ -94,3 +94,43 @@ func (h *ProjectCommitStatusDeliveries) do(w http.ResponseWriter, r *http.Reques
 
 	return commitStatusDeliveries, nil
 }
+
+type ProjectCommitStatusRedelivery struct {
+	log zerolog.Logger
+	ah  *action.ActionHandler
+}
+
+func NewProjectCommitStatusRedeliveryHandler(log zerolog.Logger, ah *action.ActionHandler) *ProjectCommitStatusRedelivery {
+	return &ProjectCommitStatusRedelivery{log: log, ah: ah}
+}
+
+func (h *ProjectCommitStatusRedelivery) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	err := h.do(w, r)
+	if util.HTTPError(w, err) {
+		h.log.Err(err).Send()
+		return
+	}
+
+	if err := util.HTTPResponse(w, http.StatusOK, nil); err != nil {
+		h.log.Err(err).Send()
+	}
+}
+
+func (h *ProjectCommitStatusRedelivery) do(w http.ResponseWriter, r *http.Request) error {
+	ctx := r.Context()
+
+	vars := mux.Vars(r)
+	projectRef := vars["projectref"]
+	commitStatusDeliveryID := vars["commitstatusdeliveryid"]
+
+	areq := &action.ProjectCommitStatusRedeliveryRequest{
+		ProjectRef:             projectRef,
+		CommitStatusDeliveryID: commitStatusDeliveryID,
+	}
+	err := h.ah.ProjectCommitStatusRedelivery(ctx, areq)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}

internal/services/gateway/gateway.go

@@ -202,6 +202,7 @@ func (g *Gateway) Run(ctx context.Context) error {
 	projectRunWebhookDeliveriesHandler := api.NewProjectRunWebhookDeliveriesHandler(g.log, g.ah)
 	projectRunWebhookRedeliveryHandler := api.NewProjectRunWebhookRedeliveryHandler(g.log, g.ah)
 	projectCommitStatusDeliveriesHandler := api.NewProjectCommitStatusDeliveriesHandler(g.log, g.ah)
+	projectCommitStatusRedeliveryHandler := api.NewProjectCommitStatusRedeliveryHandler(g.log, g.ah)
 
 	secretHandler := api.NewSecretHandler(g.log, g.ah)
 	createSecretHandler := api.NewCreateSecretHandler(g.log, g.ah)
@@ -324,6 +325,7 @@ func (g *Gateway) Run(ctx context.Context) error {
 	apirouter.Handle("/projects/{projectref}/runwebhookdeliveries", authForcedHandler(projectRunWebhookDeliveriesHandler)).Methods("GET")
 	apirouter.Handle("/projects/{projectref}/runwebhookdeliveries/{runwebhookdeliveryid}/redelivery", authForcedHandler(projectRunWebhookRedeliveryHandler)).Methods("PUT")
 	apirouter.Handle("/projects/{projectref}/commitstatusdeliveries", authForcedHandler(projectCommitStatusDeliveriesHandler)).Methods("GET")
+	apirouter.Handle("/projects/{projectref}/commitstatusdeliveries/{commitstatusdeliveryid}/redelivery", authForcedHandler(projectCommitStatusRedeliveryHandler)).Methods("PUT")
 
 	apirouter.Handle("/projectgroups/{projectgroupref}/secrets", authForcedHandler(secretHandler)).Methods("GET")
 	apirouter.Handle("/projects/{projectref}/secrets", authForcedHandler(secretHandler)).Methods("GET")

internal/services/notification/action/commitstatusdelivery.go

@@ -20,6 +20,7 @@ import (
 	"github.com/sorintlab/errors"
 
 	"agola.io/agola/internal/sqlg/sql"
+	"agola.io/agola/internal/util"
 	"agola.io/agola/services/notification/types"
 )
 
@@ -72,3 +73,51 @@ func (h *ActionHandler) GetProjectCommitStatusDeliveries(ctx context.Context, re
 		HasMore:                hasMore,
 	}, nil
 }
+
+func (h *ActionHandler) CommitStatusRedelivery(ctx context.Context, projectID string, commitStatusDeliveryID string) error {
+	err := h.d.Do(ctx, func(tx *sql.Tx) error {
+		var err error
+		commitStatusDelivery, err := h.d.GetCommitStatusDeliveryByID(tx, commitStatusDeliveryID)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		if commitStatusDelivery == nil {
+			return util.NewAPIError(util.ErrNotExist, errors.Errorf("commitStatusDelivery %q doesn't exist", commitStatusDeliveryID))
+		}
+
+		commitStatus, err := h.d.GetCommitStatusByID(tx, commitStatusDelivery.CommitStatusID)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		if commitStatus == nil {
+			return util.NewAPIError(util.ErrNotExist, errors.Errorf("commitStatus %q doesn't exist", commitStatusDelivery.CommitStatusID))
+		}
+		if commitStatus.ProjectID != projectID {
+			return util.NewAPIError(util.ErrNotExist, errors.Errorf("commitStatusDelivery %q doesn't belong to project %q", commitStatusDeliveryID, projectID))
+		}
+
+		commitStatusDeliveries, err := h.d.GetCommitStatusDeliveriesByCommitStatusID(tx, commitStatusDelivery.CommitStatusID, []types.DeliveryStatus{types.DeliveryStatusNotDelivered}, 1, types.SortDirectionDesc)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		// check if commitStatus has delivery not delivered
+		if len(commitStatusDeliveries) != 0 {
+			return util.NewAPIError(util.ErrBadRequest, errors.Errorf("the previous delivery of commit status %q hasn't already been delivered", commitStatusDelivery.CommitStatusID))
+		}
+
+		newCommitStatusDelivery := types.NewCommitStatusDelivery(tx)
+		newCommitStatusDelivery.DeliveryStatus = types.DeliveryStatusNotDelivered
+		newCommitStatusDelivery.CommitStatusID = commitStatusDelivery.CommitStatusID
+		err = h.d.InsertCommitStatusDelivery(tx, newCommitStatusDelivery)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		return nil
+	})
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}

internal/services/notification/api/commitstatusdelivery.go

@@ -89,3 +89,39 @@ func (h *CommitStatusDeliveriesHandler) do(w http.ResponseWriter, r *http.Reques
 
 	return ares.CommitStatusDeliveries, nil
 }
+
+type CommitStatusRedeliveryHandler struct {
+	log zerolog.Logger
+	ah  *action.ActionHandler
+}
+
+func NewCommitStatusRedeliveryHandler(log zerolog.Logger, ah *action.ActionHandler) *CommitStatusRedeliveryHandler {
+	return &CommitStatusRedeliveryHandler{log: log, ah: ah}
+}
+
+func (h *CommitStatusRedeliveryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	err := h.do(w, r)
+	if util.HTTPError(w, err) {
+		h.log.Err(err).Send()
+		return
+	}
+
+	if err := util.HTTPResponse(w, http.StatusOK, nil); err != nil {
+		h.log.Err(err).Send()
+	}
+}
+
+func (h *CommitStatusRedeliveryHandler) do(w http.ResponseWriter, r *http.Request) error {
+	ctx := r.Context()
+
+	vars := mux.Vars(r)
+	projectID := vars["projectid"]
+	commitStatusDeliveryID := vars["commitstatusdeliveryid"]
+
+	err := h.ah.CommitStatusRedelivery(ctx, projectID, commitStatusDeliveryID)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	return nil
+}

internal/services/notification/db/db.go

@@ -328,6 +328,29 @@ func (d *DB) GetProjectCommitStatusDeliveriesAfterSequenceByProjectID(tx *sql.Tx
 	return commitStatusDeliveries, errors.WithStack(err)
 }
 
+func (d *DB) GetCommitStatusDeliveriesByCommitStatusID(tx *sql.Tx, commitStatusID string, deliveryStatusFilter []types.DeliveryStatus, limit int, sortDirection types.SortDirection) ([]*types.CommitStatusDelivery, error) {
+	q := commitStatusDeliverySelect().OrderBy("sequence")
+	q.Where(q.E("commit_status_id", commitStatusID))
+
+	if len(deliveryStatusFilter) > 0 {
+		q.Where(q.In("delivery_status", sq.Flatten(deliveryStatusFilter)...))
+	}
+
+	switch sortDirection {
+	case types.SortDirectionAsc:
+		q.Asc()
+	case types.SortDirectionDesc:
+		q.Desc()
+	}
+
+	if limit > 0 {
+		q.Limit(limit)
+	}
+
+	commitStatusDeliveries, _, err := d.fetchCommitStatusDeliverys(tx, q)
+	return commitStatusDeliveries, errors.WithStack(err)
+}
+
 func (d *DB) DeleteCommitStatusDeliveriesByCommitStatusID(tx *sql.Tx, commitStatusID string) error {
 	q := sq.NewDeleteBuilder()
 	q.DeleteFrom("commitstatusdelivery")

internal/services/notification/notification.go

@@ -123,6 +123,7 @@ func (n *NotificationService) setupDefaultRouter() http.Handler {
 	runWebhookDeliveriesHandler := api.NewRunWebhookDeliveriesHandler(n.log, n.ah)
 	runWebhookReliveryHandler := api.NewRunWebhookRedeliveryHandler(n.log, n.ah)
 	commitStatusDeliveriesHandler := api.NewCommitStatusDeliveriesHandler(n.log, n.ah)
+	commitStatusReliveryHandler := api.NewCommitStatusRedeliveryHandler(n.log, n.ah)
 
 	authHandler := handlers.NewInternalAuthChecker(n.log, n.c.APIToken)
 
@@ -140,6 +141,8 @@ func (n *NotificationService) setupDefaultRouter() http.Handler {
 
 	apirouter.Handle("/projects/{projectid}/commitstatusdeliveries", commitStatusDeliveriesHandler).Methods("GET")
 
+	apirouter.Handle("/projects/{projectid}/commitstatusdeliveries/{commitstatusdeliveryid}/redelivery", commitStatusReliveryHandler).Methods("PUT")
+
 	mainrouter := mux.NewRouter().UseEncodedPath().SkipClean(true)
 	mainrouter.PathPrefix("/").Handler(router)