odidere/internal/db/schema.go

package db

import (
	"context"
	"embed"
	"fmt"
	"log/slog"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
)

//go:embed schema/*.sql
var migrations embed.FS

// ApplySchema reads the current database version from PRAGMA user_version
// and applies any pending migrations in a single transaction. Each migration
// file runs in order; if any fail the entire transaction rolls back and the
// version is unchanged. log must be non-nil; pass
// slog.New(slog.DiscardHandler) to run without logs.
//
// Migration files are named <version>.<description>. The version
// must be a non-negative integer. On success, PRAGMA user_version is set
// to the highest migration version within the same transaction.
//
// If any migration fails the application should refuse to start, since
// the application code expects a schema that matches the latest migration.
func (s *SQLite) ApplySchema(ctx context.Context, log *slog.Logger) error {
	if log == nil {
		return fmt.Errorf("missing logger")
	}

	var currentVersion int
	if err := s.Writer.QueryRowContext(
		ctx,
		"PRAGMA user_version",
	).Scan(&currentVersion); err != nil {
		return fmt.Errorf("read user_version: %w", err)
	}
	log.InfoContext(
		ctx,
		"current schema version",
		slog.Int("user_version", currentVersion),
	)

	entries, err := migrations.ReadDir("schema")
	if err != nil {
		return fmt.Errorf("read schema directory: %w", err)
	}

	var pending []struct {
		name    string
		sql     []byte
		version int
	}
	for _, e := range entries {
		if e.IsDir() {
			continue
		}
		if !strings.HasSuffix(e.Name(), ".sql") {
			continue
		}
		version, err := parseVersion(e.Name())
		if err != nil {
			return err
		}
		if version <= currentVersion {
			log.InfoContext(
				ctx,
				"skipping migration",
				slog.String("file", e.Name()),
				slog.Int("version", version),
			)
			continue
		}
		sql, err := migrations.ReadFile(
			filepath.Join("schema", e.Name()),
		)
		if err != nil {
			return fmt.Errorf(
				"read migration %q: %w", e.Name(), err,
			)
		}
		pending = append(pending, struct {
			name    string
			sql     []byte
			version int
		}{name: e.Name(), sql: sql, version: version})
	}
	sort.Slice(pending, func(i, j int) bool {
		return pending[i].version < pending[j].version
	})
	if len(pending) == 0 {
		log.InfoContext(ctx, "no pending migrations")
		return nil
	}
	log.InfoContext(
		ctx,
		"pending migrations",
		slog.Int("count", len(pending)),
	)

	tx, err := s.Writer.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin transaction: %w", err)
	}
	for _, m := range pending {
		if _, err := tx.ExecContext(ctx, string(m.sql)); err != nil {
			if err := tx.Rollback(); err != nil {
				log.ErrorContext(
					ctx,
					"rollback failed after apply error",
					slog.Any("error", err),
				)
			}
			return fmt.Errorf(
				"apply migration %q (v%d): %w",
				m.name, m.version, err,
			)
		}
	}

	// Set version to the latest migration within the transaction.
	latestVersion := pending[len(pending)-1].version
	if _, err := tx.ExecContext(
		ctx,
		fmt.Sprintf("PRAGMA user_version = %d", latestVersion),
	); err != nil {
		if err := tx.Rollback(); err != nil {
			log.ErrorContext(
				ctx,
				"rollback failed after version set error",
				slog.Any("error", err),
			)
		}
		return fmt.Errorf(
			"set user_version to %d: %w",
			latestVersion, err,
		)
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit migrations: %w", err)
	}
	log.InfoContext(
		ctx,
		"schema up to date",
		slog.Int("user_version", latestVersion),
	)

	return nil
}

// parseVersion extracts the leading integer from a migration filename
// such as "1.init.sql" -> 1 or "12.add-foo.sql" -> 12.
func parseVersion(name string) (int, error) {
	dot := strings.Index(name, ".")
	if dot <= 0 {
		return 0, fmt.Errorf(
			"invalid filename %q: no dot separator", name,
		)
	}
	v, err := strconv.Atoi(name[:dot])
	if err != nil {
		return 0, fmt.Errorf(
			"invalid filename %q: version is not an integer: %w",
			name, err,
		)
	}
	return v, nil
}