webhooker/internal/delivery/archive_sweeper.go

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/lifecycle"
	"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. The start hook's context is deliberately ignored
// (see start for why the background loop must not inherit it);
// the stop hook's context is honoured (see stop).
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(ctx context.Context) error {
			return s.stop(ctx)
		},
	})
}

// 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(),
	)
}

// stop cancels the sweep loop's context and waits for it to
// exit, bounded by the stop hook's context: a prune wedged on a
// locked archive must not hang the process past fx's stop
// timeout.
func (s *ArchiveSweeper) stop(ctx context.Context) error {
	s.log.Info("archive sweeper stopping")

	if s.cancel != nil {
		s.cancel()
	}

	err := lifecycle.WaitForShutdown(
		ctx, s.log, "archive sweeper", &s.wg,
	)
	if err != nil {
		return err
	}

	s.log.Info("archive sweeper stopped")

	return nil
}

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,
	)
}