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

Capturing real webhook traffic and firing it repeatedly at a backend
under development is a primary function of this service, and
per-delivery replay cannot do it: it only ever resolves the delivery's
own original target, so a target created for a dev backend has no
prior delivery and nothing can be replayed to it.

The event log now offers a per-event Resubmit action. It stores a NEW
event copying the stored one's method, headers, body and content type
verbatim, and fans it out to the webhook's currently ACTIVE targets,
resolved fresh by the query the receiver uses -- so a target created
long after the original event arrived receives it. The original
event's deliveries have no bearing on where the copy goes, inactive
targets are skipped as the receiver skips them, and the action is
repeatable: replay's in-flight refusal is deliberately not ported,
because firing one captured event over and over is the point.

The receiver and the resubmit path share one construction and one
fan-out site. An eventSource value carries where the fields came from,
live request or stored event, and createAndFanOut writes the event and
its pending deliveries in one transaction and hands the tasks to the
same Notifier, so a resubmitted delivery is retried, SSRF-guarded and
circuit-broken exactly as a first one is. buildDeliveryTasks returns
an error instead of writing a response, which is what lets both
callers share it.

The stored event is read once, before the write transaction, with a
cast to blob, so a body over delivery.MaxInlineBodySize is copied byte
for byte and the engine loads it from the new event row.

A nullable resubmitted_from_id records provenance -- empty for an
event that arrived on the receiver -- and the event log reports the
relationship in both directions, without which the log is unreadable
after a few resubmits of one event. The route sits in the owned-source
group, so auth, CSRF and the body cap apply, with its own rate limit
bucket and an events_resubmitted_total counter.

Inbound signature verification is not re-run: there is no inbound
signature to check on a copy an authenticated, CSRF-protected operator
action submits.

Per-delivery replay is unchanged; it serves recovery, which resubmit
does not replace. The README claimed in four places that replay was
unimplemented, one of them telling the operator that a delivery
stranded by a target type change was lost; all four are corrected and
resubmit is documented beside replay.

internal/handlers/event_log_view.go

@@ -20,6 +20,7 @@ const maxRenderedBodyBytes = 8192
 // rather than in Go is the point of the projection — an
 // oversized body never becomes a Go string at all.
 const eventLogColumns = "id, created_at, method, content_type, " +
+	"resubmitted_from_id, " +
 	"substr(cast(body as blob), 1, ?) AS body, " +
 	"length(cast(body as blob)) AS body_bytes"
 
@@ -45,9 +46,25 @@ type EventLogView struct {
 	// than the cap, so the page owes the reader a marker.
 	BodyTruncated bool
 
+	// ResubmittedFromID names the event this one was copied
+	// from, empty for an event that arrived on the receiver.
+	ResubmittedFromID string
+
+	// ResubmitCount is how many events have been resubmitted
+	// from this one. Both directions are shown, because after
+	// a few resubmits of one captured event the log is
+	// otherwise a row of identical bodies with nothing saying
+	// which came from which.
+	ResubmitCount int
+
 	Deliveries []DeliveryView
 }
 
+// ResubmittedFrom reports that this event is a copy of another.
+func (v EventLogView) ResubmittedFrom() bool {
+	return v.ResubmittedFromID != ""
+}
+
 // BodyShownBytes is how many body bytes the page is actually
 // rendering, which the truncation marker reports beside the
 // true size.
