plumber/internal/store/postgres.go

package store

import (
	"database/sql"
	"fmt"
	"net/url"
	"strings"
	"time"

	_ "github.com/jackc/pgx/v5/stdlib"
)

// applySchema runs semicolon-separated DDL statements.
func applySchema(db *sql.DB, schema string) error {
	for _, stmt := range strings.Split(schema, ";") {
		stmt = strings.TrimSpace(stmt)
		if stmt == "" {
			continue
		}
		if _, err := db.Exec(stmt); err != nil {
			return fmt.Errorf("%w: %s", err, stmt)
		}
	}
	return nil
}

// postgresDSN normalizes DATABASE_URL for pgx (sslmode default, strip unsupported params).
func postgresDSN(raw string) (string, error) {
	u, err := url.Parse(raw)
	if err != nil {
		return "", fmt.Errorf("DATABASE_URL: %w", err)
	}
	switch u.Scheme {
	case "postgres", "postgresql":
	default:
		return "", fmt.Errorf("DATABASE_URL must be a postgres URL")
	}
	q := u.Query()
	if strings.EqualFold(q.Get("sslrootcert"), "system") {
		q.Del("sslrootcert")
	}
	q.Del("sslnegotiation")
	if q.Get("sslmode") == "" {
		q.Set("sslmode", "verify-full")
	}
	u.RawQuery = q.Encode()
	return u.String(), nil
}

// OpenPostgres connects to Postgres, applies schema/migrations, and starts session cleanup.
func OpenPostgres(databaseURL, schema string) (*sql.DB, *SessionStore, error) {
	dsn, err := postgresDSN(databaseURL)
	if err != nil {
		return nil, nil, err
	}
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		return nil, nil, err
	}
	db.SetMaxOpenConns(20)
	db.SetMaxIdleConns(5)
	if err := db.Ping(); err != nil {
		_ = db.Close()
		return nil, nil, fmt.Errorf("postgres ping: %w", err)
	}
	if err := applySchema(db, schema); err != nil {
		_ = db.Close()
		return nil, nil, fmt.Errorf("apply schema: %w", err)
	}
	if err := migrateUserProfileColumns(db); err != nil {
		_ = db.Close()
		return nil, nil, fmt.Errorf("migrate profile columns: %w", err)
	}
	sessions := NewSessionStore(db, 5*time.Minute)
	return db, sessions, nil
}