routewatch/internal/routewatch/dbmaintainer.go

package routewatch

import (
	"context"
	"errors"
	"log/slog"
	"sync"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/database"
)

// Database maintenance configuration constants.
const (
	// checkpointInterval is how often to run WAL checkpoint.
	// Frequent checkpoints keep the WAL small, improving read performance.
	// Under heavy write load, we need aggressive checkpointing.
	checkpointInterval = 5 * time.Second

	// vacuumInterval is how often to run incremental vacuum.
	// Since incremental vacuum only frees ~1000 pages (~4MB) per run,
	// we run it frequently to keep up with deletions.
	vacuumInterval = 10 * time.Minute

	// analyzeInterval is how often to run ANALYZE.
	analyzeInterval = 1 * time.Hour

	// originChangesPruneInterval is how often to prune the origin_changes
	// table.
	originChangesPruneInterval = 1 * time.Hour

	// checkpointTimeout is the max time for WAL checkpoint.
	checkpointTimeout = 10 * time.Second

	// vacuumTimeout is the max time for incremental vacuum (should be quick).
	vacuumTimeout = 30 * time.Second

	// analyzeTimeout is the max time for ANALYZE.
	analyzeTimeout = 5 * time.Minute

	// originChangesPruneTimeout is the max time for the origin changes prune.
	originChangesPruneTimeout = 5 * time.Minute
)

// checkpointIncompleteMessage is logged, at debug level like a completed
// checkpoint, when a checkpoint leaves part of the WAL for a later one: while
// reads overlap, most do. The database warns when the WAL file stays over its
// size limit.
const checkpointIncompleteMessage = "WAL checkpoint incomplete"

// DBMaintainer handles background database maintenance tasks.
type DBMaintainer struct {
	db     database.Store
	logger *slog.Logger
	stopCh chan struct{}
	done   chan struct{} // closed when the maintenance goroutine returns

	// Stats tracking
	statsMu             sync.Mutex
	lastCheckpoint      time.Time
	lastVacuum          time.Time
	lastAnalyze         time.Time
	checkpointCount     int
	vacuumCount         int
	analyzeCount        int
	lastCheckpointError error
	lastVacuumError     error
	lastAnalyzeError    error
}

// NewDBMaintainer creates a new database maintainer.
func NewDBMaintainer(db database.Store, logger *slog.Logger) *DBMaintainer {
	return &DBMaintainer{
		db:     db,
		logger: logger.With("component", "db_maintainer"),
		stopCh: make(chan struct{}),
		done:   make(chan struct{}),
	}
}

// Start begins the background maintenance goroutine. It runs until Stop:
// the end of ctx does not stop it.
func (m *DBMaintainer) Start(ctx context.Context) {
	go m.run(context.WithoutCancel(ctx))

	m.logger.Info("Database maintainer started",
		"checkpoint_interval", checkpointInterval,
		"vacuum_interval", vacuumInterval,
		"analyze_interval", analyzeInterval,
		"origin_changes_prune_interval", originChangesPruneInterval,
	)
}

// Stop stops the maintainer and waits, until ctx ends, for a maintenance task
// in progress to finish.
func (m *DBMaintainer) Stop(ctx context.Context) {
	close(m.stopCh)

	select {
	case <-m.done:
		m.logger.Info("Database maintainer stopped")
	case <-ctx.Done():
		m.logger.Warn("Stop deadline passed before the database maintainer " +
			"stopped")
	}
}

// run is the main background loop.
func (m *DBMaintainer) run(ctx context.Context) {
	defer close(m.done)

	// Use different timers for each task
	checkpointTimer := time.NewTimer(checkpointInterval)
	defer checkpointTimer.Stop()

	vacuumTimer := time.NewTimer(vacuumInterval)
	defer vacuumTimer.Stop()

	analyzeTimer := time.NewTimer(analyzeInterval)
	defer analyzeTimer.Stop()

	pruneTimer := time.NewTimer(originChangesPruneInterval)
	defer pruneTimer.Stop()

	for {
		select {
		case <-m.stopCh:
			return

		case <-checkpointTimer.C:
			m.runCheckpoint(ctx)
			checkpointTimer.Reset(checkpointInterval)

		case <-vacuumTimer.C:
			m.runVacuum(ctx)
			vacuumTimer.Reset(vacuumInterval)

		case <-analyzeTimer.C:
			m.runAnalyze(ctx)
			analyzeTimer.Reset(analyzeInterval)

		case <-pruneTimer.C:
			m.runPruneOriginChanges(ctx)
			pruneTimer.Reset(originChangesPruneInterval)
		}
	}
}

// runCheckpoint performs a WAL checkpoint to keep the WAL file small.
func (m *DBMaintainer) runCheckpoint(ctx context.Context) {
	duration, err := runWithTimeout(ctx, checkpointTimeout, m.db.Checkpoint)
	m.recordResult(&m.lastCheckpoint, &m.lastCheckpointError, &m.checkpointCount, err)

	switch {
	case errors.Is(err, database.ErrCheckpointIncomplete):
		m.logger.Debug(checkpointIncompleteMessage,
			"error", err, "duration", duration)
	case err != nil:
		m.logger.Error("WAL checkpoint failed", "error", err, "duration", duration)
	default:
		m.logger.Debug("WAL checkpoint completed", "duration", duration)
	}
}

// runVacuum performs an incremental vacuum operation on the database.
func (m *DBMaintainer) runVacuum(ctx context.Context) {
	m.logger.Debug("Running incremental vacuum")

	duration, err := runWithTimeout(ctx, vacuumTimeout, m.db.Vacuum)
	m.recordResult(&m.lastVacuum, &m.lastVacuumError, &m.vacuumCount, err)

	if err != nil {
		m.logger.Error("Incremental vacuum failed", "error", err, "duration", duration)
	} else {
		m.logger.Debug("Incremental vacuum completed", "duration", duration)
	}
}

// runAnalyze performs an ANALYZE operation on the database.
func (m *DBMaintainer) runAnalyze(ctx context.Context) {
	m.logger.Info("Starting database ANALYZE")

	duration, err := runWithTimeout(ctx, analyzeTimeout, m.db.Analyze)
	m.recordResult(&m.lastAnalyze, &m.lastAnalyzeError, &m.analyzeCount, err)

	if err != nil {
		m.logger.Error("ANALYZE failed", "error", err, "duration", duration)
	} else {
		m.logger.Info("ANALYZE completed", "duration", duration)
	}
}

// runPruneOriginChanges prunes the origin_changes table.
func (m *DBMaintainer) runPruneOriginChanges(ctx context.Context) {
	duration, err := runWithTimeout(ctx, originChangesPruneTimeout,
		m.db.PruneOriginChanges)
	if err != nil {
		m.logger.Error("Origin changes prune failed", "error", err,
			"duration", duration)
	} else {
		m.logger.Debug("Origin changes pruned", "duration", duration)
	}
}

// runWithTimeout runs operation with a context that ends after timeout and
// returns how long it took and its error.
func runWithTimeout(
	ctx context.Context,
	timeout time.Duration,
	operation func(context.Context) error,
) (time.Duration, error) {
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	startTime := time.Now()
	err := operation(ctx)

	return time.Since(startTime), err
}

// recordResult stores when a maintenance task last ran and its error, and
// counts it when it succeeded.
func (m *DBMaintainer) recordResult(
	lastRun *time.Time, lastErr *error, successCount *int, err error,
) {
	m.statsMu.Lock()
	defer m.statsMu.Unlock()

	*lastRun = time.Now()
	*lastErr = err

	if err == nil {
		*successCount++
	}
}