webhooker/internal/handlers/delivery_replay.go

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)
}