Resubmit a stored event as a new undelivered event (closes #250)

internal/handlers/webhook.go

@@ -3,6 +3,7 @@ package handlers
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 
@@ -255,8 +256,8 @@ func (h *Handlers) readWebhookBody(
 	return body, true
 }
 
-// createAndDeliverEvent creates the event and delivery records
-// then notifies the delivery engine.
+// createAndDeliverEvent stores the received event, fans it out to the
+// webhook's targets, and answers the sender.
 func (h *Handlers) createAndDeliverEvent(
 	w http.ResponseWriter,
 	r *http.Request,
@@ -264,69 +265,130 @@ func (h *Handlers) createAndDeliverEvent(
 	body, headersJSON []byte,
 	targets []database.Target,
 ) {
-	tx, err := h.beginWebhookTx(w, entrypoint.WebhookID)
-	if err != nil {
-		return
-	}
-
-	event := h.buildEvent(r, entrypoint, headersJSON, body)
-
-	err = tx.Create(event).Error
-	if err != nil {
-		tx.Rollback()
-		h.serverError(w, "failed to create event", err)
-
-		return
-	}
-
-	bodyPtr := inlineBody(body)
-
-	tasks := h.buildDeliveryTasks(
-		w, tx, event, entrypoint, targets, bodyPtr,
+	event, tasks, err := h.createAndFanOut(
+		requestEventSource(r, entrypoint, headersJSON, body),
+		targets,
 	)
-	if tasks == nil {
-		return
-	}
-
-	err = tx.Commit().Error
 	if err != nil {
-		h.serverError(w, "failed to commit transaction", err)
+		h.serverError(w, "failed to store webhook event", err)
 
 		return
 	}
 
-	// Counted here, after the commit: an event is received once it
-	// is durably stored, which is what the delivery counters are
-	// compared against on a dashboard.
-	h.mtr.EventReceived()
-
 	h.finishWebhookResponse(w, event, entrypoint, tasks)
 }
 
-// beginWebhookTx opens a transaction on the per-webhook DB.
-func (h *Handlers) beginWebhookTx(
-	w http.ResponseWriter,
-	webhookID string,
-) (*gorm.DB, error) {
-	webhookDB, err := h.dbMgr.GetDB(webhookID)
-	if err != nil {
-		h.serverError(
-			w, "failed to get webhook database", err,
-		)
+// eventSource carries the fields a new event is built from. The
+// receiver fills it from the live request; the resubmit handler fills
+// it from a stored event. Both then go through createAndFanOut, so an
+// event is constructed and fanned out in one place however it entered
+// the system.
+type eventSource struct {
+	WebhookID    string
+	EntrypointID string
+	Method       string
+	HeadersJSON  string
+	ContentType  string
+	Body         []byte
 
-		return nil, err
+	// ResubmittedFromID names the event this one copies. Only the
+	// resubmit path sets it.
+	ResubmittedFromID *string
+}
+
+// event builds the row this source stores.
+func (s eventSource) event() *database.Event {
+	return &database.Event{
+		WebhookID:         s.WebhookID,
+		EntrypointID:      s.EntrypointID,
+		Method:            s.Method,
+		Headers:           s.HeadersJSON,
+		Body:              string(s.Body),
+		ContentType:       s.ContentType,
+		ResubmittedFromID: s.ResubmittedFromID,
+	}
+}
+
+// requestEventSource describes the event a live receiver request
+// stores.
+func requestEventSource(
+	r *http.Request,
+	entrypoint database.Entrypoint,
+	headersJSON, body []byte,
+) eventSource {
+	return eventSource{
+		WebhookID:    entrypoint.WebhookID,
+		EntrypointID: entrypoint.ID,
+		Method:       r.Method,
+		HeadersJSON:  string(headersJSON),
+		ContentType:  r.Header.Get("Content-Type"),
+		Body:         body,
+	}
+}
+
+// createAndFanOut writes the event and one pending delivery per target
+// in a single transaction, then hands the tasks to the delivery
+// engine. It is the only path by which an event and its deliveries are
+// created, so a resubmitted event is retried, SSRF-guarded and
+// circuit-broken exactly as a received one is.
+//
+// The tasks are returned as well as queued, so a caller can report how
+// many targets the event went to.
+func (h *Handlers) createAndFanOut(
+	src eventSource,
+	targets []database.Target,
+) (*database.Event, []delivery.Task, error) {
+	webhookDB, err := h.dbMgr.GetDB(src.WebhookID)
+	if err != nil {
+		return nil, nil, fmt.Errorf(
+			"getting webhook database: %w", err,
+		)
 	}
 
 	tx := webhookDB.Begin()
 	if tx.Error != nil {
-		h.serverError(
-			w, "failed to begin transaction", tx.Error,
+		return nil, nil, fmt.Errorf(
+			"beginning transaction: %w", tx.Error,
 		)
-
-		return nil, tx.Error
 	}
 
-	return tx, nil
+	event := src.event()
+
+	err = tx.Create(event).Error
+	if err != nil {
+		tx.Rollback()
+
+		return nil, nil, fmt.Errorf("creating event: %w", err)
+	}
+
+	tasks, err := buildDeliveryTasks(
+		tx, event, targets, inlineBody(src.Body),
+	)
+	if err != nil {
+		tx.Rollback()
+
+		return nil, nil, err
+	}
+
+	err = tx.Commit().Error
+	if err != nil {
+		return nil, nil, fmt.Errorf(
+			"committing transaction: %w", err,
+		)
+	}
+
+	// Counted here, after the commit: an event exists once it is
+	// durably stored, which is what the delivery counters are
+	// compared against on a dashboard. A resubmitted event counts
+	// too, because it produces deliveries that the delivery side
+	// counts; the resubmit counter is what separates the two.
+	h.mtr.EventReceived()
+
+	if len(tasks) > 0 {
+		h.notifier.Notify(tasks)
+	}
+
+	return event, tasks, nil
 }
 
 // inlineBody returns a pointer to body as a string if it fits
@@ -341,18 +403,13 @@ func inlineBody(body []byte) *string {
 	return nil
 }
 
-// finishWebhookResponse notifies the delivery engine, logs the
-// event, and writes the HTTP response.
+// finishWebhookResponse logs the event and writes the HTTP response.
 func (h *Handlers) finishWebhookResponse(
 	w http.ResponseWriter,
 	event *database.Event,
 	entrypoint database.Entrypoint,
 	tasks []delivery.Task,
 ) {
-	if len(tasks) > 0 {
-		h.notifier.Notify(tasks)
-	}
-
 	h.log.Info("webhook event created",
 		"event_id", event.ID,
 		"webhook_id", entrypoint.WebhookID,
@@ -370,33 +427,15 @@ func (h *Handlers) finishWebhookResponse(
 	}
 }
 
-// buildEvent creates a new Event struct from request data.
-func (h *Handlers) buildEvent(
-	r *http.Request,
-	entrypoint database.Entrypoint,
-	headersJSON, body []byte,
-) *database.Event {
-	return &database.Event{
-		WebhookID:    entrypoint.WebhookID,
-		EntrypointID: entrypoint.ID,
-		Method:       r.Method,
-		Headers:      string(headersJSON),
-		Body:         string(body),
-		ContentType:  r.Header.Get("Content-Type"),
-	}
-}
-
-// buildDeliveryTasks creates delivery records in the
-// transaction and returns tasks for the delivery engine.
-// Returns nil if an error occurred.
-func (h *Handlers) buildDeliveryTasks(
-	w http.ResponseWriter,
+// buildDeliveryTasks creates one pending delivery per target in the
+// transaction and returns the tasks for the delivery engine. The
+// caller owns the transaction and rolls it back on error.
+func buildDeliveryTasks(
 	tx *gorm.DB,
 	event *database.Event,
-	entrypoint database.Entrypoint,
 	targets []database.Target,
 	bodyPtr *string,
-) []delivery.Task {
+) ([]delivery.Task, error) {
 	tasks := make([]delivery.Task, 0, len(targets))
 
 	for i := range targets {
@@ -408,25 +447,17 @@ func (h *Handlers) buildDeliveryTasks(
 
 		err := tx.Create(dlv).Error
 		if err != nil {
-			tx.Rollback()
-			h.log.Error(
-				"failed to create delivery",
-				"target_id", targets[i].ID,
-				"error", err,
+			return nil, fmt.Errorf(
+				"creating delivery for target %s: %w",
+				targets[i].ID, err,
 			)
-			http.Error(
-				w, "Internal server error",
-				http.StatusInternalServerError,
-			)
-
-			return nil
 		}
 
 		tasks = append(tasks, delivery.Task{
 			DeliveryID:   dlv.ID,
 			EventID:      event.ID,
-			WebhookID:    entrypoint.WebhookID,
-			EntrypointID: entrypoint.ID,
+			WebhookID:    event.WebhookID,
+			EntrypointID: event.EntrypointID,
 			TargetID:     targets[i].ID,
 			TargetName:   targets[i].Name,
 			TargetType:   targets[i].Type,
@@ -440,5 +471,5 @@ func (h *Handlers) buildDeliveryTasks(
 		})
 	}
 
-	return tasks
+	return tasks, nil
 }