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

The receiver, resubmit and replay handlers now read and check 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, and the engine's notify is
no longer exported. 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

README.md

@@ -2187,7 +2187,7 @@ A **Resubmit** stores its copy of an event through the same `Engine.StoreEvent`,
 and a **Replay** adds one delivery of a stored event through
 `Engine.CreateReplay`, which writes the delivery and its target's totals in one
 transaction and queues it. Those two methods are the only code that creates a
-delivery; the handlers parse the request and make the one call.
+delivery; the handlers read and check the request, then make the one call.
 
 ### Bounded Worker Pool
 
@@ -2198,8 +2198,9 @@ preventing goroutine explosions regardless of queue depth.
 **Architecture:**
 
 - **Channels as queues:** Two buffered channels serve as bounded queues: a
-  delivery channel (the tasks of deliveries the engine has just stored) and a
+  delivery channel (the tasks of newly stored deliveries and of the `pending`
-  retry channel (tasks from backoff timers). Both are buffered to 10,000.
+  deliveries the recovery routine hands back) and a retry channel (tasks from
+  backoff timers). Both are buffered to 10,000.
 - **Fan-out via channel, not goroutines:** When an event arrives with multiple
   targets, each `delivery.Task` is sent to the delivery channel. Workers pick
   them up and process them — no goroutine-per-target.
@@ -3161,7 +3162,7 @@ Components are wired via Uber fx in this order:
 11. `metrics.New` — The delivery collectors, registered on that registry
 12. `middleware.New` — HTTP middleware
 13. `delivery.New` — Event-driven delivery engine, which also stores each event
-    with its deliveries
+    with its deliveries and creates replay deliveries
 14. `archive.NewWriters` — The database targets' archive writers, and the lock
     every rename of their files runs under
 15. `archive.NewSweeper` — Periodic pruning of idle archives

internal/delivery/engine.go

@@ -23,10 +23,11 @@ import (
 
 const (
 	// deliveryChannelSize is the buffer size for the delivery
-	// channel. The Tasks of newly stored deliveries are sent
+	// channel. The Tasks of newly stored deliveries, and of the
-	// here. Workers drain this channel. Sized large enough
+	// pending deliveries the recovery routine hands back, are
-	// that storing an event should never block under normal
+	// sent here. Workers drain this channel. Sized large enough
-	// load.
+	// that no task should be dropped under normal load: one that
+	// does not fit is left for the sweep.
 	deliveryChannelSize = 10000
 
 	// retryChannelSize is the buffer size for the retry
@@ -151,12 +152,13 @@ type EngineParams struct {
 }
 
 // Engine processes queued deliveries in the background
-// using a bounded worker pool architecture. It owns only
+// using a bounded worker pool architecture. It also stores each
-// the cross-target machinery: the worker pool, the queue
+// event with its deliveries and creates replay deliveries (see
-// and retry channels, the recovery routine, and the persistence
+// store.go). Beyond that it owns only the cross-target machinery:
-// helpers and Scheduler that individual targets rely on.
+// the worker pool, the delivery and retry channels, the recovery
-// Each target type owns its own delivery, including retries,
+// routine, and the persistence helpers and Scheduler that
-// backoff, and circuit breaking.
+// individual targets rely on. Each target type owns its own
+// delivery, including retries, backoff, and circuit breaking.
 type Engine struct {
 	database   *database.Database
 	dbManager  *database.WebhookDBManager

internal/delivery/inflight_test.go

@@ -84,7 +84,7 @@ func TestSweepLeavesAQueuedDeliveryAlone(t *testing.T) {
 	)
 	rAgePending(t, s.WebhookDB, d.ID)
 
-	// Queued exactly as the receiver queues it, and never dequeued:
+	// Queued exactly as StoreEvent queues it, and never dequeued:
 	// no workers are running in this engine.
 	s.Engine.ExportNotify([]delivery.Task{{
 		DeliveryID: d.ID,
@@ -248,9 +248,9 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
 }
 
 // TestNotifyAfterRecoveryDoesNotSendAgain is the startup race of
-// https://git.eeqj.de/sneak/webhooker/issues/299. The receiver has
+// https://git.eeqj.de/sneak/webhooker/issues/299. StoreEvent has
 // written a delivery, the recovery routine's run at start finds it
-// pending, sends it and releases it, and only then does the receiver's
+// pending, sends it and releases it, and only then does StoreEvent's
 // notify for it arrive. Nothing owns the delivery by then, so notify
 // takes it.
 func TestNotifyAfterRecoveryDoesNotSendAgain(t *testing.T) {

internal/handlers/event_resubmit.go

@@ -172,8 +172,10 @@ func loadResubmitSource(
 }
 
 // queueResubmit has the delivery engine store the copy, with a delivery
-// for each of the webhook's active targets as they stand now, exactly
+// for each of the webhook's active targets as they stand now, through
-// as it stores a received event.
+// the same StoreEvent the receiver uses. The one difference is that the
+// copy leaves its entrypoint's last event as it is, since it did not
+// arrive on that URL.
 func (h *Handlers) queueResubmit(
 	w http.ResponseWriter,
 	r *http.Request,

internal/handlers/handlers_test.go

@@ -239,8 +239,8 @@ func newTestAppWithConfig(
 			session.New,
 			archive.NewWriters,
 			// The engine is built on a lifecycle that is never started,
-			// so no worker runs: every delivery a handler creates stays
+			// so no worker runs: every delivery the engine creates for a
-			// pending, where the test can read it.
+			// handler stays pending, where the test can read it.
 			func(p delivery.EngineParams) *delivery.Engine {
 				return delivery.New(fxtest.NewLifecycle(t), p)
 			},