package database import ( "context" "database/sql" "embed" "errors" "fmt" "log/slog" "os" "path/filepath" "runtime" "strings" "time" _ "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 { writer.Close() return nil, err } if err := writer.PingContext(ctx); err != nil { writer.Close() 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 { db.Close() return nil, err } } reader, err := sql.Open("sqlite3", dsn(opts, false)) if err != nil { db.Close() 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 { reader.Close() db.Close() return nil, err } if err := reader.PingContext(ctx); err != nil { reader.Close() db.Close() 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 EnsureDir(path string) error { if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil { return fmt.Errorf("create data directory: %w", err) } return nil }