glchat/internal/database/database.go

package database

import (
	"context"
	"database/sql"
	"embed"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"time"

	// Регистрирует драйвер "sqlite3" для database/sql (CGO, AGENT.md 5.1).
	_ "github.com/mattn/go-sqlite3"
	"github.com/pressly/goose/v3"
)

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

type DB struct {
	Writer *sql.DB
	Reader *sql.DB
	path   string
}

type Options struct {
	Path      string
	ReadPool  int
	CacheSize string
	MmapSize  string
	Migrate   bool
}

const (
	writerMaxOpenConns = 1
	connMaxLifetime    = time.Hour
)

func Open(ctx context.Context, opts Options) (*DB, error) {
	if opts.Path == "" {
		return nil, errors.New("database path is required")
	}
	if opts.ReadPool < 1 {
		opts.ReadPool = max(2, runtime.NumCPU())
	}
	if err := EnsureDir(opts.Path); err != nil {
		return nil, err
	}

	writer, err := sql.Open("sqlite3", dsn(opts, true))
	if err != nil {
		return nil, fmt.Errorf("open writer connection: %w", err)
	}
	writer.SetMaxOpenConns(writerMaxOpenConns)
	writer.SetMaxIdleConns(writerMaxOpenConns)
	writer.SetConnMaxLifetime(connMaxLifetime)

	if err := applyPragmas(ctx, writer, opts); err != nil {
		closeQuietly(writer)
		return nil, err
	}
	if err := writer.PingContext(ctx); err != nil {
		closeQuietly(writer)
		return nil, fmt.Errorf("ping writer connection: %w", err)
	}

	db := &DB{Writer: writer, path: opts.Path}
	if opts.Migrate {
		if err := db.Migrate(ctx); err != nil {
			closeQuietly(db.Writer)
			return nil, err
		}
	}

	reader, err := sql.Open("sqlite3", dsn(opts, false))
	if err != nil {
		closeQuietly(db.Writer)
		return nil, fmt.Errorf("open reader connection: %w", err)
	}
	reader.SetMaxOpenConns(opts.ReadPool)
	reader.SetMaxIdleConns(opts.ReadPool)
	reader.SetConnMaxLifetime(connMaxLifetime)
	if err := applyPragmas(ctx, reader, opts); err != nil {
		closeQuietly(reader)
		closeQuietly(db.Writer)
		return nil, err
	}
	if err := reader.PingContext(ctx); err != nil {
		closeQuietly(reader)
		closeQuietly(db.Writer)
		return nil, fmt.Errorf("ping reader connection: %w", err)
	}
	db.Reader = reader
	return db, nil
}

// dsn builds the mattn/go-sqlite3 DSN. The journal mode needs an exclusive
// lock to switch, so it is set once on the writer pool; readers reuse the
// mode already persisted in the database file.
func dsn(opts Options, writer bool) string {
	params := []string{"_foreign_keys=ON", "_busy_timeout=5000"}
	if writer {
		params = append(params, "_journal_mode=WAL", "_txlock=immediate")
	} else {
		params = append(params, "mode=ro", "_query_only=ON")
	}
	return "file:" + opts.Path + "?" + strings.Join(params, "&")
}

func applyPragmas(ctx context.Context, db *sql.DB, opts Options) error {
	pragmas := []string{
		"PRAGMA foreign_keys = ON",
		"PRAGMA busy_timeout = 5000",
		"PRAGMA synchronous = NORMAL",
		"PRAGMA temp_store = MEMORY",
	}
	if opts.CacheSize != "" {
		pragmas = append(pragmas, "PRAGMA cache_size = "+opts.CacheSize)
	}
	if opts.MmapSize != "" {
		pragmas = append(pragmas, "PRAGMA mmap_size = "+opts.MmapSize)
	}
	for _, stmt := range pragmas {
		if _, err := db.ExecContext(ctx, stmt); err != nil {
			return fmt.Errorf("apply %q: %w", stmt, err)
		}
	}
	return nil
}

func (db *DB) Path() string { return db.path }

func (db *DB) Migrate(ctx context.Context) error {
	goose.SetBaseFS(migrationsFS)
	goose.SetLogger(goose.NopLogger())
	if err := goose.SetDialect("sqlite3"); err != nil {
		return fmt.Errorf("set goose dialect: %w", err)
	}
	if err := goose.UpContext(ctx, db.Writer, "migrations"); err != nil {
		return fmt.Errorf("apply migrations: %w", err)
	}
	return nil
}

func (db *DB) SchemaVersion(ctx context.Context) (int64, error) {
	var version int64
	err := db.Reader.QueryRowContext(ctx,
		`SELECT version_id FROM goose_db_version WHERE is_applied = 1 ORDER BY id DESC LIMIT 1`,
	).Scan(&version)
	if err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return 0, nil
		}
		return 0, fmt.Errorf("read schema version: %w", err)
	}
	return version, nil
}

func (db *DB) JournalMode(ctx context.Context) (string, error) {
	var mode string
	if err := db.Reader.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&mode); err != nil {
		return "", fmt.Errorf("read journal mode: %w", err)
	}
	return mode, nil
}

func (db *DB) Ready(ctx context.Context) error {
	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()
	var one int
	if err := db.Reader.QueryRowContext(ctx, "SELECT 1").Scan(&one); err != nil {
		return fmt.Errorf("database not ready: %w", err)
	}
	return nil
}

func (db *DB) Optimize(ctx context.Context) error {
	if _, err := db.Writer.ExecContext(ctx, "PRAGMA optimize"); err != nil {
		return fmt.Errorf("pragma optimize: %w", err)
	}
	return nil
}

func (db *DB) RunMaintenance(ctx context.Context, logger *slog.Logger, interval time.Duration) func() {
	ctx, cancel := context.WithCancel(ctx)
	done := make(chan struct{})
	go func() {
		defer close(done)
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if err := db.Optimize(ctx); err != nil {
					logger.WarnContext(ctx, "database maintenance failed", slog.Any("error", err))
				}
			}
		}
	}()
	return func() {
		cancel()
		<-done
	}
}

func (db *DB) Close() error {
	var errs []error
	if db.Reader != nil {
		if err := db.Reader.Close(); err != nil {
			errs = append(errs, err)
		}
	}
	if db.Writer != nil {
		if err := db.Writer.Close(); err != nil {
			errs = append(errs, err)
		}
	}
	if len(errs) > 0 {
		return fmt.Errorf("close database: %w", errors.Join(errs...))
	}
	return nil
}

func closeQuietly(closer io.Closer) {
	if err := closer.Close(); err != nil {
		slog.Debug("close sql handle", slog.Any("error", err))
	}
}

func EnsureDir(path string) error {
	if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
		return fmt.Errorf("create data directory: %w", err)
	}
	return nil
}