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 }