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 }