Add per-delivery replay to the event log (closes #203) (#240)

There was no redelivery path anywhere: once a delivery exhausted
max_retries it was failed permanently, even though the event body is
durably stored. Storing an event and being unable to re-send it defeats
the reason it is stored, and the ordinary case is a destination that was
down longer than the backoff ladder.

Adds POST /source/{sourceID}/deliveries/{deliveryID}/replay, inside the
authenticated group so it inherits MaxBodySize, CSRF, NoCache and
RequireAuth. Replay creates a NEW pending delivery against the target's
CURRENT config and hands it to the engine through the same notifier the
receiver uses, so it runs the normal path with the retry ladder, the
SSRF-guarded transport and the circuit breaker. The original delivery's
rows are never touched, and the stored event body is re-sent, never the
recorded response.

Replay is refused, with a distinct message, for a non-terminal delivery, a
deleted target, a deactivated target, and when an earlier replay of the
same event and target is still in flight. Bounded by a per-client rate
limit and by that in-flight check.

The new delivery row is written with Omit(clause.Associations) and with
neither Event nor Target populated, so it cannot upsert a targets row into
the per-webhook event database (#206).

Counted by webhooker_delivery_replays_total on the existing target_type
label. A replay also moves the ordinary attempt, outcome and duration
series, because it is a real delivery.

internal/handlers/delivery_replay.go

@@ -0,0 +1,378 @@
+package handlers
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/go-chi/chi"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/delivery"
+)
+
+// replayOutcomeParam is the query parameter the replay POST redirects
+// with and the event log page reads its banner from.
+const replayOutcomeParam = "replay"
+
+// replayOutcomeCode is the outcome of a replay POST. The redirect
+// carries one of these fixed codes rather than a message, so nothing a
+// client submits can reach the rendered page through it.
+type replayOutcomeCode string
+
+const (
+	// replayQueued reports that a new delivery was created and handed
+	// to the delivery engine.
+	replayQueued replayOutcomeCode = "queued"
+
+	// replayTargetDeleted reports a target that once existed and has
+	// since been deleted. Deletes are soft and deliveries carry no
+	// foreign key to the target row, so the history survives its
+	// target and this is the ordinary case for an old event.
+	replayTargetDeleted replayOutcomeCode = "target-deleted"
+
+	// replayTargetMissing reports a target id that names no row at
+	// all, deleted or otherwise.
+	replayTargetMissing replayOutcomeCode = "target-missing"
+
+	// replayTargetInactive reports a target the operator has
+	// deactivated. A deactivated target receives no new deliveries, so
+	// a replay to it would be a delivery they switched off.
+	replayTargetInactive replayOutcomeCode = "target-inactive"
+
+	// replayNotTerminal reports a delivery the engine has not finished
+	// with.
+	replayNotTerminal replayOutcomeCode = "not-terminal"
+
+	// replayInFlight reports that an earlier replay of this event to
+	// this target is still running.
+	replayInFlight replayOutcomeCode = "in-flight"
+)
+
+// replayOutcome returns the banner the event log page shows for an
+// outcome code, and whether the replay was queued. An unrecognised
+// code yields no banner.
+func replayOutcome(code string) (string, bool) {
+	switch replayOutcomeCode(code) {
+	case replayQueued:
+		return "Replay queued: a new delivery was created against " +
+			"the target's current configuration.", true
+	case replayTargetDeleted:
+		return "Not replayed: the target this delivery was for has " +
+			"been deleted. Recreate the target, then replay.", false
+	case replayTargetMissing:
+		return "Not replayed: the target this delivery was for no " +
+			"longer exists.", false
+	case replayTargetInactive:
+		return "Not replayed: the target this delivery was for is " +
+			"deactivated. Activate it, then replay.", false
+	case replayNotTerminal:
+		return "Not replayed: this delivery has not finished yet.",
+			false
+	case replayInFlight:
+		return "Not replayed: a delivery of this event to this " +
+			"target is already in flight.", false
+	default:
+		return "", false
+	}
+}
+
+// HandleDeliveryReplay re-sends a finished delivery's event to its
+// target.
+//
+// A replay never touches the delivery it repeats. It creates a NEW
+// pending delivery row for the same event and target and hands it to
+// the delivery engine through the same Notifier the receiver uses, so
+// the original's status, attempts and timestamps stand as the record
+// of what actually happened, and the replay is retried, SSRF-guarded
+// and circuit-broken exactly as a first attempt is.
+//
+// What is re-sent is the stored EVENT body, never the response the
+// original delivery received.
+//
+// The target's configuration is read now rather than as it stood when
+// the original ran: a replay exists to deliver where the operator
+// currently wants the event to go. That is also why a deleted target
+// is refused rather than delivered to from stale configuration.
+func (h *Handlers) HandleDeliveryReplay() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		webhook, ok := h.ownedWebhook(w, r)
+		if !ok {
+			return
+		}
+
+		// The body size cap is enforced by the MaxBodySize
+		// middleware, which runs before CSRF parses the form.
+		err := r.ParseForm()
+		if err != nil {
+			http.Error(
+				w, "Bad request", http.StatusBadRequest,
+			)
+
+			return
+		}
+
+		h.replayDelivery(w, r, webhook)
+	}
+}
+
+// replayDelivery performs the replay for a webhook the caller has
+// already established the session's user owns.
+func (h *Handlers) replayDelivery(
+	w http.ResponseWriter,
+	r *http.Request,
+	webhook database.Webhook,
+) {
+	if !h.dbMgr.DBExists(webhook.ID) {
+		http.NotFound(w, r)
+
+		return
+	}
+
+	webhookDB, err := h.dbMgr.GetDB(webhook.ID)
+	if err != nil {
+		h.serverError(w, "failed to get webhook database", err)
+
+		return
+	}
+
+	original, ok := h.loadReplaySource(w, r, webhookDB)
+	if !ok {
+		return
+	}
+
+	if !original.Status.Terminal() {
+		h.finishReplay(w, r, webhook, replayNotTerminal)
+
+		return
+	}
+
+	target, code := h.replayTarget(webhook.ID, original.TargetID)
+	if target == nil {
+		h.finishReplay(w, r, webhook, code)
+
+		return
+	}
+
+	h.queueReplay(w, r, webhookDB, webhook, original, target)
+}
+
+// loadReplaySource loads the delivery to be replayed, selecting only
+// the columns the replay needs so no association is populated. A
+// delivery id that names no row in this webhook's database is a 404.
+func (h *Handlers) loadReplaySource(
+	w http.ResponseWriter,
+	r *http.Request,
+	webhookDB *gorm.DB,
+) (*database.Delivery, bool) {
+	var original database.Delivery
+
+	err := webhookDB.
+		Select("id", "event_id", "target_id", "status").
+		First(
+			&original, "id = ?", chi.URLParam(r, "deliveryID"),
+		).Error
+	if err != nil {
+		http.NotFound(w, r)
+
+		return nil, false
+	}
+
+	return &original, true
+}
+
+// queueReplay writes the new delivery and hands it to the engine.
+func (h *Handlers) queueReplay(
+	w http.ResponseWriter,
+	r *http.Request,
+	webhookDB *gorm.DB,
+	webhook database.Webhook,
+	original *database.Delivery,
+	target *database.Target,
+) {
+	inFlight, err := countInFlightDeliveries(
+		webhookDB, original.EventID, target.ID,
+	)
+	if err != nil {
+		h.serverError(
+			w, "failed to count in-flight deliveries", err,
+		)
+
+		return
+	}
+
+	if inFlight > 0 {
+		h.finishReplay(w, r, webhook, replayInFlight)
+
+		return
+	}
+
+	var event database.Event
+
+	err = webhookDB.
+		First(&event, "id = ?", original.EventID).Error
+	if err != nil {
+		h.serverError(w, "failed to load event for replay", err)
+
+		return
+	}
+
+	task, err := createReplayDelivery(
+		webhookDB, webhook.ID, &event, target,
+	)
+	if err != nil {
+		h.serverError(
+			w, "failed to create replay delivery", err,
+		)
+
+		return
+	}
+
+	h.mtr.DeliveryReplayed(target.Type)
+	h.notifier.Notify([]delivery.Task{task})
+
+	h.log.Info(
+		"delivery replay queued",
+		"webhook_id", webhook.ID,
+		"event_id", event.ID,
+		"target_id", target.ID,
+		"replayed_delivery_id", original.ID,
+		"delivery_id", task.DeliveryID,
+	)
+
+	h.finishReplay(w, r, webhook, replayQueued)
+}
+
+// replayTarget loads the delivery's target as it stands now.
+//
+// The load is Unscoped so that a soft-deleted row is still found:
+// deletes are soft and a delivery carries no foreign key to its
+// target, so a target's history outlives it, and without the deleted
+// row there is no way to tell "you deleted this target" from "this id
+// never named anything". A nil target means the replay is refused,
+// with the returned code saying why.
+func (h *Handlers) replayTarget(
+	webhookID, targetID string,
+) (*database.Target, replayOutcomeCode) {
+	var target database.Target
+
+	err := h.db.DB().Unscoped().Where(
+		"id = ? AND webhook_id = ?", targetID, webhookID,
+	).First(&target).Error
+	if err != nil {
+		return nil, replayTargetMissing
+	}
+
+	if target.DeletedAt.Valid {
+		return nil, replayTargetDeleted
+	}
+
+	if !target.Active {
+		return nil, replayTargetInactive
+	}
+
+	return &target, replayQueued
+}
+
+// countInFlightDeliveries reports how many deliveries of this event to
+// this target the engine has not finished.
+//
+// It is the replay-storm guard: a replay is refused while an earlier
+// one is still pending or retrying, so a held-down button or a scripted
+// loop cannot stack copies of work already queued. It is a check and
+// not a lock, so two simultaneous POSTs can still both pass it; the
+// per-client rate limit on the route is what bounds that.
+func countInFlightDeliveries(
+	webhookDB *gorm.DB, eventID, targetID string,
+) (int64, error) {
+	var count int64
+
+	err := webhookDB.Model(&database.Delivery{}).Where(
+		"event_id = ? AND target_id = ? AND status IN ?",
+		eventID, targetID,
+		[]database.DeliveryStatus{
+			database.DeliveryStatusPending,
+			database.DeliveryStatusRetrying,
+		},
+	).Count(&count).Error
+
+	return count, err
+}
+
+// createReplayDelivery writes the new pending delivery row and returns
+// the task that carries it to the delivery engine.
+//
+// The row is written with associations omitted, and neither Event nor
+// Target is populated on it: GORM's SaveBeforeAssociations would
+// otherwise upsert the whole target row — plaintext config, which for a
+// Slack target is the credential — into the per-webhook event database.
+// See https://git.eeqj.de/sneak/webhooker/issues/206.
+func createReplayDelivery(
+	webhookDB *gorm.DB,
+	webhookID string,
+	event *database.Event,
+	target *database.Target,
+) (delivery.Task, error) {
+	dlv := &database.Delivery{
+		EventID:  event.ID,
+		TargetID: target.ID,
+		Status:   database.DeliveryStatusPending,
+	}
+
+	err := webhookDB.Omit(clause.Associations).Create(dlv).Error
+	if err != nil {
+		return delivery.Task{}, err
+	}
+
+	return delivery.Task{
+		DeliveryID:   dlv.ID,
+		EventID:      event.ID,
+		WebhookID:    webhookID,
+		EntrypointID: event.EntrypointID,
+		TargetID:     target.ID,
+		TargetName:   target.Name,
+		TargetType:   target.Type,
+		TargetConfig: target.Config,
+		MaxRetries:   target.MaxRetries,
+		Method:       event.Method,
+		Headers:      event.Headers,
+		ContentType:  event.ContentType,
+		Body:         replayBody(event.Body),
+		AttemptNum:   1,
+	}, nil
+}
+
+// replayBody returns the stored event body for a replay task to carry
+// inline, or nil when it is large enough that the engine should fetch
+// it from the per-webhook database instead.
+func replayBody(body string) *string {
+	if len(body) >= delivery.MaxInlineBodySize {
+		return nil
+	}
+
+	return &body
+}
+
+// finishReplay redirects back to the event log the replay was
+// triggered from, carrying the outcome code the page turns into a
+// banner and the page number the form submitted.
+func (h *Handlers) finishReplay(
+	w http.ResponseWriter,
+	r *http.Request,
+	webhook database.Webhook,
+	code replayOutcomeCode,
+) {
+	dest := "/source/" + webhook.ID + "/logs?" +
+		replayOutcomeParam + "=" + string(code)
+
+	// The page is read from the form rather than the query string:
+	// this is a POST, and its query string is what logs and Referer
+	// headers record.
+	if page := parseNonNegativeInt(
+		r.PostFormValue("page"),
+	); page > 1 {
+		dest += "&page=" + strconv.Itoa(page)
+	}
+
+	http.Redirect(w, r, dest, http.StatusSeeOther)
+}