Add per-webhook event retention reaper (closes #63)

cmd/webhooker/main.go

@@ -34,6 +34,7 @@ func main() {
 			config.New,
 			database.New,
 			database.NewWebhookDBManager,
+			database.NewRetentionReaper,
 			healthcheck.New,
 			session.New,
 			handlers.New,
@@ -44,6 +45,13 @@ func main() {
 			func(e *delivery.Engine) delivery.Notifier { return e },
 			server.New,
 		),
-		fx.Invoke(func(*server.Server, *delivery.Engine) {}),
+		fx.Invoke(
+			func(
+				*server.Server,
+				*delivery.Engine,
+				*database.RetentionReaper,
+			) {
+			},
+		),
 	).Run()
 }

internal/config/config.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	"go.uber.org/fx"
 	"sneak.berlin/go/webhooker/internal/globals"
@@ -26,6 +27,10 @@ const (
 
 	// defaultPort is the default HTTP listen port.
 	defaultPort = 8080
+
+	// defaultRetentionSweepInterval is how often the retention
+	// reaper deletes events older than each webhook's RetentionDays.
+	defaultRetentionSweepInterval = time.Hour
 )
 
 // ErrInvalidEnvironment is returned when WEBHOOKER_ENVIRONMENT
@@ -51,6 +56,10 @@ type Config struct {
 	MetricsUsername string
 	Port            int
 	SentryDSN       string
+
+	// RetentionSweepInterval is how often the retention reaper runs.
+	RetentionSweepInterval time.Duration
+
 	params *ConfigParams
 	log    *slog.Logger
 }
@@ -95,6 +104,23 @@ func envInt(key string, defaultValue int) int {
 	return defaultValue
 }
 
+// envDuration returns the value of the named environment variable
+// parsed as a Go duration (e.g. "1h", "30m"). Returns defaultValue if
+// not set or unparseable.
+func envDuration(
+	key string,
+	defaultValue time.Duration,
+) time.Duration {
+	if v := os.Getenv(key); v != "" {
+		d, err := time.ParseDuration(v)
+		if err == nil {
+			return d
+		}
+	}
+
+	return defaultValue
+}
+
 // New creates a Config by reading environment variables.
 //
 //nolint:revive // lc parameter is required by fx even if unused.
@@ -128,6 +154,10 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
 		MetricsPassword: envString("METRICS_PASSWORD"),
 		Port:            envInt("PORT", defaultPort),
 		SentryDSN:       envString("SENTRY_DSN"),
+		RetentionSweepInterval: envDuration(
+			"RETENTION_SWEEP_INTERVAL",
+			defaultRetentionSweepInterval,
+		),
 		log:    log,
 		params: &params,
 	}
@@ -151,6 +181,7 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) {
 		"debug", s.Debug,
 		"maintenanceMode", s.MaintenanceMode,
 		"dataDir", s.DataDir,
+		"retentionSweepInterval", s.RetentionSweepInterval.String(),
 		"hasSentryDSN", s.SentryDSN != "",
 		"hasMetricsAuth",
 		s.MetricsUsername != "" && s.MetricsPassword != "",

internal/database/export_test.go

@@ -0,0 +1,31 @@
+package database
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"time"
+)
+
+// NewTestRetentionReaper builds a RetentionReaper backed by the given
+// main database and per-webhook database manager, without the fx
+// lifecycle. Intended for tests.
+func NewTestRetentionReaper(
+	db *Database,
+	mgr *WebhookDBManager,
+) *RetentionReaper {
+	return &RetentionReaper{
+		db:        db,
+		dbManager: mgr,
+		log: slog.New(slog.NewTextHandler(
+			os.Stderr,
+			&slog.HandlerOptions{Level: slog.LevelDebug},
+		)),
+		interval: time.Hour,
+	}
+}
+
+// ExportSweep runs a single retention sweep synchronously for tests.
+func (r *RetentionReaper) ExportSweep(ctx context.Context) {
+	r.sweep(ctx)
+}

internal/database/retention.go