@@ -59,12 +76,13 @@ func (v EventLogView) BodyShownBytes() int {
 // body column arrives already cut to the cap by SQLite, with
 // the true size beside it.
 type eventLogRow struct {
-	ID          string
-	CreatedAt   time.Time
-	Method      string
-	ContentType string
-	Body        []byte
-	BodyBytes   int64
+	ID                string
+	CreatedAt         time.Time
+	Method            string
+	ContentType       string
+	ResubmittedFromID *string
+	Body              []byte
+	BodyBytes         int64
 }
 
 // view projects a loaded row for rendering.
@@ -79,14 +97,20 @@ func (r *eventLogRow) view() EventLogView {
 		body = trimPartialRune(body)
 	}
 
+	var from string
+	if r.ResubmittedFromID != nil {
+		from = *r.ResubmittedFromID
+	}
+
 	return EventLogView{
-		ID:            r.ID,
-		CreatedAt:     r.CreatedAt,
-		Method:        r.Method,
-		ContentType:   r.ContentType,
-		Body:          string(body),
-		BodyBytes:     r.BodyBytes,
-		BodyTruncated: truncated,
+		ID:                r.ID,
+		CreatedAt:         r.CreatedAt,
+		Method:            r.Method,
+		ContentType:       r.ContentType,
+		Body:              string(body),
+		BodyBytes:         r.BodyBytes,
+		BodyTruncated:     truncated,
+		ResubmittedFromID: from,
 	}
 }
 

internal/handlers/event_resubmit.go

@@ -0,0 +1,277 @@
+package handlers
+
+import (
+	"errors"
+	"net/http"
+	"strconv"
+
+	"github.com/go-chi/chi"
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+	"sneak.berlin/go/webhooker/internal/database"
+)
+
+// resubmitOutcomeParam is the query parameter the resubmit POST
+// redirects with and the event log page reads its banner from.
+const resubmitOutcomeParam = "resubmit"
+
+// resubmitOutcomeCode is the outcome of a resubmit 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 resubmitOutcomeCode string
+
+const (
+	// resubmitQueued reports that a new event was stored and its
+	// deliveries handed to the delivery engine.
+	resubmitQueued resubmitOutcomeCode = "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 resubmitOutcomeCode = "no-targets"
+)
+
+// resubmitOutcome returns the banner the event log page shows for an
+// outcome code, and whether the resubmit was queued. An unrecognised
+// code yields no banner.
+func resubmitOutcome(code string) (string, bool) {
+	switch resubmitOutcomeCode(code) {
+	case resubmitQueued:
+		return "Resubmitted: a new event was created from the stored " +
+			"one and queued to every active target.", true
+	case resubmitNoTargets:
+		return "Resubmitted: a new event was created, but this " +
+				"source has no active targets, so nothing was queued.",
+			true
+	default:
+		return "", false
+	}
+}
+
+// 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.
+//
+// Inbound signature verification is deliberately not re-run. There is
+// no inbound signature to check on a copy the operator submits; the
+// route is authenticated and CSRF-protected as an operator action.
+//
+// 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 {
+			http.Error(w, "Bad request", 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 {
+		http.NotFound(w, r)
+
+		return
+	}
+
+	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
+	}
+
+	// Read before the write transaction is opened. The body can be up
+	// to the 1 MB ingest cap, and holding a read of it inside the
+	// transaction would extend how long the per-webhook database is
+	// locked against the receiver, which runs these files in
+	// SQLite's default journal mode rather than WAL.
+	src, found, err := loadResubmitSource(
+		webhookDB, webhook.ID, eventID.String(),
+	)
+	if err != nil {
+		h.serverError(w, "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 {
+		http.NotFound(w, r)
+
+		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. Going through Model applies GORM's soft-delete scope, which is
+// what stops a reaped event being resubmitted.
+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, "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, "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
+	}
+
+	h.finishResubmit(w, r, webhook, code)
+}
+
+// finishResubmit redirects back to the event log the resubmit was
+// triggered from, carrying the outcome code the page turns into a
+// banner and the page number the form submitted.
+func (h *Handlers) finishResubmit(
+	w http.ResponseWriter,
+	r *http.Request,
+	webhook database.Webhook,
+	code resubmitOutcomeCode,
+) {
+	dest := "/source/" + webhook.ID + "/logs?" +
+		resubmitOutcomeParam + "=" + 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)
+}

internal/handlers/event_resubmit_test.go

