Add a statistics pane to the webhook page (closes #368)

The webhook page opens with a statistics pane: entrypoints and targets, deliveries in progress, the last arrival, the retention period; events, deliveries and failures, lifetime and within retention; and events, failures and failure percentage over 10 minutes and 24 hours.

Running totals, one row per target plus one for events, sit in each webhook's event database and are written in the same transaction as the rows they count; retention prunes in paused, stoppable batches and subtracts what it removes. Window figures are index-range counts on a new finished_at column. An existing event database must be recreated.

Model: opus-5-5

internal/delivery/delivery_totals_test.go

@@ -0,0 +1,177 @@
+package delivery_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+	"sneak.berlin/go/webhooker/internal/database"
+)
+
+// targetTotals reads one target's totals from a webhook database, all
+// zero when it has no row.
+func targetTotals(
+	t *testing.T, db *gorm.DB, targetID string,
+) database.TargetTotals {
+	t.Helper()
+
+	var rows []database.TargetTotals
+
+	require.NoError(t, db.Where("target_id = ?", targetID).
+		Find(&rows).Error)
+
+	if len(rows) == 0 {
+		return database.TargetTotals{TargetID: targetID}
+	}
+
+	return rows[0]
+}
+
+// TestUpdateDeliveryStatus_FinishTimeAndTargetTotals pins what a status
+// write records for the webhook page's statistics: the time a delivery
+// finished, set only when it becomes delivered or failed, and one more
+// on its target's delivered or failed total.
+func TestUpdateDeliveryStatus_FinishTimeAndTargetTotals(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		status    database.DeliveryStatus
+		finished  bool
+		delivered int64
+		failed    int64
+	}{
+		{database.DeliveryStatusRetrying, false, 0, 0},
+		{database.DeliveryStatusDelivered, true, 1, 0},
+		{database.DeliveryStatusFailed, true, 0, 1},
+	}
+
+	for _, tt := range tests {
+		t.Run(string(tt.status), func(t *testing.T) {
+			t.Parallel()
+
+			db := testWebhookDB(t)
+			e := testEngine(t, 1)
+			event := seedEvent(t, db, `{}`)
+			targetID := uuid.New().String()
+			d := seedDelivery(
+				t, db, event.ID, targetID,
+				database.DeliveryStatusPending,
+			)
+
+			before := time.Now()
+
+			require.NoError(t, e.ExportUpdateDeliveryStatus(
+				db, &d, tt.status,
+			))
+
+			var stored database.Delivery
+
+			require.NoError(t, db.First(&stored, "id = ?", d.ID).Error)
+			assert.Equal(t, tt.status, stored.Status)
+
+			if tt.finished {
+				require.NotNil(t, stored.FinishedAt)
+				assert.False(t, stored.FinishedAt.Before(before))
+			} else {
+				assert.Nil(t, stored.FinishedAt)
+			}
+
+			assert.Equal(t, database.TargetTotals{
+				TargetID:  targetID,
+				Delivered: tt.delivered,
+				Failed:    tt.failed,
+			}, targetTotals(t, db, targetID))
+		})
+	}
+}
+
+// TestUpdateDeliveryStatus_DeletedDeliveryIsNotCounted covers a
+// delivery retention deleted while the engine still held it. Failing
+// it afterwards writes no row, so it adds no failure either: retention
+// has already counted what it removed.
+func TestUpdateDeliveryStatus_DeletedDeliveryIsNotCounted(t *testing.T) {
+	t.Parallel()
+
+	db := testWebhookDB(t)
+	e := testEngine(t, 1)
+	event := seedEvent(t, db, `{}`)
+	targetID := uuid.New().String()
+	d := seedDelivery(
+		t, db, event.ID, targetID,
+		database.DeliveryStatusRetrying,
+	)
+
+	require.NoError(t, db.Unscoped().
+		Delete(&database.Delivery{}, "id = ?", d.ID).Error)
+
+	require.NoError(t, e.ExportUpdateDeliveryStatus(
+		db, &d, database.DeliveryStatusFailed,
+	))
+
+	assert.Equal(t, database.TargetTotals{TargetID: targetID},
+		targetTotals(t, db, targetID))
+}
+
+// TestUpdateDeliveryStatus_FinishedDeliveryIsNotSettledAgain covers a
+// delivery settled a second time, as recovery can do when a worker has
+// settled it since recovery read it. Neither status writes over the
+// first, and the totals do not move.
+func TestUpdateDeliveryStatus_FinishedDeliveryIsNotSettledAgain(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	finished := []database.DeliveryStatus{
+		database.DeliveryStatusDelivered,
+		database.DeliveryStatusFailed,
+	}
+
+	for _, first := range finished {
+		t.Run(string(first), func(t *testing.T) {
+			t.Parallel()
+
+			db := testWebhookDB(t)
+			e := testEngine(t, 1)
+			event := seedEvent(t, db, `{}`)
+			targetID := uuid.New().String()
+			d := seedDelivery(
+				t, db, event.ID, targetID,
+				database.DeliveryStatusRetrying,
+			)
+
+			// The delivery as recovery read it, before the worker
+			// settled it.
+			readBefore := d
+
+			require.NoError(t, e.ExportUpdateDeliveryStatus(
+				db, &d, first,
+			))
+
+			var settled database.Delivery
+
+			require.NoError(t, db.First(&settled, "id = ?", d.ID).Error)
+			require.NotNil(t, settled.FinishedAt)
+
+			totals := targetTotals(t, db, targetID)
+
+			for _, again := range finished {
+				stale := readBefore
+
+				require.NoError(t, e.ExportUpdateDeliveryStatus(
+					db, &stale, again,
+				))
+			}
+
+			var stored database.Delivery
+
+			require.NoError(t, db.First(&stored, "id = ?", d.ID).Error)
+			assert.Equal(t, first, stored.Status)
+			require.NotNil(t, stored.FinishedAt)
+			assert.True(t, settled.FinishedAt.Equal(*stored.FinishedAt))
+			assert.Equal(t, totals, targetTotals(t, db, targetID))
+		})
+	}
+}

