webhooker/internal/handlers/event_resubmit.go

package handlers

import (
	"errors"
	"net/http"

	"github.com/go-chi/chi"
	"github.com/google/uuid"
	"gorm.io/gorm"
	"sneak.berlin/go/webhooker/internal/database"
)

// The outcomes of a resubmit POST, as the notice codes its redirect
// carries. noticeFor holds the line each one shows.
const (
	// resubmitQueued reports that a new event was stored and its
	// deliveries handed to the delivery engine.
	resubmitQueued noticeCode = "resubmit-queued"

	// resubmitNoTargets reports a source with no active targets. The
	// new event is stored either way, exactly as a received event
	// with no targets is.
	resubmitNoTargets noticeCode = "resubmit-no-targets"
)

// resubmitSource is the stored event a resubmit copies. Its body is
// read as bytes rather than as a string so the copy is byte-identical
// to what was received, whatever the payload's encoding.
type resubmitSource struct {
	ID           string
	EntrypointID string
	Method       string
	Headers      string
	ContentType  string
	Body         []byte
}

// resubmitColumns is the projection resubmitSource is loaded through.
// The cast to blob is what makes the driver hand back the stored bytes
// rather than a string conversion, the same reason eventBodyQuery
// casts.
const resubmitColumns = "id, entrypoint_id, method, headers, " +
	"content_type, cast(body as blob) AS body"

// HandleEventResubmit re-injects a stored event as a new undelivered
// event.
//
// This is the testing counterpart to per-delivery replay, and the two
// select targets differently on purpose. A replay re-sends ONE
// finished delivery to ITS OWN target, which is recovery. A resubmit
// stores a NEW event copied from the stored one and fans it out to the
// webhook's currently ACTIVE targets, resolved fresh by the query the
// receiver uses — so a target created after the original event arrived
// receives it, which is what makes capturing real traffic and firing
// it at a backend under development possible. The original event's
// deliveries have no bearing on where the copy goes.
//
// Nothing about the original delivery is re-sent: what is re-injected
// is the stored EVENT. The response bodies and headers the original
// deliveries received stay where they are.
//
// Resubmitting the same event repeatedly is supported and is the point
// of the feature, so replay's in-flight refusal is deliberately not
// applied here. The route's rate limit is what bounds a held-down
// button.
func (h *Handlers) HandleEventResubmit() 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 {
			h.renderError(w, r, http.StatusBadRequest)

			return
		}

		h.resubmitEvent(w, r, webhook)
	}
}

// resubmitEvent performs the resubmit for a webhook the caller has
// already established the session's user owns.
func (h *Handlers) resubmitEvent(
	w http.ResponseWriter,
	r *http.Request,
	webhook database.Webhook,
) {
	// Parsing the id before use keeps a malformed id out of the SQL
	// and makes the value the query sees come from uuid's own fixed
	// alphabet rather than from the request.
	eventID, err := uuid.Parse(chi.URLParam(r, "eventID"))
	if err != nil {
		h.renderError(w, r, http.StatusNotFound)

		return
	}

	if !h.dbMgr.DBExists(webhook.ID) {
		h.renderError(w, r, http.StatusNotFound)

		return
	}

	webhookDB, err := h.dbMgr.GetDB(webhook.ID)
	if err != nil {
		h.serverError(w, r, "failed to get webhook database", err)

		return
	}

	// Read before the write transaction is opened. The body can be up
	// to the 1 MB ingest cap, and every transaction on these files
	// takes the write lock at BEGIN (_txlock=immediate, see
	// internal/database/sqlite_open.go), so reading inside it would
	// hold that lock against the receiver for the length of the read.
	src, found, err := loadResubmitSource(
		webhookDB, webhook.ID, eventID.String(),
	)
	if err != nil {
		h.serverError(w, r, "failed to load event to resubmit", err)

		return
	}

	// A miss is a 404 whether the event was reaped, belongs to
	// another webhook, or never existed.
	if !found {
		h.renderError(w, r, http.StatusNotFound)

		return
	}

	h.queueResubmit(w, r, webhook, src)
}

// loadResubmitSource reads the stored event a resubmit copies, and
// whether it exists within the webhook.
//
// The webhook_id predicate is currently redundant against the
// per-webhook database files — a sibling webhook's event is not in the
// database being queried at all — and is there so the scoping survives
// any future change that puts more than one webhook's events in one
// file. A reaped event is not found because the retention reaper
// deletes its row outright rather than marking it deleted; see
// deleteEvents in internal/database/retention.go.
func loadResubmitSource(
	webhookDB *gorm.DB,
	webhookID, eventID string,
) (resubmitSource, bool, error) {
	var src resubmitSource

	err := webhookDB.Model(&database.Event{}).
		Select(resubmitColumns).
		Where("id = ? AND webhook_id = ?", eventID, webhookID).
		First(&src).Error

	if errors.Is(err, gorm.ErrRecordNotFound) {
		return src, false, nil
	}

	if err != nil {
		return src, false, err
	}

	return src, true, nil
}

// queueResubmit stores the copy and fans it out to the webhook's
// active targets.
func (h *Handlers) queueResubmit(
	w http.ResponseWriter,
	r *http.Request,
	webhook database.Webhook,
	src resubmitSource,
) {
	// The receiver's own query, run now: an active target created
	// after the original event arrived is included, and an
	// inactive one is skipped rather than refused.
	targets, err := h.loadActiveTargets(webhook.ID)
	if err != nil {
		h.serverError(w, r, "failed to query targets", err)

		return
	}

	event, tasks, err := h.createAndFanOut(
		eventSource{
			WebhookID:         webhook.ID,
			EntrypointID:      src.EntrypointID,
			Method:            src.Method,
			HeadersJSON:       src.Headers,
			ContentType:       src.ContentType,
			Body:              src.Body,
			ResubmittedFromID: &src.ID,
		},
		targets,
	)
	if err != nil {
		h.serverError(w, r, "failed to store resubmitted event", err)

		return
	}

	h.mtr.EventResubmitted()

	h.log.Info(
		"event resubmitted",
		"webhook_id", webhook.ID,
		"event_id", event.ID,
		"resubmitted_from_id", src.ID,
		"target_count", len(tasks),
	)

	code := resubmitQueued
	if len(tasks) == 0 {
		code = resubmitNoTargets
	}

	redirectToEventLog(w, r, webhook, code)
}