From f6abc9b39280b4391a10017159ce2b63b6ee93d0 Mon Sep 17 00:00:00 2001 From: sneak Date: Fri, 7 Aug 2026 20:19:01 +0700 Subject: [PATCH] Add per-webhook event retention reaper (closes #63) --- cmd/webhooker/main.go | 10 +- internal/config/config.go | 39 +++- internal/database/export_test.go | 31 ++++ internal/database/retention.go | 252 +++++++++++++++++++++++++ internal/database/retention_test.go | 277 ++++++++++++++++++++++++++++ 5 files changed, 604 insertions(+), 5 deletions(-) create mode 100644 internal/database/export_test.go create mode 100644 internal/database/retention.go create mode 100644 internal/database/retention_test.go diff --git a/cmd/webhooker/main.go b/cmd/webhooker/main.go index f53b5ce..59fc655 100644 --- a/cmd/webhooker/main.go +++ b/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() } diff --git a/internal/config/config.go b/internal/config/config.go index 90e5d8d..95749ac 100644 --- a/internal/config/config.go +++ b/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,8 +56,12 @@ type Config struct { MetricsUsername string Port int SentryDSN string - params *ConfigParams - log *slog.Logger + + // RetentionSweepInterval is how often the retention reaper runs. + RetentionSweepInterval time.Duration + + params *ConfigParams + log *slog.Logger } // IsDev returns true if running in development environment. @@ -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,8 +154,12 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) { MetricsPassword: envString("METRICS_PASSWORD"), Port: envInt("PORT", defaultPort), SentryDSN: envString("SENTRY_DSN"), - log: log, - params: ¶ms, + RetentionSweepInterval: envDuration( + "RETENTION_SWEEP_INTERVAL", + defaultRetentionSweepInterval, + ), + log: log, + params: ¶ms, } // Set default DataDir. All SQLite databases (main application @@ -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 != "", diff --git a/internal/database/export_test.go b/internal/database/export_test.go new file mode 100644 index 0000000..29321fe --- /dev/null +++ b/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) +} diff --git a/internal/database/retention.go b/internal/database/retention.go new file mode 100644 index 0000000..23d516f --- /dev/null +++ b/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 +} diff --git a/internal/database/retention_test.go b/internal/database/retention_test.go new file mode 100644 index 0000000..2df0b0e --- /dev/null +++ b/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) +}