package delivery import ( "context" "errors" "log/slog" "sync" "time" "go.uber.org/fx" "sneak.berlin/go/webhooker/internal/config" "sneak.berlin/go/webhooker/internal/database" "sneak.berlin/go/webhooker/internal/logger" ) // ArchiveSweeperParams holds the fx dependencies for the // ArchiveSweeper. type ArchiveSweeperParams struct { fx.In Config *config.Config Database *database.Database Engine *Engine Logger *logger.Logger } // ArchiveSweeper periodically prunes expired rows from // per-webhook archive databases whose database target carries a // positive expiry. // // Without it, pruning happens only when an archive is // (re)opened, and archives are only ever reopened by writes: an // archive belonging to a webhook that has stopped receiving // events would keep its expired rows forever. The sweep closes // that gap without changing anything for archives whose expiry // is unset or "never". // // It reuses Config.RetentionSweepInterval rather than // introducing a second interval: this is a retention sweep with // the same semantics as the event retention reaper. type ArchiveSweeper struct { db *database.Database eng *Engine log *slog.Logger interval time.Duration cancel context.CancelFunc wg sync.WaitGroup } // NewArchiveSweeper creates the archive sweeper and registers // its fx lifecycle hooks. The background sweep loop starts on // OnStart and stops cleanly on OnStop via context cancellation. func NewArchiveSweeper( lc fx.Lifecycle, params ArchiveSweeperParams, ) *ArchiveSweeper { s := &ArchiveSweeper{ db: params.Database, eng: params.Engine, log: params.Logger.Get(), interval: params.Config.RetentionSweepInterval, } s.registerHooks(lc) return s } // registerHooks wires the sweeper's start and stop into the fx // lifecycle. Both hook contexts are deliberately ignored: see // start for why the background loop must not inherit the start // hook's context, and stop for why shutdown blocks on the loop // rather than on the stop hook's deadline. func (s *ArchiveSweeper) registerHooks(lc fx.Lifecycle) { lc.Append(fx.Hook{ //nolint:contextcheck // Not passing the hook context is // the point: see start. OnStart: func(_ context.Context) error { s.start() return nil }, OnStop: func(_ context.Context) error { s.stop() return nil }, }) } // start launches the background sweep loop. // // The loop's context is derived from context.Background(), NOT // from the fx OnStart hook context. The hook context carries // fx's start timeout (15s by default), so a loop derived from it // is cancelled 15 seconds after the application starts — long // before the first tick under the default one-hour sweep // interval, leaving a sweeper that never sweeps. A long-lived // goroutine must outlive the startup phase, so its lifetime is // bounded by OnStop instead: stop cancels this context and waits // on the WaitGroup. func (s *ArchiveSweeper) start() { ctx, cancel := context.WithCancel(context.Background()) s.cancel = cancel s.wg.Add(1) go s.run(ctx) s.log.Info( "archive sweeper started", "interval", s.interval.String(), ) } func (s *ArchiveSweeper) stop() { s.log.Info("archive sweeper stopping") if s.cancel != nil { s.cancel() } s.wg.Wait() s.log.Info("archive sweeper stopped") } func (s *ArchiveSweeper) run(ctx context.Context) { defer s.wg.Done() ticker := time.NewTicker(s.interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: s.sweep(ctx) } } } // sweep prunes every archive whose database target declares a // positive expiry. Targets belonging to a deleted webhook are // soft-deleted along with it, so GORM's default scope already // excludes them. // // A failure for one webhook is logged and the sweep continues, // matching how the write path already treats a prune error as // non-fatal. func (s *ArchiveSweeper) sweep(ctx context.Context) { var targets []database.Target err := s.db.DB(). Model(&database.Target{}). Where("type = ?", database.TargetTypeDatabase). Find(&targets).Error if err != nil { s.log.Error( "archive sweep: failed to list database targets", "error", err, ) return } for i := range targets { select { case <-ctx.Done(): return default: } s.sweepTarget(&targets[i]) } } // sweepTarget prunes the archive of a single database target. // A missing, empty, or "never" expiry parses as a zero duration // and is skipped entirely, so those archives keep exactly the // behaviour they had before the sweep existed. func (s *ArchiveSweeper) sweepTarget(target *database.Target) { expiry, err := parseArchiveExpiry(target.Config) if err != nil { s.log.Error( "archive sweep: invalid database target config", "webhook_id", target.WebhookID, "target_id", target.ID, "error", err, ) return } if expiry <= 0 { return } if s.eng == nil || s.eng.dbTarget == nil { return } err = s.eng.dbTarget.sweepWebhook(target.WebhookID, expiry) if err == nil { return } // A writer evicted underneath the sweep means the operator // deleted the webhook (or its last database target) while the // sweep was walking the target list. That is an ordinary // interleaving, not a failure, so it must not produce an // error line. if errors.Is(err, errArchiveWriterEvicted) { s.log.Debug( "archive sweep: writer evicted mid-sweep", "webhook_id", target.WebhookID, "target_id", target.ID, ) return } s.log.Error( "archive sweep: failed to prune archive", "webhook_id", target.WebhookID, "target_id", target.ID, "error", err, ) }