@@ -0,0 +1,252 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	"go.uber.org/fx"
+	"gorm.io/gorm"
+	"sneak.berlin/go/webhooker/internal/config"
+	"sneak.berlin/go/webhooker/internal/logger"
+)
+
+// hoursPerDay converts a RetentionDays count into hours for cutoff
+// computation.
+const hoursPerDay = 24
+
+// RetentionReaperParams holds the fx dependencies for the
+// RetentionReaper.
+type RetentionReaperParams struct {
+	fx.In
+
+	Config    *config.Config
+	Database  *Database
+	DBManager *WebhookDBManager
+	Logger    *logger.Logger
+}
+
+// RetentionReaper periodically deletes expired events (and their
+// dependent deliveries and delivery results) from each per-webhook
+// database, enforcing every webhook's RetentionDays. Rows are removed
+// permanently so that per-webhook SQLite files do not grow without
+// bound.
+type RetentionReaper struct {
+	db        *Database
+	dbManager *WebhookDBManager
+	log       *slog.Logger
+	interval  time.Duration
+	cancel    context.CancelFunc
+	wg        sync.WaitGroup
+}
+
+// NewRetentionReaper creates the retention reaper and registers its
+// fx lifecycle hooks. The background sweep loop starts on OnStart and
+// stops cleanly on OnStop via context cancellation.
+func NewRetentionReaper(
+	lc fx.Lifecycle,
+	params RetentionReaperParams,
+) *RetentionReaper {
+	r := &RetentionReaper{
+		db:        params.Database,
+		dbManager: params.DBManager,
+		log:       params.Logger.Get(),
+		interval:  params.Config.RetentionSweepInterval,
+	}
+
+	lc.Append(fx.Hook{
+		OnStart: func(ctx context.Context) error {
+			r.start(ctx)
+
+			return nil
+		},
+		OnStop: func(_ context.Context) error {
+			r.stop()
+
+			return nil
+		},
+	})
+
+	return r
+}
+
+func (r *RetentionReaper) start(ctx context.Context) {
+	ctx, cancel := context.WithCancel(ctx)
+	r.cancel = cancel
+
+	r.wg.Add(1)
+
+	go r.run(ctx)
+
+	r.log.Info(
+		"retention reaper started",
+		"interval", r.interval.String(),
+	)
+}
+
+func (r *RetentionReaper) stop() {
+	r.log.Info("retention reaper stopping")
+
+	if r.cancel != nil {
+		r.cancel()
+	}
+
+	r.wg.Wait()
+	r.log.Info("retention reaper stopped")
+}
+
+func (r *RetentionReaper) run(ctx context.Context) {
+	defer r.wg.Done()
+
+	ticker := time.NewTicker(r.interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			r.sweep(ctx)
+		}
+	}
+}
+
+// sweep lists every webhook from the main database and reaps expired
+// rows from each per-webhook database whose RetentionDays is positive.
+func (r *RetentionReaper) sweep(ctx context.Context) {
+	var webhooks []Webhook
+
+	err := r.db.DB().
+		Model(&Webhook{}).
+		Find(&webhooks).Error
+	if err != nil {
+		r.log.Error(
+			"retention sweep: failed to list webhooks",
+			"error", err,
+		)
+
+		return
+	}
+
+	for i := range webhooks {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		wh := webhooks[i]
+
+		// RetentionDays of zero or less means retain forever.
+		if wh.RetentionDays <= 0 {
+			continue
+		}
+
+		// Nothing to reap if the per-webhook database has never
+		// been created.
+		if !r.dbManager.DBExists(wh.ID) {
+			continue
+		}
+
+		r.reapWebhook(wh.ID, wh.RetentionDays)
+	}
+}
+
+// reapWebhook removes every expired event (and its dependents) from a
+// single webhook's database.
+func (r *RetentionReaper) reapWebhook(
+	webhookID string,
+	retentionDays int,
+) {
+	db, err := r.dbManager.GetDB(webhookID)
+	if err != nil {
+		r.log.Error(
+			"retention sweep: failed to open webhook database",
+			"webhook_id", webhookID,
+			"error", err,
+		)
+
+		return
+	}
+
+	cutoff := time.Now().Add(
+		-time.Duration(retentionDays*hoursPerDay) * time.Hour,
+	)
+
+	deleted, err := reapExpired(db, cutoff)
+	if err != nil {
+		r.log.Error(
+			"retention sweep: failed to reap expired events",
+			"webhook_id", webhookID,
+			"error", err,
+		)
+
+		return
+	}
+
+	if deleted > 0 {
+		r.log.Info(
+			"retention sweep: reaped expired events",
+			"webhook_id", webhookID,
+			"retention_days", retentionDays,
+			"events_deleted", deleted,
+		)
+	}
+}
+
+// reapExpired hard-deletes, in foreign-key-safe order, the delivery
+// results, deliveries, and events associated with events older than
+// cutoff. Deletes are unscoped so rows are physically removed rather
+// than soft-deleted, reclaiming disk. It returns the number of events
+// deleted.
+func reapExpired(db *gorm.DB, cutoff time.Time) (int64, error) {
+	// Fresh subqueries are built per statement to avoid reusing a
+	// mutated builder across executions.
+	expiredEventIDs := func() *gorm.DB {
+		return db.Model(&Event{}).
+			Select("id").
+			Where("created_at < ?", cutoff)
+	}
+	expiredDeliveryIDs := func() *gorm.DB {
+		return db.Model(&Delivery{}).
+			Select("id").
+			Where("event_id IN (?)", expiredEventIDs())
+	}
+
+	// 1. Delivery results whose delivery belongs to an expired event.
+	res := db.Unscoped().
+		Where("delivery_id IN (?)", expiredDeliveryIDs()).
+		Delete(&DeliveryResult{})
+	if res.Error != nil {
+		return 0, fmt.Errorf(
+			"deleting expired delivery results: %w",
+			res.Error,
+		)
+	}
+
+	// 2. Deliveries belonging to an expired event.
+	del := db.Unscoped().
+		Where("event_id IN (?)", expiredEventIDs()).
+		Delete(&Delivery{})
+	if del.Error != nil {
+		return 0, fmt.Errorf(
+			"deleting expired deliveries: %w",
+			del.Error,
+		)
+	}
+
+	// 3. The expired events themselves.
+	ev := db.Unscoped().
+		Where("created_at < ?", cutoff).
+		Delete(&Event{})
+	if ev.Error != nil {
+		return 0, fmt.Errorf(
+			"deleting expired events: %w",
+			ev.Error,
+		)
+	}
+
+	return ev.RowsAffected, nil
+}

