MusicServer/internal/db/database.go

package db

import (
	"context"
	"embed"
	"errors"
	"fmt"
	"net"
	"net/url"
	"time"

	"music-server/internal/logging"

	"github.com/golang-migrate/migrate/v4"
	pgxmigrate "github.com/golang-migrate/migrate/v4/database/pgx/v5"
	"github.com/golang-migrate/migrate/v4/source/iofs"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/jackc/pgx/v5/stdlib"
	"go.uber.org/zap"
)

//go:embed "migrations/*.sql"
var MigrationsFs embed.FS

// Database holds the database connection pool
type Database struct {
	Pool *pgxpool.Pool
}

// ConnString builds a PostgreSQL connection URL.
func ConnString(host, port, user, password, dbname string) string {
	u := url.URL{
		Scheme:   "postgres",
		User:     url.UserPassword(user, password),
		Host:     net.JoinHostPort(host, port),
		Path:     dbname,
		RawQuery: "sslmode=disable",
	}
	return u.String()
}

// NewDatabase creates a new Database instance with connection pool
func NewDatabase(ctx context.Context, connString string) (*Database, error) {
	pool, err := pgxpool.New(ctx, connString)
	if err != nil {
		return nil, fmt.Errorf("unable to connect to database: %w", err)
	}

	if err := pool.Ping(ctx); err != nil {
		pool.Close()
		return nil, fmt.Errorf("database ping failed: %w", err)
	}

	logging.GetLogger().Info("Database connected")
	return &Database{Pool: pool}, nil
}

// Close closes the database connection pool
func (db *Database) Close() {
	if db.Pool != nil {
		logging.GetLogger().Info("Closing database connection")
		db.Pool.Close()
	}
}

// Health checks the health of the database connection by pinging the database.
func (db *Database) Health(ctx context.Context) map[string]string {
	ctx, cancel := context.WithTimeout(ctx, time.Second)
	defer cancel()

	if err := db.Pool.Ping(ctx); err != nil {
		return map[string]string{"status": "down", "error": err.Error()}
	}
	return map[string]string{"status": "up"}
}

// RunMigrations runs all pending database migrations to the latest version.
func (db *Database) RunMigrations() error {
	m, closeFn, err := db.NewMigrator()
	if err != nil {
		return err
	}
	defer closeFn()

	version, _, err := m.Version()
	if err != nil && !errors.Is(err, migrate.ErrNilVersion) {
		logging.GetLogger().Error("Failed to get migration version", zap.String("error", err.Error()))
	}
	logging.GetLogger().Info("Migration version before", zap.Uint("version", version))

	if err := m.Up(); err != nil {
		if !errors.Is(err, migrate.ErrNoChange) {
			return fmt.Errorf("migration failed: %w", err)
		}
		logging.GetLogger().Info("Database already up to date")
		return nil
	}

	versionAfter, _, _ := m.Version()
	logging.GetLogger().Info("Migrated to version", zap.Uint("version", versionAfter))
	return nil
}

// NewMigrator returns a migrator for the embedded migrations using the pool's connection settings.
// The returned function closes the underlying connection.
func (db *Database) NewMigrator() (*migrate.Migrate, func(), error) {
	sqlDB := stdlib.OpenDB(*db.Pool.Config().ConnConfig)

	driver, err := pgxmigrate.WithInstance(sqlDB, &pgxmigrate.Config{})
	if err != nil {
		sqlDB.Close()
		return nil, nil, fmt.Errorf("failed to create migration driver: %w", err)
	}

	files, err := iofs.New(MigrationsFs, "migrations")
	if err != nil {
		sqlDB.Close()
		return nil, nil, fmt.Errorf("failed to read migration files: %w", err)
	}

	m, err := migrate.NewWithInstance("iofs", files, "pgx5", driver)
	if err != nil {
		sqlDB.Close()
		return nil, nil, fmt.Errorf("failed to create migrator: %w", err)
	}
	return m, func() { m.Close() }, nil
}