neoirc/internal/db/db.go

// Package db provides database access and migration management.
package db

import (
	"context"
	"database/sql"
	"embed"
	"errors"
	"fmt"
	"io/fs"
	"log/slog"
	"net/url"
	"sort"
	"strconv"
	"strings"

	"go.uber.org/fx"
	"golang.org/x/crypto/bcrypt"
	"sneak.berlin/go/neoirc/internal/config"
	"sneak.berlin/go/neoirc/internal/logger"

	_ "github.com/joho/godotenv/autoload" // .env
	_ "modernc.org/sqlite"                // driver
)

const minMigrationParts = 2

var errForeignKeysInDBURL = errors.New(
	"DBURL must not set foreign_keys, which the server always turns on",
)

// SchemaFiles contains embedded SQL migration files.
//
//go:embed schema/*.sql
var SchemaFiles embed.FS

// Params defines the dependencies for creating a Database.
type Params struct {
	fx.In

	Logger *logger.Logger
	Config *config.Config

	// PasswordCost is the bcrypt cost at which SetPassword hashes
	// a password. The server leaves it unset, which means
	// bcrypt.DefaultCost. Tests set a lower cost to save time.
	PasswordCost int `optional:"true"`
}

// Database manages the SQLite connection and migrations.
type Database struct {
	conn         *sql.DB
	log          *slog.Logger
	params       *Params
	passwordCost int
}

// New creates a new Database and registers lifecycle hooks.
func New(
	lifecycle fx.Lifecycle,
	params Params,
) (*Database, error) {
	passwordCost := params.PasswordCost
	if passwordCost == 0 {
		passwordCost = bcrypt.DefaultCost
	}

	database := &Database{ //nolint:exhaustruct // conn set in OnStart
		params:       &params,
		log:          params.Logger.Get(),
		passwordCost: passwordCost,
	}

	database.log.Info("Database instantiated")

	lifecycle.Append(fx.Hook{
		OnStart: func(ctx context.Context) error {
			database.log.Info("Database OnStart Hook")

			return database.connect(ctx)
		},
		OnStop: func(_ context.Context) error {
			database.log.Info("Database OnStop Hook")

			if database.conn != nil {
				closeErr := database.conn.Close()
				if closeErr != nil {
					return fmt.Errorf(
						"close db: %w", closeErr,
					)
				}
			}

			return nil
		},
	})

	return database, nil
}

func (database *Database) connect(ctx context.Context) error {
	dbURL := database.params.Config.DBURL
	if dbURL == "" {
		dbURL = config.DefaultDBURL
	}

	// SQLite keeps the foreign keys setting per connection. The driver
	// runs each _pragma parameter on every connection it opens. So the
	// one added below also covers the connection the pool opens after
	// a cancelled query breaks the old one. A foreign_keys pragma in
	// DBURL could undo it, because the driver sorts the pragmas.
	_, query, hasQuery := strings.Cut(dbURL, "?")

	urlParams, err := url.ParseQuery(query)
	if err != nil {
		return fmt.Errorf("parse DBURL: %w", err)
	}

	for _, pragma := range urlParams["_pragma"] {
		if strings.Contains(strings.ToLower(pragma), "foreign_keys") {
			return fmt.Errorf(
				"%w: _pragma=%s", errForeignKeysInDBURL, pragma,
			)
		}
	}

	separator := "?"
	if hasQuery {
		separator = "&"
	}

	dbURL += separator + "_pragma=foreign_keys(1)"

	database.log.Info(
		"connecting to database", "url", dbURL,
	)

	conn, err := sql.Open("sqlite", dbURL)
	if err != nil {
		return fmt.Errorf("open database: %w", err)
	}

	err = conn.PingContext(ctx)
	if err != nil {
		return fmt.Errorf("ping database: %w", err)
	}

	conn.SetMaxOpenConns(1)

	database.conn = conn
	database.log.Info("database connected")

	return database.runMigrations(ctx)
}

type migration struct {
	version int
	name    string
	sql     string
}

func (database *Database) runMigrations(
	ctx context.Context,
) error {
	bootstrap, err := SchemaFiles.ReadFile(
		"schema/000.sql",
	)
	if err != nil {
		return fmt.Errorf(
			"read bootstrap migration: %w", err,
		)
	}

	_, err = database.conn.ExecContext(
		ctx, string(bootstrap),
	)
	if err != nil {
		return fmt.Errorf(
			"execute bootstrap migration: %w", err,
		)
	}

	migrations, err := database.loadMigrations()
	if err != nil {
		return err
	}

	for _, mig := range migrations {
		err = database.applyMigration(ctx, mig)
		if err != nil {
			return err
		}
	}

	database.log.Info("database migrations complete")

	return nil
}

func (database *Database) applyMigration(
	ctx context.Context,
	mig migration,
) error {
	var exists int

	err := database.conn.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM schema_migrations
		 WHERE version = ?`,
		mig.version,
	).Scan(&exists)
	if err != nil {
		return fmt.Errorf(
			"check migration %d: %w", mig.version, err,
		)
	}

	if exists > 0 {
		return nil
	}

	database.log.Info(
		"applying migration",
		"version", mig.version,
		"name", mig.name,
	)

	return database.execMigration(ctx, mig)
}

func (database *Database) execMigration(
	ctx context.Context,
	mig migration,
) error {
	transaction, err := database.conn.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf(
			"begin tx for migration %d: %w",
			mig.version, err,
		)
	}

	_, err = transaction.ExecContext(ctx, mig.sql)
	if err != nil {
		_ = transaction.Rollback()

		return fmt.Errorf(
			"apply migration %d (%s): %w",
			mig.version, mig.name, err,
		)
	}

	_, err = transaction.ExecContext(ctx,
		`INSERT INTO schema_migrations (version)
		 VALUES (?)`,
		mig.version,
	)
	if err != nil {
		_ = transaction.Rollback()

		return fmt.Errorf(
			"record migration %d: %w",
			mig.version, err,
		)
	}

	err = transaction.Commit()
	if err != nil {
		return fmt.Errorf(
			"commit migration %d: %w",
			mig.version, err,
		)
	}

	return nil
}

func (database *Database) loadMigrations() (
	[]migration,
	error,
) {
	entries, err := fs.ReadDir(SchemaFiles, "schema")
	if err != nil {
		return nil, fmt.Errorf(
			"read schema dir: %w", err,
		)
	}

	migrations := make([]migration, 0, len(entries))

	for _, entry := range entries {
		if entry.IsDir() ||
			!strings.HasSuffix(entry.Name(), ".sql") {
			continue
		}

		parts := strings.SplitN(
			entry.Name(), "_", minMigrationParts,
		)
		if len(parts) < minMigrationParts {
			continue
		}

		version, parseErr := strconv.Atoi(parts[0])
		if parseErr != nil {
			continue
		}

		// Skip bootstrap migration; it is executed separately.
		if version == 0 {
			continue
		}

		content, readErr := SchemaFiles.ReadFile(
			"schema/" + entry.Name(),
		)
		if readErr != nil {
			return nil, fmt.Errorf(
				"read migration %s: %w",
				entry.Name(), readErr,
			)
		}

		migrations = append(migrations, migration{
			version: version,
			name:    entry.Name(),
			sql:     string(content),
		})
	}

	sort.Slice(migrations, func(i, j int) bool {
		return migrations[i].version < migrations[j].version
	})

	return migrations, nil
}