internal/database/retention_test.go

@@ -0,0 +1,277 @@
+package database_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/fx/fxtest"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+	"sneak.berlin/go/webhooker/internal/config"
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/globals"
+	"sneak.berlin/go/webhooker/internal/logger"
+)
+
+// retentionTestEnv bundles the pieces a retention test drives.
+type retentionTestEnv struct {
+	reaper *database.RetentionReaper
+	mainDB *database.Database
+	mgr    *database.WebhookDBManager
+}
+
+func setupRetentionTest(t *testing.T) *retentionTestEnv {
+	t.Helper()
+
+	lc := fxtest.NewLifecycle(t)
+
+	g := &globals.Globals{
+		Appname: "webhooker-test",
+		Version: "test",
+	}
+
+	l, err := logger.New(lc, logger.LoggerParams{Globals: g})
+	require.NoError(t, err)
+
+	cfg := &config.Config{
+		DataDir:     t.TempDir(),
+		Environment: "dev",
+	}
+
+	mainDB, err := database.New(lc, database.DatabaseParams{
+		Config: cfg,
+		Logger: l,
+	})
+	require.NoError(t, err)
+
+	mgr, err := database.NewWebhookDBManager(
+		lc,
+		database.WebhookDBManagerParams{Config: cfg, Logger: l},
+	)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	require.NoError(t, lc.Start(ctx))
+	t.Cleanup(func() { require.NoError(t, lc.Stop(ctx)) })
+
+	return &retentionTestEnv{
+		reaper: database.NewTestRetentionReaper(mainDB, mgr),
+		mainDB: mainDB,
+		mgr:    mgr,
+	}
+}
+
+// createWebhook inserts a webhook row into the main database with the
+// given retention policy and returns its ID.
+func createWebhook(
+	t *testing.T,
+	db *gorm.DB,
+	retentionDays int,
+) string {
+	t.Helper()
+
+	wh := &database.Webhook{
+		UserID:        uuid.New().String(),
+		Name:          "test-webhook",
+		RetentionDays: retentionDays,
+	}
+	require.NoError(
+		t,
+		db.Omit(clause.Associations).Create(wh).Error,
+	)
+
+	// The RetentionDays column carries a GORM default of 30, so a
+	// zero (or negative) value passed to Create is replaced by that
+	// default. Force the requested value explicitly so the
+	// retain-forever (<= 0) path can be exercised.
+	require.NoError(
+		t,
+		db.Model(wh).
+			Update("retention_days", retentionDays).Error,
+	)
+
+	return wh.ID
+}
+
+// eventChain is the set of row IDs seeded for a single event.
+type eventChain struct {
+	eventID    string
+	deliveryID string
+	resultID   string
+}
+
+// seedEventChain creates an event with one delivery and one delivery
+// result, all stamped with createdAt, and returns their IDs.
+func seedEventChain(
+	t *testing.T,
+	db *gorm.DB,
+	webhookID string,
+	createdAt time.Time,
+) eventChain {
+	t.Helper()
+
+	event := &database.Event{
+		WebhookID:    webhookID,
+		EntrypointID: uuid.New().String(),
+		Method:       "POST",
+		Body:         `{"seed": true}`,
+		ContentType:  "application/json",
+	}
+	event.CreatedAt = createdAt
+	require.NoError(t, db.Create(event).Error)
+
+	delivery := &database.Delivery{
+		EventID:  event.ID,
+		TargetID: uuid.New().String(),
+		Status:   database.DeliveryStatusDelivered,
+	}
+	delivery.CreatedAt = createdAt
+	require.NoError(t, db.Create(delivery).Error)
+
+	result := &database.DeliveryResult{
+		DeliveryID: delivery.ID,
+		AttemptNum: 1,
+		Success:    true,
+		StatusCode: 200,
+		Duration:   10,
+	}
+	result.CreatedAt = createdAt
+	require.NoError(t, db.Create(result).Error)
+
+	return eventChain{
+		eventID:    event.ID,
+		deliveryID: delivery.ID,
+		resultID:   result.ID,
+	}
+}
+
+// countByID returns how many rows of model match the given id,
+// counting even hard-deletable rows via Unscoped.
+func countByID(
+	t *testing.T,
+	db *gorm.DB,
+	model any,
+	id string,
+) int64 {
+	t.Helper()
+
+	var n int64
+
+	require.NoError(
+		t,
+		db.Unscoped().Model(model).
+			Where("id = ?", id).Count(&n).Error,
+	)
+
+	return n
+}
+
+func assertChainGone(
+	t *testing.T,
+	db *gorm.DB,
+	chain eventChain,
+) {
+	t.Helper()
+
+	assert.Zero(
+		t,
+		countByID(t, db, &database.Event{}, chain.eventID),
+		"expired event should be removed",
+	)
+	assert.Zero(
+		t,
+		countByID(t, db, &database.Delivery{}, chain.deliveryID),
+		"expired delivery should be removed",
+	)
+	assert.Zero(
+		t,
+		countByID(
+			t, db, &database.DeliveryResult{}, chain.resultID,
+		),
+		"expired delivery result should be removed",
+	)
+}
+
+func assertChainPresent(
+	t *testing.T,
+	db *gorm.DB,
+	chain eventChain,
+) {
+	t.Helper()
+
+	assert.Equal(
+		t,
+		int64(1),
+		countByID(t, db, &database.Event{}, chain.eventID),
+		"recent event should be retained",
+	)
+	assert.Equal(
+		t,
+		int64(1),
+		countByID(t, db, &database.Delivery{}, chain.deliveryID),
+		"recent delivery should be retained",
+	)
+	assert.Equal(
+		t,
+		int64(1),
+		countByID(
+			t, db, &database.DeliveryResult{}, chain.resultID,
+		),
+		"recent delivery result should be retained",
+	)
+}
+
+func TestRetentionReaper_ReapsExpiredKeepsRecent(t *testing.T) {
+	t.Parallel()
+
+	env := setupRetentionTest(t)
+
+	const retentionDays = 30
+
+	webhookID := createWebhook(
+		t, env.mainDB.DB(), retentionDays,
+	)
+
+	db, err := env.mgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	now := time.Now()
+	old := seedEventChain(
+		t, db, webhookID,
+		now.Add(-40*24*time.Hour),
+	)
+	recent := seedEventChain(
+		t, db, webhookID,
+		now.Add(-1*24*time.Hour),
+	)
+
+	env.reaper.ExportSweep(context.Background())
+
+	assertChainGone(t, db, old)
+	assertChainPresent(t, db, recent)
+}
+
+func TestRetentionReaper_RetainsForeverWhenNonPositive(t *testing.T) {
+	t.Parallel()
+
+	env := setupRetentionTest(t)
+
+	// RetentionDays of zero means retain forever.
+	webhookID := createWebhook(t, env.mainDB.DB(), 0)
+
+	db, err := env.mgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	ancient := seedEventChain(
+		t, db, webhookID,
+		time.Now().Add(-365*24*time.Hour),
+	)
+
+	env.reaper.ExportSweep(context.Background())
+
+	assertChainPresent(t, db, ancient)
+}