FlowReader/internal/database/database.go

// Package database provides database connection and migration utilities.
package database

import (
	"context"
	"fmt"
	"log"

	"os"
	"path/filepath"
	"sort"
	"strings"

	"github.com/jackc/pgx/v5/pgxpool"
)

// Connect creates a new connection pool to the database.
func Connect(ctx context.Context, databaseURL string) (*pgxpool.Pool, error) {
	config, err := pgxpool.ParseConfig(databaseURL)
	if err != nil {
		return nil, fmt.Errorf("parsing database URL: %w", err)
	}

	// Connection pool settings (optimized for low memory)
	config.MaxConns = 10
	config.MinConns = 2

	pool, err := pgxpool.NewWithConfig(ctx, config)
	if err != nil {
		return nil, fmt.Errorf("creating connection pool: %w", err)
	}

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

	log.Println("Database connection established")
	return pool, nil
}

// RunMigrations executes pending database migrations.
func RunMigrations(ctx context.Context, pool *pgxpool.Pool) error {
	// 1. Create schema_migrations table if it doesn't exist
	_, err := pool.Exec(ctx, `
		CREATE TABLE IF NOT EXISTS schema_migrations (
			version VARCHAR(255) PRIMARY KEY,
			applied_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
		)
	`)
	if err != nil {
		return fmt.Errorf("creating schema_migrations table: %w", err)
	}

	// 2. Get applied migrations
	rows, err := pool.Query(ctx, "SELECT version FROM schema_migrations")
	if err != nil {
		return fmt.Errorf("querying applied migrations: %w", err)
	}
	defer rows.Close()

	applied := make(map[string]bool)
	for rows.Next() {
		var version string
		if err := rows.Scan(&version); err != nil {
			return fmt.Errorf("scanning migration version: %w", err)
		}
		applied[version] = true
	}

	// 3. Special case for legacy system:
	// If 'users' table exists but 'schema_migrations' is empty,
	// mark all migrations up to 005 as applied to avoid duplicate creation errors.
	if len(applied) == 0 {
		var usersExists bool
		pool.QueryRow(ctx, "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = 'users')").Scan(&usersExists)
		if usersExists {
			log.Println("Legacy database detected, bootstrapping migration tracking...")
			// We assume 001 to 005 are already there if users table exists in the old system
			legacyVersions := []string{"001_create_users.up.sql", "002_create_sessions.up.sql", "003_create_feeds.up.sql", "004_add_user_roles.up.sql", "005_search_articles.up.sql"}
			for _, v := range legacyVersions {
				_, err := pool.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", v)
				if err != nil {
					log.Printf("Warning: failed to bootstrap version %s: %v", v, err)
				}
				applied[v] = true
			}
		}
	}

	// 4. Read migration files
	entries, err := os.ReadDir("./migrations")
	if err != nil {
		return fmt.Errorf("reading migrations dir: %w", err)
	}

	var upFiles []string
	for _, entry := range entries {
		if !entry.IsDir() && strings.HasSuffix(entry.Name(), ".up.sql") {
			upFiles = append(upFiles, entry.Name())
		}
	}
	sort.Strings(upFiles)

	// 5. Apply pending migrations
	for _, file := range upFiles {
		if applied[file] {
			continue
		}

		log.Printf("Applying migration: %s", file)
		content, err := os.ReadFile(filepath.Join("./migrations", file))
		if err != nil {
			return fmt.Errorf("reading migration %s: %w", file, err)
		}

		// Execute in transaction
		tx, err := pool.Begin(ctx)
		if err != nil {
			return fmt.Errorf("starting transaction for %s: %w", file, err)
		}
		defer tx.Rollback(ctx)

		if _, err := tx.Exec(ctx, string(content)); err != nil {
			return fmt.Errorf("executing migration %s: %w", file, err)
		}

		if _, err := tx.Exec(ctx, "INSERT INTO schema_migrations (version) VALUES ($1)", file); err != nil {
			return fmt.Errorf("recording migration %s: %w", file, err)
		}

		if err := tx.Commit(ctx); err != nil {
			return fmt.Errorf("committing migration %s: %w", file, err)
		}
	}

	log.Println("Database schema is up to date")
	return nil
}