package database import ( "context" "embed" "errors" "fmt" "log" "path" "sort" "strconv" "strings" "github.com/jackc/pgx/v5" ) //go:embed migrations/*.sql var migrationFiles embed.FS // Migration represents a database migration type Migration struct { Version int Name string SQL string } // Migrator handles database migrations type Migrator struct { conn *pgx.Conn } // NewMigrator creates a new migrator instance func NewMigrator(conn *pgx.Conn) *Migrator { return &Migrator{conn: conn} } // ensureMigrationsTable creates the migrations tracking table if it doesn't exist func (m *Migrator) ensureMigrationsTable(ctx context.Context) error { query := ` CREATE TABLE IF NOT EXISTS schema_migrations ( version INTEGER PRIMARY KEY, name VARCHAR(255) NOT NULL, applied_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ) ` _, err := m.conn.Exec(ctx, query) return err } // getAppliedMigrations returns a map of already applied migration versions func (m *Migrator) getAppliedMigrations(ctx context.Context) (map[int]bool, error) { query := "SELECT version FROM schema_migrations ORDER BY version" rows, err := m.conn.Query(ctx, query) if err != nil { return nil, fmt.Errorf("failed to query applied migrations: %w", err) } defer rows.Close() applied := make(map[int]bool) for rows.Next() { var version int if err := rows.Scan(&version); err != nil { return nil, fmt.Errorf("failed to scan migration version: %w", err) } applied[version] = true } return applied, rows.Err() } // loadMigrations loads all migration files from the embedded filesystem func (m *Migrator) loadMigrations() ([]Migration, error) { entries, err := migrationFiles.ReadDir("migrations") if err != nil { return nil, fmt.Errorf("failed to read migrations directory: %w", err) } var migrations []Migration for _, entry := range entries { if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") { continue } // Parse version from filename (e.g., "001_initial_schema.sql" -> version 1) name := entry.Name() parts := strings.SplitN(name, "_", 2) if len(parts) != 2 { log.Printf("Skipping migration file with invalid name format: %s", name) continue } version, err := strconv.Atoi(parts[0]) if err != nil { log.Printf("Skipping migration file with invalid version: %s", name) continue } // Read the migration SQL content, err := migrationFiles.ReadFile(path.Join("migrations", name)) if err != nil { return nil, fmt.Errorf("failed to read migration file %s: %w", name, err) } migrations = append(migrations, Migration{ Version: version, Name: strings.TrimSuffix(name, ".sql"), SQL: string(content), }) } // Sort migrations by version sort.Slice(migrations, func(i, j int) bool { return migrations[i].Version < migrations[j].Version }) return migrations, nil } // Migrate runs all pending migrations func (m *Migrator) Migrate(ctx context.Context) error { // Ensure the migrations table exists if err := m.ensureMigrationsTable(ctx); err != nil { return fmt.Errorf("failed to create migrations table: %w", err) } // Get applied migrations applied, err := m.getAppliedMigrations(ctx) if err != nil { return fmt.Errorf("failed to get applied migrations: %w", err) } // Load all migration files migrations, err := m.loadMigrations() if err != nil { return fmt.Errorf("failed to load migrations: %w", err) } // Find pending migrations var pending []Migration for _, migration := range migrations { if !applied[migration.Version] { pending = append(pending, migration) } } if len(pending) == 0 { log.Println("No pending migrations") return nil } log.Printf("Applying %d pending migrations", len(pending)) // Apply each pending migration in a transaction for _, migration := range pending { if err := m.applyMigration(ctx, migration); err != nil { return fmt.Errorf("failed to apply migration %d (%s): %w", migration.Version, migration.Name, err) } log.Printf("Applied migration %d: %s", migration.Version, migration.Name) } log.Println("All migrations applied successfully") return nil } // applyMigration applies a single migration in a transaction func (m *Migrator) applyMigration(ctx context.Context, migration Migration) error { tx, err := m.conn.Begin(ctx) if err != nil { return fmt.Errorf("failed to begin transaction: %w", err) } defer func() { if err := tx.Rollback(ctx); err != nil && !errors.Is(err, pgx.ErrTxClosed) { log.Printf("Failed to rollback migration transaction: %v", err) } }() // Execute the migration SQL _, err = tx.Exec(ctx, migration.SQL) if err != nil { return fmt.Errorf("failed to execute migration SQL: %w", err) } // Record the migration as applied _, err = tx.Exec(ctx, "INSERT INTO schema_migrations (version, name) VALUES ($1, $2)", migration.Version, migration.Name) if err != nil { return fmt.Errorf("failed to record migration: %w", err) } return tx.Commit(ctx) }