webhooker/internal/delivery/archive_sweeper.go

package delivery

import (
	"context"
	"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,
	}

	lc.Append(fx.Hook{
		OnStart: func(ctx context.Context) error {
			s.start(ctx)

			return nil
		},
		OnStop: func(_ context.Context) error {
			s.stop()

			return nil
		},
	})

	return s
}

func (s *ArchiveSweeper) start(ctx context.Context) {
	ctx, cancel := context.WithCancel(ctx)
	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 {
		s.log.Error(
			"archive sweep: failed to prune archive",
			"webhook_id", target.WebhookID,
			"target_id", target.ID,
			"error", err,
		)
	}
}