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

internal/delivery/store.go

@@ -0,0 +1,164 @@
+package delivery
+
+import (
+	"fmt"
+
+	"gorm.io/gorm"
+	"sneak.berlin/go/webhooker/internal/database"
+)
+
+// StoreEvent stores event with one pending delivery for each of its
+// webhook's active targets, and adds them to the totals, in one
+// transaction, then queues the deliveries. It returns how many
+// deliveries it created.
+//
+// The receiver stores every event it accepts through here, and a
+// resubmit stores its copy of a stored event, so every event is
+// stored, retried, SSRF-guarded and circuit-broken the same way however
+// it entered the system. CreateReplay is the only other place a
+// delivery is created.
+//
+// The targets are the webhook's active ones as they stand now: a
+// target created after a resubmitted event first arrived is included,
+// and an inactive one is skipped. An event with no active target is
+// stored with no deliveries.
+func (e *Engine) StoreEvent(event *database.Event) (int, error) {
+	var targets []database.Target
+
+	err := e.database.DB().Where(
+		"webhook_id = ? AND active = ?", event.WebhookID, true,
+	).Find(&targets).Error
+	if err != nil {
+		return 0, fmt.Errorf("loading active targets: %w", err)
+	}
+
+	webhookDB, err := e.dbManager.GetDB(event.WebhookID)
+	if err != nil {
+		return 0, fmt.Errorf("getting webhook database: %w", err)
+	}
+
+	event.BodyBytes = int64(len(event.Body))
+	tasks := make([]Task, 0, len(targets))
+
+	err = webhookDB.Transaction(func(tx *gorm.DB) error {
+		err := tx.Create(event).Error
+		if err != nil {
+			return fmt.Errorf("creating event: %w", err)
+		}
+
+		for i := range targets {
+			dlv, err := createDelivery(tx, event.ID, targets[i].ID, false)
+			if err != nil {
+				return err
+			}
+
+			tasks = append(tasks, buildTask(
+				dlv, event.WebhookID, event, &targets[i], 1,
+			))
+		}
+
+		err = database.AddEventTotals(tx, database.EventTotals{
+			Events: 1, LastEventAt: &event.CreatedAt,
+		})
+		if err != nil {
+			return err
+		}
+
+		// A resubmitted copy did not arrive on its entrypoint's URL, so
+		// it leaves the entrypoint's last event as it is.
+		if event.ResubmittedFromID != nil {
+			return nil
+		}
+
+		return database.AddEntrypointTotals(tx, database.EntrypointTotals{
+			EntrypointID: event.EntrypointID, LastEventAt: event.CreatedAt,
+		})
+	})
+	if err != nil {
+		return 0, 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.
+	e.mtr.EventReceived()
+
+	e.notify(tasks)
+
+	return len(tasks), nil
+}
+
+// CreateReplay creates one more pending delivery of the stored event
+// eventID to target, marked as a replay, adds it to the target's totals
+// in the same transaction, and queues it. It returns the new
+// delivery's id.
+//
+// The caller passes the target as it stands now and has already
+// decided the replay may go ahead. The delivery being replayed is not
+// touched.
+func (e *Engine) CreateReplay(
+	webhookID, eventID string, target *database.Target,
+) (string, error) {
+	webhookDB, err := e.dbManager.GetDB(webhookID)
+	if err != nil {
+		return "", fmt.Errorf("getting webhook database: %w", err)
+	}
+
+	event, err := e.loadEvent(webhookDB, eventID)
+	if err != nil {
+		return "", err
+	}
+
+	var dlv *database.Delivery
+
+	err = webhookDB.Transaction(func(tx *gorm.DB) error {
+		var err error
+
+		dlv, err = createDelivery(tx, event.ID, target.ID, true)
+
+		return err
+	})
+	if err != nil {
+		return "", err
+	}
+
+	e.notify([]Task{buildTask(dlv, webhookID, &event, target, 1)})
+
+	return dlv.ID, nil
+}
+
+// createDelivery writes a pending delivery of the event to the target
+// on the caller's transaction, and adds it to the target's totals.
+//
+// Neither Event nor Target is set on the row, and the per-webhook
+// database omits associations on every write in any case (see
+// internal/database/event_db_isolation.go), so the target's settings
+// are never copied into the event database.
+func createDelivery(
+	tx *gorm.DB, eventID, targetID string, replay bool,
+) (*database.Delivery, error) {
+	dlv := &database.Delivery{
+		EventID:  eventID,
+		TargetID: targetID,
+		Status:   database.DeliveryStatusPending,
+		Replay:   replay,
+	}
+
+	err := tx.Create(dlv).Error
+	if err != nil {
+		return nil, fmt.Errorf(
+			"creating delivery for target %s: %w", targetID, err,
+		)
+	}
+
+	err = database.AddTargetTotals(tx, database.TargetTotals{
+		TargetID: targetID, Deliveries: 1,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return dlv, nil
+}