Fix migrate lock scope and stop RealIP from bypassing proxy trust.
Hold a session advisory lock for the full migration apply path, and remove Chi RealIP so clientIP can validate the TCP peer before walking X-Forwarded-For.
This commit is contained in:
+32
-19
@@ -1,57 +1,70 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"log"
|
||||
)
|
||||
|
||||
const migrateLockKey int64 = 0x706c756d5f6d6967 // "plum_mig"
|
||||
|
||||
// migrateUserProfileColumns adds avatar_url and state when missing (existing DBs).
|
||||
func migrateUserProfileColumns(db *sql.DB) error {
|
||||
func migrateUserProfileColumns(ctx context.Context, exec execContext) error {
|
||||
cols := []string{"avatar_url", "state"}
|
||||
for _, col := range cols {
|
||||
stmt := fmt.Sprintf(`ALTER TABLE users ADD COLUMN IF NOT EXISTS %s TEXT NOT NULL DEFAULT ''`, col)
|
||||
if _, err := db.Exec(stmt); err != nil {
|
||||
if _, err := exec.ExecContext(ctx, stmt); err != nil {
|
||||
return fmt.Errorf("add column %s: %w", col, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
const migrateLockKey int64 = 0x706c756d5f6d6967 // "plum_mig"
|
||||
type execContext interface {
|
||||
ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
|
||||
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
|
||||
}
|
||||
|
||||
// applyMigrations runs versioned migrations under an advisory lock.
|
||||
// Fresh databases apply schemaSQL as version 001; later versions are incremental.
|
||||
// applyMigrations runs versioned migrations under a session-level advisory lock
|
||||
// held for the entire process (check versions → apply → record).
|
||||
func applyMigrations(db *sql.DB, schemaSQL string) error {
|
||||
tx, err := db.Begin()
|
||||
ctx := context.Background()
|
||||
conn, err := db.Conn(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if _, err := tx.Exec(`SELECT pg_advisory_xact_lock($1)`, migrateLockKey); err != nil {
|
||||
defer conn.Close()
|
||||
|
||||
if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock($1)`, migrateLockKey); err != nil {
|
||||
return fmt.Errorf("migrate lock: %w", err)
|
||||
}
|
||||
if _, err := tx.Exec(`
|
||||
defer func() {
|
||||
if _, unlockErr := conn.ExecContext(ctx, `SELECT pg_advisory_unlock($1)`, migrateLockKey); unlockErr != nil {
|
||||
log.Printf("migrate unlock: %v", unlockErr)
|
||||
}
|
||||
}()
|
||||
|
||||
if _, err := conn.ExecContext(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version TEXT PRIMARY KEY,
|
||||
applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)`); err != nil {
|
||||
return fmt.Errorf("schema_migrations: %w", err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
applied, err := appliedVersions(db)
|
||||
applied, err := appliedVersions(ctx, conn)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
migrations := []struct {
|
||||
version string
|
||||
run func(*sql.DB) error
|
||||
run func(context.Context, execContext) error
|
||||
}{
|
||||
{"001_schema", func(db *sql.DB) error { return applySchema(db, schemaSQL) }},
|
||||
{"001_schema", func(ctx context.Context, exec execContext) error {
|
||||
return applySchema(ctx, exec, schemaSQL)
|
||||
}},
|
||||
{"002_user_profile_columns", migrateUserProfileColumns},
|
||||
}
|
||||
for _, m := range migrations {
|
||||
@@ -59,18 +72,18 @@ CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
continue
|
||||
}
|
||||
log.Printf("migrate: applying %s", m.version)
|
||||
if err := m.run(db); err != nil {
|
||||
if err := m.run(ctx, conn); err != nil {
|
||||
return fmt.Errorf("migrate %s: %w", m.version, err)
|
||||
}
|
||||
if _, err := db.Exec(`INSERT INTO schema_migrations (version) VALUES ($1)`, m.version); err != nil {
|
||||
if _, err := conn.ExecContext(ctx, `INSERT INTO schema_migrations (version) VALUES ($1)`, m.version); err != nil {
|
||||
return fmt.Errorf("record %s: %w", m.version, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func appliedVersions(db *sql.DB) (map[string]bool, error) {
|
||||
rows, err := db.Query(`SELECT version FROM schema_migrations`)
|
||||
func appliedVersions(ctx context.Context, exec execContext) (map[string]bool, error) {
|
||||
rows, err := exec.QueryContext(ctx, `SELECT version FROM schema_migrations`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user