From 89f3b984d211f7237ef9db0289c9b28cfee8ae6e Mon Sep 17 00:00:00 2001 From: clawbot Date: Mon, 24 Aug 2026 00:53:37 +0200 Subject: [PATCH] Resubmit a stored event as a new undelivered event (closes #250) --- README.md | 58 ++- internal/database/model_event.go | 8 + internal/handlers/event_log_view.go | 50 +- internal/handlers/event_resubmit.go | 277 +++++++++++ internal/handlers/event_resubmit_test.go | 603 +++++++++++++++++++++++ internal/handlers/source_management.go | 130 +++-- internal/handlers/webhook.go | 213 ++++---- internal/metrics/metrics.go | 27 + internal/middleware/ratelimit.go | 28 ++ internal/server/routes.go | 12 + templates/source_logs.html | 21 + 11 files changed, 1279 insertions(+), 148 deletions(-) create mode 100644 internal/handlers/event_resubmit.go create mode 100644 internal/handlers/event_resubmit_test.go diff --git a/README.md b/README.md index 433db92..486233a 100644 --- a/README.md +++ b/README.md @@ -903,10 +903,13 @@ webhooker solves this by acting as a durable intermediary: targets simultaneously. This enables patterns like forwarding a GitHub webhook to both a deployment service and a Slack channel. -5. **Replay** (not yet implemented) — Every received event is stored in - full, which is what manual redelivery for debugging or testing will - be built on. No redelivery exists today, in the web UI or the API; - see [TODO.md](TODO.md). +5. **Replay and resubmit** — Every received event is stored in full, + and the event log offers two redelivery actions built on that. + **Replay** re-sends one finished delivery to its own target, for + recovering a delivery that failed. **Resubmit** re-injects the stored + event as a new undelivered event and fans it out to every currently + active target, for firing captured traffic at a backend under + development. Both are web UI actions; there is no API for either. ### Use Cases @@ -915,8 +918,8 @@ webhooker solves this by acting as a durable intermediary: - **Observability** via Prometheus metrics on webhook frequency, payload size, and delivery performance - **Debugging** and introspection of webhook payloads in the web UI -- **Replay** of webhook events for application testing and development - (planned; not yet implemented) +- **Resubmit** of captured webhook events for application testing and + development, and **replay** of a single failed delivery for recovery - **Fan-out** delivery of a single webhook to multiple downstream targets - **High-availability ingestion** for delivery to less reliable backend @@ -1224,7 +1227,7 @@ A programmatic access credential for API authentication. #### Event A captured incoming webhook request. Stores the complete HTTP request -data for auditing and for the planned replay capability. +data for auditing, for replay, and for resubmission. | Field | Type | Description | | -------------- | ------ | ----------- | @@ -1235,6 +1238,7 @@ data for auditing and for the planned replay capability. | `headers` | JSON | Complete request headers | | `body` | text | Raw request body | | `content_type` | string | Content-Type header value | +| `resubmitted_from_id` | UUID | The event this one was copied from by a resubmit (nullable; empty for an event that arrived on the receiver). Not a foreign key: the source event can be reaped by retention while its copies remain | **Relations:** Belongs to Webhook. Belongs to Entrypoint. Has many Deliveries. @@ -1283,6 +1287,30 @@ message on the event log rather than delivering from stale configuration, and a replay is refused while an earlier one for the same event and target is still pending or retrying. +**Resubmit.** Replay recovers one delivery; **resubmit** re-injects one +EVENT. The event log offers a per-event **Resubmit** action that stores +a NEW event copying the stored one's `method`, `headers`, `body` and +`content_type` verbatim, then fans it out to the webhook's currently +**active** targets — resolved fresh by the same query the receiver +uses, so a target created long after the original event arrived +receives it. That is the difference that matters: a target added to +test a backend under development has no prior delivery, so there is +nothing to replay to it, while a resubmit reaches it like any other +active target. Inactive targets are skipped, exactly as the receiver +skips them. + +The new event is a first-class event in the log with its own +deliveries, not a marker on the one it came from, and the original's +deliveries are left untouched. It records `resubmitted_from_id`, and +the event log shows the relationship both ways, so a captured event +fired twenty times at a backend stays traceable. Resubmitting the same +event repeatedly is supported and is the point of the action — there is +no in-flight refusal; the route's own rate limit is what bounds it. + +Inbound signature verification is not re-run on a resubmit. There is no +inbound signature to check on a copy the operator submits, and the +route is authenticated and CSRF-protected as an operator action. + #### DeliveryResult The result of a single delivery attempt. Every attempt (including @@ -1539,10 +1567,11 @@ unknown) one while one of its deliveries is still `retrying`, both recovery paths above terminally mark that delivery `failed` and record a `DeliveryResult` naming the current target type as the reason, logging it at warn level. The delivery is not re-dispatched under the new type — the -operator never asked for that delivery — and while the event itself -remains stored in the per-webhook event database, there is no way to -redeliver it: manual redelivery is planned, not implemented (see -[TODO.md](TODO.md)). +operator never asked for that delivery — and the event itself remains +stored in the per-webhook event database. Nothing is lost: the delivery +is `failed`, which is terminal, so the event log offers **Replay** on +it to re-send it to that target under the new type, and **Resubmit** on +the event to re-inject it to every currently active target. ### Circuit Breaker (HTTP and Slack Targets with Retries) @@ -1615,13 +1644,16 @@ arriving and being stored, they are just not getting anywhere. | `webhooker_deliveries_failed_total` | counter | Deliveries that failed terminally and will not be retried | | `webhooker_delivery_retries_total` | counter | Deliveries put back into `retrying` | | `webhooker_delivery_replays_total` | counter | Deliveries an operator replayed from the event log. A replay runs the ordinary engine path, so it also moves the attempt, outcome and duration series; this is the only one that separates it from ordinary traffic | +| `webhooker_events_resubmitted_total` | counter | Stored events an operator re-injected from the event log. The new event also moves `webhooker_events_received_total`, since it is a stored event the delivery side is compared against; this counter is what separates the two. Unlabelled: the target types it fans out to belong to the delivery series | | `webhooker_delivery_duration_seconds` | histogram | Wall time of a single dispatched delivery attempt, the same duration the attempt's `DeliveryResult` records | | `webhooker_deliveries_pending` | gauge | Deliveries currently in `pending` | | `webhooker_deliveries_retrying` | gauge | Deliveries currently in `retrying` | | `webhooker_circuit_breakers_open` | gauge | Circuit breakers currently open | Every delivery metric carries exactly one label, `target_type`, and -cardinality is the whole reason for that restriction. A target type is +cardinality is the whole reason for that restriction. The two +event-level counters carry no label at all — an event is not the +property of any one target type. A target type is one of four compile-time constants, so the label domain is bounded by construction; a value outside that set collapses to `unknown` rather than minting a series of its own. Target ids, event ids and entrypoint @@ -2213,6 +2245,7 @@ abuse limit later; they are tracked as future work. | `POST` | `/source/{id}/delete` | Delete webhook | | `GET` | `/source/{id}/logs` | Webhook event logs | | `POST` | `/source/{id}/deliveries/{deliveryID}/replay` | Replay a finished delivery: creates a new delivery for the same event against the target's current configuration (30 per minute per bucket, then `429`) | +| `POST` | `/source/{id}/events/{eventID}/resubmit` | Resubmit a stored event: creates a new event copying it and fans that out to every currently active target (30 per minute per bucket, then `429`) | | `POST` | `/source/{id}/entrypoints` | Add entrypoint to webhook | | `POST` | `/source/{id}/entrypoints/{entrypointID}/delete` | Delete an entrypoint | | `POST` | `/source/{id}/entrypoints/{entrypointID}/toggle` | Enable or disable an entrypoint | @@ -2303,6 +2336,7 @@ webhooker/ │ │ ├── handlers.go # Base handler struct, JSON helpers, template rendering │ │ ├── auth.go # Login, logout handlers │ │ ├── delivery_replay.go # Per-delivery replay: new delivery, current target config +│ │ ├── event_resubmit.go # Event resubmit: new event, all currently active targets │ │ ├── entrypoint_view.go # Masked entrypoint view for templates │ │ ├── event_log_view.go # Event log projection, byte-capped in SQL │ │ ├── healthcheck.go # Health check handler diff --git a/internal/database/model_event.go b/internal/database/model_event.go index bd332d6..dafc235 100644 --- a/internal/database/model_event.go +++ b/internal/database/model_event.go @@ -13,6 +13,14 @@ type Event struct { Body string `gorm:"type:text" json:"body"` ContentType string `json:"contentType"` + // ResubmittedFromID names the event this one was copied from by + // an operator resubmit. It is nil for an event that arrived on + // the receiver, which is every event created before the column + // existed. It is not a foreign key: the source event can be + // reaped by retention while its copies remain, and the id is + // kept as the record of where the copy came from either way. + ResubmittedFromID *string `gorm:"type:uuid;index" json:"resubmittedFromId,omitempty"` + // Relations Webhook Webhook `json:"webhook,omitzero"` Entrypoint Entrypoint `json:"entrypoint,omitzero"` diff --git a/internal/handlers/event_log_view.go b/internal/handlers/event_log_view.go index 8d5837b..5cfc0d1 100644 --- a/internal/handlers/event_log_view.go +++ b/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, } } diff --git a/internal/handlers/event_resubmit.go b/internal/handlers/event_resubmit.go new file mode 100644 index 0000000..d1b677b --- /dev/null +++ b/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) +} diff --git a/internal/handlers/event_resubmit_test.go b/internal/handlers/event_resubmit_test.go new file mode 100644 index 0000000..b2ea384 --- /dev/null +++ b/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, ¬if) + 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, ¬if) + 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, ¬if) + 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, ¬if) + 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, ¬if) + 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, ¬if) + 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", + ) +} diff --git a/internal/handlers/source_management.go b/internal/handlers/source_management.go index 8219e56..a48a2bc 100644 --- a/internal/handlers/source_management.go +++ b/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 diff --git a/internal/handlers/webhook.go b/internal/handlers/webhook.go index a1ebc40..44e4940 100644 --- a/internal/handlers/webhook.go +++ b/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 } diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 5fc5354..1b3d68e 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -83,6 +83,7 @@ type Set struct { deliveriesFailed *prometheus.CounterVec deliveryRetries *prometheus.CounterVec deliveryReplays *prometheus.CounterVec + eventsResubmitted prometheus.Counter deliveryDuration *prometheus.HistogramVec deliveriesPending *prometheus.GaugeVec deliveriesRetrying *prometheus.GaugeVec @@ -166,6 +167,23 @@ func (s *Set) DeliveryReplayed(t database.TargetType) { Inc() } +// EventResubmitted counts one stored event an operator re-injected +// from the event log. +// +// It counts the operator action once, not the deliveries it fans out +// to: those already move the attempt, outcome and duration series, and +// the new event moves events_received_total, since it is a stored +// event that the delivery side will be compared against. This counter +// is what separates a resubmitted event from a received one. +// +// It carries no labels. The only label available at the call site +// would be the route pattern, which has exactly one value and so would +// distinguish nothing; the target types the event fans out to belong +// to the delivery series, not to this one. +func (s *Set) EventResubmitted() { + s.eventsResubmitted.Inc() +} + // DeliveryStatusChanged counts a delivery's transition into a new // status. The mapping from status to counter lives here, next to the // collectors, so the engine has a single call for every transition it @@ -298,6 +316,15 @@ func (s *Set) registerCounters(factory promauto.Factory) { }, []string{targetTypeLabel}, ) + + s.eventsResubmitted = factory.NewCounter( + prometheus.CounterOpts{ + Namespace: namespace, + Name: "events_resubmitted_total", + Help: "Stored events an operator re-injected from " + + "the event log as new events.", + }, + ) } func (s *Set) registerGauges(factory promauto.Factory) { diff --git a/internal/middleware/ratelimit.go b/internal/middleware/ratelimit.go index 9b13396..dc602ad 100644 --- a/internal/middleware/ratelimit.go +++ b/internal/middleware/ratelimit.go @@ -44,6 +44,18 @@ const ( // replayRateInterval is the time window for the replay limit. replayRateInterval = 1 * time.Minute + // resubmitRateLimit is the maximum number of event resubmits one + // client may queue per interval. A resubmit stores an event and + // queues one delivery per active target, so it costs more + // outbound work per press than a replay does. Firing a captured + // event repeatedly at a backend under development is the point of + // the action, so the ceiling stays well above the rate a person + // iterates at. + resubmitRateLimit = 30 + + // resubmitRateInterval is the time window for the resubmit limit. + resubmitRateInterval = 1 * time.Minute + // receiverRateInterval is the time window for the webhook // receiver rate limit. The configured limit is expressed in // requests per minute. @@ -315,6 +327,22 @@ func (m *Middleware) ReplayRateLimit() func(http.Handler) http.Handler { ) } +// ResubmitRateLimit returns middleware that enforces per-IP rate +// limiting on event resubmits. +// +// It is a separate bucket from the replay limit so that exhausting one +// does not take the other away: replay is a recovery action and +// resubmit is a testing action, and an operator iterating on a backend +// must not lose the ability to re-send a failed delivery. +func (m *Middleware) ResubmitRateLimit() func(http.Handler) http.Handler { + return m.postRateLimit( + resubmitRateLimit, + resubmitRateInterval, + "event resubmit rate limit exceeded", + "Too many resubmits. Please try again later.", + ) +} + // postRateLimit builds middleware that enforces a per-IP rate // limit on POST requests only; all other methods pass through // unaffected. Requests over the limit receive a 429 with the diff --git a/internal/server/routes.go b/internal/server/routes.go index 846d3a3..0770fd2 100644 --- a/internal/server/routes.go +++ b/internal/server/routes.go @@ -213,6 +213,18 @@ func (s *Server) setupSourceRoutes() { "/deliveries/{deliveryID}/replay", s.h.HandleDeliveryReplay(), ) + // Resubmit is the other page action that queues outbound + // work: it copies a stored event into a new one and fans + // that out to every currently active target. It is + // deliberately repeatable, so the rate limit is the only + // bound on a held-down button; it gets its own bucket so + // that spending it does not also disable replay. POST + // only, so the action cannot be taken by a link, a + // prefetch or an image tag. + r.With(s.mw.ResubmitRateLimit()).Post( + "/events/{eventID}/resubmit", + s.h.HandleEventResubmit(), + ) r.Post( "/entrypoints", s.h.HandleEntrypointCreate(), diff --git a/templates/source_logs.html b/templates/source_logs.html index e0ea7f7..0dc1981 100644 --- a/templates/source_logs.html +++ b/templates/source_logs.html @@ -16,6 +16,10 @@
{{.ReplayMessage}}
{{end}} + {{if .ResubmitMessage}} +
{{.ResubmitMessage}}
+ {{end}} +
{{range .Events}} @@ -25,6 +29,12 @@ {{.Method}} {{.ID}} {{.ContentType}} + {{if .ResubmittedFrom}} + resubmitted copy + {{end}} + {{if .ResubmitCount}} + resubmitted {{.ResubmitCount}} time{{if ne .ResubmitCount 1}}s{{end}} + {{end}}
{{range .Deliveries}} @@ -40,6 +50,17 @@
+
+
+ {{if .ResubmittedFrom}}Resubmitted from event {{.ResubmittedFromID}}.{{end}} + {{if .ResubmitCount}}Resubmitted as {{.ResubmitCount}} new event{{if ne .ResubmitCount 1}}s{{end}}.{{end}} +
+
+ + + +
+ 
{{.Body}}
{{if .BodyTruncated}}

Body truncated for display: showing {{.BodyShownBytes}} of {{.BodyBytes}} bytes. The stored body is unchanged — download the full body.