codegirl007
f61b4fc7ab
Require email at registration (editable on profile), add a Resend mailer with idempotent first-answer sends, and keep answer saves independent of delivery.
116 lines
3.3 KiB
Go
116 lines
3.3 KiB
Go
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(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 := exec.ExecContext(ctx, stmt); err != nil {
|
|
return fmt.Errorf("add column %s: %w", col, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// migrateUserEmail adds email and a partial unique index on lower(email).
|
|
func migrateUserEmail(ctx context.Context, exec execContext) error {
|
|
if _, err := exec.ExecContext(ctx, `ALTER TABLE users ADD COLUMN IF NOT EXISTS email TEXT NOT NULL DEFAULT ''`); err != nil {
|
|
return fmt.Errorf("add column email: %w", err)
|
|
}
|
|
if _, err := exec.ExecContext(ctx, `
|
|
CREATE UNIQUE INDEX IF NOT EXISTS users_email_lower_uidx
|
|
ON users (lower(email))
|
|
WHERE email <> ''`); err != nil {
|
|
return fmt.Errorf("users_email_lower_uidx: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
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 a session-level advisory lock
|
|
// held for the entire process (check versions → apply → record).
|
|
func applyMigrations(db *sql.DB, schemaSQL string) error {
|
|
ctx := context.Background()
|
|
conn, err := db.Conn(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer conn.Close()
|
|
|
|
if _, err := conn.ExecContext(ctx, `SELECT pg_advisory_lock($1)`, migrateLockKey); err != nil {
|
|
return fmt.Errorf("migrate lock: %w", err)
|
|
}
|
|
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)
|
|
}
|
|
|
|
applied, err := appliedVersions(ctx, conn)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
migrations := []struct {
|
|
version string
|
|
run func(context.Context, execContext) error
|
|
}{
|
|
{"001_schema", func(ctx context.Context, exec execContext) error {
|
|
return applySchema(ctx, exec, schemaSQL)
|
|
}},
|
|
{"002_user_profile_columns", migrateUserProfileColumns},
|
|
{"003_user_email", migrateUserEmail},
|
|
}
|
|
for _, m := range migrations {
|
|
if applied[m.version] {
|
|
continue
|
|
}
|
|
log.Printf("migrate: applying %s", m.version)
|
|
if err := m.run(ctx, conn); err != nil {
|
|
return fmt.Errorf("migrate %s: %w", m.version, err)
|
|
}
|
|
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(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
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]bool{}
|
|
for rows.Next() {
|
|
var v string
|
|
if err := rows.Scan(&v); err != nil {
|
|
return nil, err
|
|
}
|
|
out[v] = true
|
|
}
|
|
return out, rows.Err()
|
|
}
|