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..414a3bb 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,30 @@ 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. If the variable is set but cannot be parsed, it returns a +// wrapped error naming the key and the bad value, so startup fails +// loudly rather than silently falling back to the default. +func envDuration( + key string, + defaultValue time.Duration, +) (time.Duration, error) { + v := os.Getenv(key) + if v == "" { + return defaultValue, nil + } + + d, err := time.ParseDuration(v) + if err != nil { + return 0, fmt.Errorf( + "invalid duration for %s: %q: %w", key, v, err, + ) + } + + return d, nil +} + // New creates a Config by reading environment variables. // //nolint:revive // lc parameter is required by fx even if unused. @@ -118,18 +151,30 @@ func New(lc fx.Lifecycle, params ConfigParams) (*Config, error) { ) } + // Parse the retention sweep interval; a set-but-unparseable value + // is a hard error so fx aborts startup rather than silently using + // the default. + retentionSweepInterval, err := envDuration( + "RETENTION_SWEEP_INTERVAL", + defaultRetentionSweepInterval, + ) + if err != nil { + return nil, err + } + // Load configuration values from environment variables s := &Config{ - DataDir: envString("DATA_DIR"), - Debug: envBool("DEBUG", false), - MaintenanceMode: envBool("MAINTENANCE_MODE", false), - Environment: environment, - MetricsUsername: envString("METRICS_USERNAME"), - MetricsPassword: envString("METRICS_PASSWORD"), - Port: envInt("PORT", defaultPort), - SentryDSN: envString("SENTRY_DSN"), - log: log, - params: ¶ms, + DataDir: envString("DATA_DIR"), + Debug: envBool("DEBUG", false), + MaintenanceMode: envBool("MAINTENANCE_MODE", false), + Environment: environment, + MetricsUsername: envString("METRICS_USERNAME"), + MetricsPassword: envString("METRICS_PASSWORD"), + Port: envInt("PORT", defaultPort), + SentryDSN: envString("SENTRY_DSN"), + RetentionSweepInterval: retentionSweepInterval, + log: log, + params: ¶ms, } // Set default DataDir. All SQLite databases (main application @@ -151,6 +196,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/config/config_test.go b/internal/config/config_test.go index b17b05d..7e3c2c8 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -3,6 +3,7 @@ package config_test import ( "os" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -120,6 +121,100 @@ func testEnvironmentConfigSuccess( assert.Equal(t, isProd, cfg.IsProd()) } +func TestRetentionSweepInterval(t *testing.T) { + tests := []struct { + name string + set bool + value string + expectError bool + expected time.Duration + }{ + { + name: "unset uses default", + set: false, + expected: time.Hour, + }, + { + name: "valid value is parsed", + set: true, + value: "15m", + expected: 15 * time.Minute, + }, + { + name: "unparseable value fails startup", + set: true, + value: "not-a-duration", + expectError: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Cannot use t.Parallel() here because t.Setenv + // is incompatible with parallel subtests. + t.Setenv("WEBHOOKER_ENVIRONMENT", "dev") + + if tt.set { + t.Setenv("RETENTION_SWEEP_INTERVAL", tt.value) + } else { + require.NoError(t, os.Unsetenv( + "RETENTION_SWEEP_INTERVAL", + )) + } + + if tt.expectError { + testRetentionSweepIntervalError(t) + } else { + testRetentionSweepIntervalSuccess(t, tt.expected) + } + }) + } +} + +func testRetentionSweepIntervalError(t *testing.T) { + t.Helper() + + var cfg *config.Config + + app := fx.New( + fx.NopLogger, + fx.Provide( + globals.New, + logger.New, + config.New, + ), + fx.Populate(&cfg), + ) + + assert.Error(t, app.Err()) +} + +func testRetentionSweepIntervalSuccess( + t *testing.T, + expected time.Duration, +) { + t.Helper() + + var cfg *config.Config + + app := fxtest.New( + t, + fx.Provide( + globals.New, + logger.New, + config.New, + ), + fx.Populate(&cfg), + ) + require.NoError(t, app.Err()) + + app.RequireStart() + + defer app.RequireStop() + + assert.Equal(t, expected, cfg.RetentionSweepInterval) +} + func TestDefaultDataDir(t *testing.T) { for _, env := range []string{"", "dev", "prod"} { name := env 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) +}