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

Each webhook's event database keeps running totals: one row for its
events, and one row per target for that target's deliveries, delivered
and failed, each with what retention removed. Every write to them
shares the transaction of the rows it counts. Deliveries get a
finished_at column; it and target_id end the status index, so each
target's deliveries finished in a window come from one index-range
query grouped by target. Retention deletes 1000 expired events per
transaction. The pane is its own template, its figures in tables.

The schema changes in place with nothing back-filled, so an existing
database must be recreated.

Model: opus-5-5

internal/delivery/delivery_totals_test.go

@@ -0,0 +1,116 @@
+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))
+}

internal/delivery/engine.go

@@ -1554,8 +1554,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 +1575,36 @@ 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. It is counted only
+// if the row was still there to update: retention may have deleted it
+// while the engine was working on it.
+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).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,