@@ -0,0 +1,603 @@
+package handlers_test
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/delivery"
+	"sneak.berlin/go/webhooker/internal/handlers"
+	"sneak.berlin/go/webhooker/internal/session"
+)
+
+// resubmitTargetURL is a public destination, so a target configured
+// with it is one the SSRF guard would accept. Nothing in these tests
+// dispatches to it: the notifier is recorded, not run.
+const resubmitTargetURL = "http://93.184.216.34/hook"
+
+// resubmitEventHeaders is the stored header JSON a seeded event
+// carries, so a test can prove the copy takes it verbatim.
+const resubmitEventHeaders = `{"X-Test":["yes"],"X-Trace":["abc"]}`
+
+// seedStoredEvent records one event in a webhook's own database with
+// no deliveries at all, which is the state a captured event is in when
+// the operator has yet to create the target to test.
+func seedStoredEvent(
+	t *testing.T,
+	dbMgr *database.WebhookDBManager,
+	webhookID, body string,
+) *database.Event {
+	t.Helper()
+
+	webhookDB, err := dbMgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	event := &database.Event{
+		WebhookID:    webhookID,
+		EntrypointID: "entrypoint-" + webhookID,
+		Method:       http.MethodPost,
+		Headers:      resubmitEventHeaders,
+		Body:         body,
+		ContentType:  contentTypeJSON,
+	}
+
+	require.NoError(t, webhookDB.Omit(
+		clause.Associations,
+	).Create(event).Error)
+
+	return event
+}
+
+// postResubmit runs the real resubmit handler for one event.
+func postResubmit(
+	t *testing.T,
+	h *handlers.Handlers,
+	sess *session.Session,
+	webhookID, eventID string,
+) *httptest.ResponseRecorder {
+	t.Helper()
+
+	req := postRequest(
+		"/source/"+webhookID+"/events/"+eventID+"/resubmit",
+		authenticatedCookies(
+			t, sess, deleteTestUserID, deleteTestUsername,
+		),
+		map[string]string{
+			paramSourceID: webhookID,
+			paramEventID:  eventID,
+		},
+	)
+
+	w := httptest.NewRecorder()
+	h.HandleEventResubmit().ServeHTTP(w, req)
+
+	return w
+}
+
+// listEvents reads every event in a webhook's database, oldest first.
+func listEvents(
+	t *testing.T, webhookDB *gorm.DB,
+) []database.Event {
+	t.Helper()
+
+	var events []database.Event
+
+	require.NoError(t, webhookDB.
+		Order("created_at ASC, id ASC").
+		Find(&events).Error)
+
+	return events
+}
+
+// theOtherEvent returns the one event that is not excludeID.
+func theOtherEvent(
+	t *testing.T, events []database.Event, excludeID string,
+) database.Event {
+	t.Helper()
+
+	var found []database.Event
+
+	for _, e := range events {
+		if e.ID != excludeID {
+			found = append(found, e)
+		}
+	}
+
+	require.Len(t, found, 1)
+
+	return found[0]
+}
+
+// TestHandleEventResubmit_DeliversToTargetCreatedAfterTheEvent is the
+// core of the feature and the thing per-delivery replay cannot do: the
+// event was captured before the target existed, so it has no delivery
+// to replay, and the resubmit must still reach the new target.
+func TestHandleEventResubmit_DeliversToTargetCreatedAfterTheEvent(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+		notif *recordingNotifier
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &notif)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	wh := seedWebhook(t, db)
+	original := seedStoredEvent(
+		t, dbMgr, wh.ID, `{"captured":"traffic"}`,
+	)
+
+	// The dev backend is registered only now, after the traffic was
+	// captured. It has no prior delivery of anything.
+	tgt := seedConfiguredTarget(
+		t, db, wh.ID, database.TargetTypeHTTP,
+		`{"url":"`+resubmitTargetURL+`"}`,
+	)
+
+	w := postResubmit(t, h, sess, wh.ID, original.ID)
+
+	require.Equal(t, http.StatusSeeOther, w.Code)
+	assert.Equal(
+		t,
+		"/source/"+wh.ID+"/logs?resubmit=queued",
+		w.Header().Get("Location"),
+	)
+
+	webhookDB, err := dbMgr.GetDB(wh.ID)
+	require.NoError(t, err)
+
+	events := listEvents(t, webhookDB)
+	require.Len(
+		t, events, 2,
+		"resubmit must create a new event, not mark the old one",
+	)
+
+	fresh := theOtherEvent(t, events, original.ID)
+	assertEventCopy(t, original, fresh)
+
+	// The delivery hangs off the NEW event, and the original event
+	// still has none.
+	assert.Empty(
+		t, listDeliveries(t, webhookDB, original.ID),
+		"the original event must be left untouched",
+	)
+
+	deliveries := listDeliveries(t, webhookDB, fresh.ID)
+	require.Len(t, deliveries, 1)
+	assert.Equal(t, tgt.ID, deliveries[0].TargetID)
+	assert.Equal(
+		t, database.DeliveryStatusPending, deliveries[0].Status,
+	)
+
+	tasks := notif.Tasks()
+	require.Len(t, tasks, 1)
+	assert.Equal(t, deliveries[0].ID, tasks[0].DeliveryID)
+	assertResubmitTask(t, tasks[0], wh.ID, &fresh, tgt)
+
+	assertNoLeakedTarget(t, webhookDB)
+}
+
+// assertEventCopy proves the resubmitted event copies every stored
+// field of the event it came from, and records the provenance that
+// keeps the log readable. It also pins that a received event carries
+// no source event of its own.
+func assertEventCopy(
+	t *testing.T, original *database.Event, fresh database.Event,
+) {
+	t.Helper()
+
+	assert.Equal(t, original.Method, fresh.Method)
+	assert.Equal(t, original.Headers, fresh.Headers)
+	assert.Equal(t, original.Body, fresh.Body)
+	assert.Equal(t, original.ContentType, fresh.ContentType)
+	assert.Equal(t, original.EntrypointID, fresh.EntrypointID)
+	assert.Equal(t, original.WebhookID, fresh.WebhookID)
+	assert.NotEqual(t, original.ID, fresh.ID)
+
+	require.NotNil(t, fresh.ResubmittedFromID)
+	assert.Equal(t, original.ID, *fresh.ResubmittedFromID)
+
+	assert.Nil(
+		t, original.ResubmittedFromID,
+		"a received event records no source event",
+	)
+}
+
+// assertResubmitTask proves the task handed to the delivery engine is
+// the one the receiver would build for the NEW event and this target.
+func assertResubmitTask(
+	t *testing.T,
+	task delivery.Task,
+	webhookID string,
+	fresh *database.Event,
+	target *database.Target,
+) {
+	t.Helper()
+
+	assert.Equal(t, fresh.ID, task.EventID)
+	assert.Equal(t, webhookID, task.WebhookID)
+	assert.Equal(t, fresh.EntrypointID, task.EntrypointID)
+	assert.Equal(t, target.ID, task.TargetID)
+	assert.Equal(t, target.Type, task.TargetType)
+	assert.Equal(t, fresh.Method, task.Method)
+	assert.Equal(t, fresh.Headers, task.Headers)
+	assert.Equal(t, fresh.ContentType, task.ContentType)
+	assert.Equal(t, 1, task.AttemptNum)
+
+	require.NotNil(t, task.Body)
+	assert.Equal(t, fresh.Body, *task.Body)
+}
+
+// TestHandleEventResubmit_IsRepeatable proves the requirement replay
+// deliberately does not meet: firing the same captured event at a
+// backend over and over must work, with no in-flight refusal, and each
+// press must produce its own event and its own delivery.
+func TestHandleEventResubmit_IsRepeatable(t *testing.T) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+		notif *recordingNotifier
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &notif)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	wh := seedWebhook(t, db)
+	tgt := seedConfiguredTarget(
+		t, db, wh.ID, database.TargetTypeHTTP,
+		`{"url":"`+resubmitTargetURL+`"}`,
+	)
+	original := seedStoredEvent(t, dbMgr, wh.ID, `{"fire":"again"}`)
+
+	// Nothing between the presses marks the earlier deliveries
+	// finished, so every one of these is submitted while the last is
+	// still pending.
+	const presses = 5
+
+	for range presses {
+		w := postResubmit(t, h, sess, wh.ID, original.ID)
+
+		require.Equal(t, http.StatusSeeOther, w.Code)
+		assert.Equal(
+			t,
+			"/source/"+wh.ID+"/logs?resubmit=queued",
+			w.Header().Get("Location"),
+			"a resubmit must not be refused while an earlier "+
+				"one is in flight",
+		)
+	}
+
+	webhookDB, err := dbMgr.GetDB(wh.ID)
+	require.NoError(t, err)
+
+	events := listEvents(t, webhookDB)
+	require.Len(t, events, presses+1)
+
+	tasks := notif.Tasks()
+	require.Len(t, tasks, presses)
+
+	seen := make(map[string]struct{}, presses)
+
+	for _, task := range tasks {
+		assert.Equal(t, tgt.ID, task.TargetID)
+		assert.NotEqual(
+			t, original.ID, task.EventID,
+			"each resubmit delivers its own new event",
+		)
+
+		_, dup := seen[task.EventID]
+		assert.False(t, dup, "each resubmit creates its own event")
+
+		seen[task.EventID] = struct{}{}
+
+		require.Len(t, listDeliveries(t, webhookDB, task.EventID), 1)
+	}
+
+	// Every copy names the same source event, so twenty presses stay
+	// traceable to the one captured request.
+	for _, e := range events {
+		if e.ID == original.ID {
+			continue
+		}
+
+		require.NotNil(t, e.ResubmittedFromID)
+		assert.Equal(t, original.ID, *e.ResubmittedFromID)
+	}
+}
+
+// TestHandleEventResubmit_OversizeBodySurvivesIntact covers the
+// non-inline case: a body above delivery.MaxInlineBodySize is not
+// carried on the task at all, so it has to be copied into the new
+// event row byte-identically for the engine to load it from there.
+func TestHandleEventResubmit_OversizeBodySurvivesIntact(t *testing.T) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+		notif *recordingNotifier
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &notif)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	// Well over the inline limit, and not text: a multibyte rune, a
+	// NUL and a byte that is not valid UTF-8, so a copy that went
+	// through a re-encode or a truncation is visible in the compare.
+	const sentinel = "TAIL-SENTINEL-1f4a9c"
+
+	stored := strings.Repeat("A", delivery.MaxInlineBodySize) +
+		"é\x00\xff" +
+		strings.Repeat("B", 4096) + sentinel
+
+	require.Greater(t, len(stored), delivery.MaxInlineBodySize)
+
+	wh := seedWebhook(t, db)
+	seedConfiguredTarget(
+		t, db, wh.ID, database.TargetTypeHTTP,
+		`{"url":"`+resubmitTargetURL+`"}`,
+	)
+	original := seedStoredEvent(t, dbMgr, wh.ID, stored)
+
+	w := postResubmit(t, h, sess, wh.ID, original.ID)
+	require.Equal(t, http.StatusSeeOther, w.Code)
+
+	webhookDB, err := dbMgr.GetDB(wh.ID)
+	require.NoError(t, err)
+
+	fresh := theOtherEvent(t, listEvents(t, webhookDB), original.ID)
+
+	assert.Len(t, fresh.Body, len(stored))
+	assert.Equal(
+		t, stored, fresh.Body,
+		"the stored body must be copied byte for byte",
+	)
+
+	tasks := notif.Tasks()
+	require.Len(t, tasks, 1)
+	assert.Nil(
+		t, tasks[0].Body,
+		"a body over the inline limit is fetched from the new "+
+			"event row rather than carried on the task",
+	)
+
+	// The engine's own read of the body, against the new event id:
+	// what it would send is what was stored.
+	var loaded database.Event
+
+	require.NoError(t, webhookDB.Select("body").
+		First(&loaded, "id = ?", tasks[0].EventID).Error)
+	assert.Equal(t, stored, loaded.Body)
+}
+
+// TestHandleEventResubmit_SkipsInactiveTarget proves a deactivated
+// target is skipped exactly as the receiver skips it — not an error,
+// and not a delivery the operator switched off.
+func TestHandleEventResubmit_SkipsInactiveTarget(t *testing.T) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+		notif *recordingNotifier
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &notif)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	wh := seedWebhook(t, db)
+	active := seedConfiguredTarget(
+		t, db, wh.ID, database.TargetTypeHTTP,
+		`{"url":"`+resubmitTargetURL+`"}`,
+	)
+	off := seedConfiguredTarget(
+		t, db, wh.ID, database.TargetTypeHTTP,
+		`{"url":"`+resubmitTargetURL+`/off"}`,
+	)
+
+	require.NoError(t, db.DB().Model(&database.Target{}).
+		Where("id = ?", off.ID).
+		Update("active", false).Error)
+
+	original := seedStoredEvent(t, dbMgr, wh.ID, `{"skip":"one"}`)
+
+	w := postResubmit(t, h, sess, wh.ID, original.ID)
+
+	require.Equal(t, http.StatusSeeOther, w.Code)
+	assert.Equal(
+		t,
+		"/source/"+wh.ID+"/logs?resubmit=queued",
+		w.Header().Get("Location"),
+		"an inactive target is skipped, not an error",
+	)
+
+	tasks := notif.Tasks()
+	require.Len(t, tasks, 1)
+	assert.Equal(t, active.ID, tasks[0].TargetID)
+
+	webhookDB, err := dbMgr.GetDB(wh.ID)
+	require.NoError(t, err)
+
+	fresh := theOtherEvent(t, listEvents(t, webhookDB), original.ID)
+	require.Len(t, listDeliveries(t, webhookDB, fresh.ID), 1)
+}
+
+// TestHandleEventResubmit_NoActiveTargetsStillStoresEvent proves a
+// source with nothing to deliver to behaves as the receiver does: the
+// event is stored, nothing is queued, and the operator is told so
+// rather than being shown an error.
+func TestHandleEventResubmit_NoActiveTargetsStillStoresEvent(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+		notif *recordingNotifier
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &notif)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	wh := seedWebhook(t, db)
+	original := seedStoredEvent(t, dbMgr, wh.ID, `{"no":"targets"}`)
+
+	w := postResubmit(t, h, sess, wh.ID, original.ID)
+
+	require.Equal(t, http.StatusSeeOther, w.Code)
+	assert.Equal(
+		t,
+		"/source/"+wh.ID+"/logs?resubmit=no-targets",
+		w.Header().Get("Location"),
+	)
+
+	assert.Empty(t, notif.Tasks())
+
+	webhookDB, err := dbMgr.GetDB(wh.ID)
+	require.NoError(t, err)
+
+	assert.Len(t, listEvents(t, webhookDB), 2)
+}
+
+// TestHandleEventResubmit_RefusesEventOfAnotherWebhook proves the
+// route cannot re-inject an event out of a webhook the session's user
+// does not own, and reports the same 404 for an id that names nothing.
+func TestHandleEventResubmit_RefusesEventOfAnotherWebhook(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+		notif *recordingNotifier
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &notif)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	theirs := seedWebhookFor(t, db, otherTestUserID)
+	theirEvent := seedStoredEvent(t, dbMgr, theirs.ID, `{"not":"mine"}`)
+
+	mine := seedWebhook(t, db)
+	seedConfiguredTarget(
+		t, db, mine.ID, database.TargetTypeHTTP,
+		`{"url":"`+resubmitTargetURL+`"}`,
+	)
+	seedStoredEvent(t, dbMgr, mine.ID, `{"mine":true}`)
+
+	// Their webhook, as its owner would address it.
+	w := postResubmit(t, h, sess, theirs.ID, theirEvent.ID)
+	assert.Equal(t, http.StatusNotFound, w.Code)
+
+	// Their event id, addressed through a webhook the user does own.
+	w = postResubmit(t, h, sess, mine.ID, theirEvent.ID)
+	assert.Equal(t, http.StatusNotFound, w.Code)
+
+	// An id that names no event at all.
+	w = postResubmit(t, h, sess, mine.ID, uuid.NewString())
+	assert.Equal(t, http.StatusNotFound, w.Code)
+
+	// A malformed id never reaches the query.
+	w = postResubmit(t, h, sess, mine.ID, "not-a-uuid")
+	assert.Equal(t, http.StatusNotFound, w.Code)
+
+	assert.Empty(
+		t, notif.Tasks(),
+		"a refused resubmit must queue nothing",
+	)
+}
+
+// TestHandleSourceLogs_ShowsResubmitProvenance proves the event log
+// reports the relationship in both directions, which is what keeps it
+// readable once one captured event has been fired repeatedly.
+func TestHandleSourceLogs_ShowsResubmitProvenance(t *testing.T) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	wh := seedWebhook(t, db)
+	seedConfiguredTarget(
+		t, db, wh.ID, database.TargetTypeHTTP,
+		`{"url":"`+resubmitTargetURL+`"}`,
+	)
+	original := seedStoredEvent(t, dbMgr, wh.ID, `{"trace":"me"}`)
+
+	for range 2 {
+		require.Equal(
+			t,
+			http.StatusSeeOther,
+			postResubmit(t, h, sess, wh.ID, original.ID).Code,
+		)
+	}
+
+	webhookDB, err := dbMgr.GetDB(wh.ID)
+	require.NoError(t, err)
+
+	events := listEvents(t, webhookDB)
+	require.Len(t, events, 3)
+
+	body := renderSourceLogsPage(t, h, sess, wh.ID)
+
+	assert.Contains(
+		t, body, "Resubmitted as 2 new events",
+		"the source event must show it has been resubmitted",
+	)
+	assert.Contains(
+		t, body, "Resubmitted from event",
+		"a copy must show where it came from",
+	)
+	assert.Contains(
+		t, body,
+		"/source/"+wh.ID+"/events/"+original.ID+"/resubmit",
+		"the log must offer the resubmit action per event",
+	)
+}

