webhooker/internal/database/retention.go

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
}