internal/delivery/engine.go

@@ -531,6 +531,11 @@ func (e *Engine) processRetryTask(
 		return
 	}
 
+	// Set before anything below can fail the delivery: the failure is
+	// added to this target's totals.
+	d.EventID = task.EventID
+	d.TargetID = task.TargetID
+
 	if d.Status != database.DeliveryStatusRetrying {
 		e.log.Debug(
 			"skipping retry for delivery "+
@@ -562,8 +567,6 @@ func (e *Engine) processRetryTask(
 	}
 
 	target := buildTargetFromTask(task)
-	d.EventID = task.EventID
-	d.TargetID = task.TargetID
 	d.Event = event
 	d.Target = target
 
@@ -1554,8 +1557,9 @@ func (e *Engine) updateDeliveryStatus(
 	targetType database.TargetType,
 	status database.DeliveryStatus,
 ) error {
-	err := webhookDB.Model(d).
-		Update("status", status).Error
+	err := webhookDB.Transaction(func(tx *gorm.DB) error {
+		return writeDeliveryStatus(tx, d, status)
+	})
 	if err != nil {
 		return fmt.Errorf(
 			"updating delivery %s to status %s: %w",
@@ -1574,6 +1578,43 @@ func (e *Engine) updateDeliveryStatus(
 	return nil
 }
 
+// writeDeliveryStatus writes a delivery's new status. A delivery that
+// becomes delivered or failed also gets the time it finished, and is
+// added to its target's delivered or failed total. That write changes
+// only a delivery not yet delivered or failed, and the total moves
+// only when it changed a row: retention may have deleted the delivery
+// while the engine was working on it, and a recovery path may settle
+// a delivery that a worker has already settled.
+func writeDeliveryStatus(
+	tx *gorm.DB,
+	d *database.Delivery,
+	status database.DeliveryStatus,
+) error {
+	if !status.Terminal() {
+		return tx.Model(d).Update("status", status).Error
+	}
+
+	res := tx.Model(d).
+		Where("status NOT IN ?", []database.DeliveryStatus{
+			database.DeliveryStatusDelivered,
+			database.DeliveryStatusFailed,
+		}).
+		Updates(map[string]any{
+			"status":      status,
+			"finished_at": time.Now(),
+		})
+	if res.Error != nil || res.RowsAffected == 0 {
+		return res.Error
+	}
+
+	add := database.TargetTotals{TargetID: d.TargetID, Delivered: 1}
+	if status == database.DeliveryStatusFailed {
+		add = database.TargetTotals{TargetID: d.TargetID, Failed: 1}
+	}
+
+	return database.AddTargetTotals(tx, add)
+}
+
 // settleStatus moves a delivery to its outcome status and reports a
 // failed write through bookkeepingFailed, which leaves the row
 // recoverable. It exists so the target call sites read as one

internal/delivery/engine_test.go

@@ -57,7 +57,10 @@ func testWebhookDB(t *testing.T) *gorm.DB {
 		&database.Event{},
 		&database.Delivery{},
 		&database.DeliveryResult{},
+		&database.EventTotals{},
+		&database.TargetTotals{},
 	))
+	require.NoError(t, db.Create(&database.EventTotals{}).Error)
 
 	return db
 }

internal/delivery/export_test.go

@@ -150,6 +150,16 @@ func (e *Engine) ExportDeliverSlack(
 	)
 }
 
+// ExportUpdateDeliveryStatus exposes updateDeliveryStatus. It passes no
+// target type, so no metric moves.
+func (e *Engine) ExportUpdateDeliveryStatus(
+	webhookDB *gorm.DB,
+	d *database.Delivery,
+	status database.DeliveryStatus,
+) error {
+	return e.updateDeliveryStatus(webhookDB, d, "", status)
+}
+
 // ExportProcessNewTask exposes processNewTask.
 func (e *Engine) ExportProcessNewTask(
 	ctx context.Context, task *Task,

internal/delivery/terminal_state_test.go

@@ -418,6 +418,38 @@ func TestProcessRetryTask_TargetDeleted_MakesNoAttempt(
 	assert.Zero(t, s.Engine.ExportInflightHeld())
 }
 
+// TestProcessRetryTask_TargetDeleted_CountsFailureOnTarget verifies
+// that the failure of a retry abandoned because its target is gone is
+// added to that target's own totals, not to a row with no target.
+func TestProcessRetryTask_TargetDeleted_CountsFailureOnTarget(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	s := newISetup(t)
+
+	var hits atomic.Int64
+
+	task, targetID := tRetryChainSetup(
+		t, s, "gone-counted", &hits,
+	)
+
+	require.NoError(t, s.MainDB.Delete(
+		&database.Target{}, "id = ?", targetID,
+	).Error)
+
+	s.Engine.ExportProcessRetryTask(
+		context.Background(), &task,
+	)
+
+	var rows []database.TargetTotals
+
+	require.NoError(t, s.WebhookDB.Find(&rows).Error)
+	assert.Equal(t, []database.TargetTotals{
+		{TargetID: targetID, Failed: 1},
+	}, rows)
+}
+
 // TestProcessRetryTask_TargetPresent_StillDelivers is the guard's
 // mutation check: a liveness check that refused every retry would pass
 // the test above and break every retry there is.