internal/handlers/source_management.go

@@ -821,25 +821,31 @@ func (h *Handlers) HandleSourceLogs() http.HandlerFunc {
 			totalPages++
 		}
 
-		// The banner a replay POST redirected back with. The
-		// message comes from a fixed set keyed by the outcome
-		// code, never from the query string itself.
+		// The banner a replay or resubmit POST redirected back
+		// with. The message comes from a fixed set keyed by the
+		// outcome code, never from the query string itself.
 		replayMsg, replayOK := replayOutcome(
 			r.URL.Query().Get(replayOutcomeParam),
 		)
 
+		resubmitMsg, resubmitOK := resubmitOutcome(
+			r.URL.Query().Get(resubmitOutcomeParam),
+		)
+
 		data := map[string]any{
-			tmplKeyWebhook:  &webhook,
-			"Events":        evts,
-			"ReplayMessage": replayMsg,
-			"ReplayQueued":  replayOK,
-			"Page":          page,
-			"TotalPages":    totalPages,
-			"TotalEvents":   total,
-			"HasPrev":       page > 1,
-			"HasNext":       page < totalPages,
-			"PrevPage":      page - 1,
-			"NextPage":      page + 1,
+			tmplKeyWebhook:    &webhook,
+			"Events":          evts,
+			"ReplayMessage":   replayMsg,
+			"ReplayQueued":    replayOK,
+			"ResubmitMessage": resubmitMsg,
+			"ResubmitQueued":  resubmitOK,
+			"Page":            page,
+			"TotalPages":      totalPages,
+			"TotalEvents":     total,
+			"HasPrev":         page > 1,
+			"HasNext":         page < totalPages,
+			"PrevPage":        page - 1,
+			"NextPage":        page + 1,
 		}
 
 		h.renderTemplate(w, r, "source_logs.html", data)
