Store events and create deliveries in the delivery engine (closes #544)

The receiver, resubmit and replay handlers now parse the request and
make one call. Engine.StoreEvent stores an event with a pending
delivery for each active target and the totals in one transaction,
then queues them; Engine.CreateReplay does the same for one replay
delivery. The handlers take *delivery.Engine, so the Notifier
interface and its bridge are gone. Archives and CircuitBreakers stay
because the handler tests fake them. The tests that read a queued
task moved to the engine's tests; the handler tests use an engine
that is never started.

Model: opus-5-5

internal/delivery/engine.go

@@ -1,5 +1,5 @@
-// Package delivery manages asynchronous event delivery
-// to configured targets.
+// Package delivery stores each event with its deliveries and delivers
+// them, asynchronously, to configured targets.
 package delivery
 
 import (
@@ -23,10 +23,10 @@ import (
 
 const (
 	// deliveryChannelSize is the buffer size for the delivery
-	// channel. New Tasks from the webhook handler are sent
+	// channel. The Tasks of newly stored deliveries are sent
 	// here. Workers drain this channel. Sized large enough
-	// that the webhook handler should never block under
-	// normal load.
+	// that storing an event should never block under normal
+	// load.
 	deliveryChannelSize = 10000
 
 	// retryChannelSize is the buffer size for the retry
@@ -127,17 +127,11 @@ type Task struct {
 	AttemptNum int
 }
 
-// Notifier is the interface for notifying the delivery
-// engine about new deliveries.
-type Notifier interface {
-	Notify(tasks []Task)
-}
-
 // CircuitBreakers is how the handlers read a target's circuit
 // breaker, so the webhook page and the event log can say that
-// deliveries to the target are paused and until when. It keeps the
-// handlers free of the engine's internals and is trivially faked in
-// tests.
+// deliveries to the target are paused and until when. The handlers
+// hold the engine itself as well; they read breakers through this
+// interface so that a test can make one read as open.
 type CircuitBreakers interface {
 	StateAndCooldown(targetID string) (CircuitState, time.Duration)
 }
@@ -232,8 +226,8 @@ func New(
 	return e
 }
 
-// Notify signals the delivery engine that new deliveries
-// are ready.
+// Notify queues the tasks of deliveries already stored. StoreEvent and
+// CreateReplay call it once their transaction has committed.
 func (e *Engine) Notify(tasks []Task) {
 	for i := range tasks {
 		// Owned before it is queued, and until the worker that runs
@@ -472,9 +466,9 @@ func (e *Engine) processNewTask(
 	}
 
 	// The recovery routine's run at start can send and release this
-	// delivery before the receiver's Notify queues it. Ownership cannot
-	// refuse a delivery nobody holds, so the row decides whether it
-	// still needs sending.
+	// delivery before StoreEvent or CreateReplay queues it. Ownership
+	// cannot refuse a delivery nobody holds, so the row decides whether
+	// it still needs sending.
 	row, err := e.loadDelivery(webhookDB, task.DeliveryID)
 	if err != nil {
 		e.log.Error(
@@ -899,7 +893,7 @@ func (e *Engine) recoverSingleRetry(
 		return
 	}
 
-	e.ScheduleRetry(buildRecoveryTask(
+	e.ScheduleRetry(buildTask(
 		d, webhookID, &event, &target, attemptNum+1,
 	), remaining)
 
@@ -1822,7 +1816,11 @@ func (e *Engine) loadTarget(
 	return target, nil
 }
 
-func buildRecoveryTask(
+// buildTask builds the task that carries delivery d of event to target
+// through the engine's channels, as attempt attemptNum. The body is
+// carried only when it is under MaxInlineBodySize; a larger one is read
+// from the event row when the attempt runs (see hydrateEvent).
+func buildTask(
 	d *database.Delivery,
 	webhookID string,
 	event *database.Event,
@@ -1998,7 +1996,7 @@ func (e *Engine) sendRecoveredDeliveries(
 			continue
 		}
 
-		task := buildRecoveryTask(
+		task := buildTask(
 			&deliveries[i], webhookID, &event, &target,
 			attempts[deliveries[i].ID]+1,
 		)