@@ -925,12 +931,10 @@ func (h *Handlers) loadEventsWithDeliveries(
 	targetMap map[string]eventLogTarget,
 	page int,
 ) ([]EventLogView, int64, bool) {
-	var totalEvents int64
-
 	var result []EventLogView
 
 	if !h.dbMgr.DBExists(webhook.ID) {
-		return result, totalEvents, true
+		return result, 0, true
 	}
 
 	webhookDB, err := h.dbMgr.GetDB(webhook.ID)
@@ -942,29 +946,20 @@ func (h *Handlers) loadEventsWithDeliveries(
 		return nil, 0, false
 	}
 
-	webhookDB.Model(&database.Event{}).Where(
-		"webhook_id = ?", webhook.ID,
-	).Count(&totalEvents)
-
-	offset := (page - 1) * paginationPerPage
-
-	var rows []eventLogRow
-
-	webhookDB.Model(&database.Event{}).Select(
-		eventLogColumns, maxRenderedBodyBytes,
-	).Where(
-		"webhook_id = ?", webhook.ID,
-	).Order("created_at DESC").Offset(offset).Limit(
-		paginationPerPage,
-	).Find(&rows)
+	rows, totalEvents := loadEventLogRows(
+		webhookDB, webhook.ID, page,
+	)
 
 	result = make([]EventLogView, len(rows))
 	eventDeliveries := make([][]database.Delivery, len(rows))
 
 	var deliveryIDs []string
 
+	eventIDs := make([]string, len(rows))
+
 	for i := range rows {
 		result[i] = rows[i].view()
+		eventIDs[i] = rows[i].ID
 
 		webhookDB.Where(
 			"event_id = ?", rows[i].ID,
@@ -988,15 +983,86 @@ func (h *Handlers) loadEventsWithDeliveries(
 		return nil, 0, false
 	}
 
+	resubmits, err := resubmitCounts(webhookDB, eventIDs)
+	if err != nil {
+		h.serverError(
+			w, "failed to count event resubmissions", err,
+		)
+
+		return nil, 0, false
+	}
+
 	for i := range rows {
 		result[i].Deliveries = newDeliveryViews(
 			eventDeliveries[i], targetMap, attempts,
 		)
+		result[i].ResubmitCount = resubmits[rows[i].ID]
 	}
 
 	return result, totalEvents, true
 }
 
+// loadEventLogRows reads one page of the event log projection, newest
+// first, and the total number of events the pager counts against.
+func loadEventLogRows(
+	webhookDB *gorm.DB, webhookID string, page int,
+) ([]eventLogRow, int64) {
+	var totalEvents int64
+
+	webhookDB.Model(&database.Event{}).Where(
+		"webhook_id = ?", webhookID,
+	).Count(&totalEvents)
+
+	var rows []eventLogRow
+
+	webhookDB.Model(&database.Event{}).Select(
+		eventLogColumns, maxRenderedBodyBytes,
+	).Where(
+		"webhook_id = ?", webhookID,
+	).Order("created_at DESC").Offset(
+		(page - 1) * paginationPerPage,
+	).Limit(paginationPerPage).Find(&rows)
+
+	return rows, totalEvents
+}
+
+// resubmitCounts reports, for each of the page's events, how many
+// events have been resubmitted from it.
+//
+// One grouped query covers the page rather than one query per event.
+// A page holds paginationPerPage ids, far below SQLite's bound
+// parameter ceiling, so it needs no chunking as the delivery result
+// load does.
+func resubmitCounts(
+	webhookDB *gorm.DB, eventIDs []string,
+) (map[string]int, error) {
+	counts := make(map[string]int, len(eventIDs))
+
+	if len(eventIDs) == 0 {
+		return counts, nil
+	}
+
+	var rows []struct {
+		ResubmittedFromID string
+		Total             int
+	}
+
+	err := webhookDB.Model(&database.Event{}).
+		Select("resubmitted_from_id, count(*) AS total").
+		Where("resubmitted_from_id IN ?", eventIDs).
+		Group("resubmitted_from_id").
+		Find(&rows).Error
+	if err != nil {
+		return nil, err
+	}
+
+	for _, row := range rows {
+		counts[row.ResubmittedFromID] = row.Total
+	}
+
+	return counts, nil
+}
+
 // deliveryIDChunkSize bounds how many delivery IDs go into one
 // IN clause. SQLite refuses a statement carrying more than
 // SQLITE_MAX_VARIABLE_NUMBER (32766) bound parameters, and a

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
 }