Janady13/missionmatch / cmd/migrate-all-tenants/main.go

1// Command migrate-all-tenants applies the migration set to the control plane database AND to every active 2// tenant database in the database per tenant model. It reads the tenant directory in the control plane to 3// discover the tenant databases, then for each target applies BOTH the public schema migrations and the 4// segregated eeo migrations (two separate goose version tables), so a tenant database can never drift one 5// schema relative to the other. 6// 7// It is a migrator role operator tool (it needs BYPASSRLS to write the directory bookkeeping and to run 8// DDL). It is idempotent and resumable: goose records applied versions, so a re run is a no op on a 9// healthy database. The failure surface is hard: it records which database errored or is behind, keeps 10// going to the rest, and exits non zero if ANY database failed or is still behind head, so a partial fan 11// out can never report success. 12// 13// Two modes: 14// 15// migrate-all-tenants apply pending migrations to the control plane and every active tenant 16// migrate-all-tenants -check dry run: report each database's version and whether it is behind, NO writes 17// 18// It uses the per instance goose Provider (NOT the process global goose API, whose SetBaseFS and 19// SetTableName state would race and cross contaminate the public versus eeo table name across N databases 20// and under concurrency, the pressly/goose multi database guidance), and a Postgres session locker so two 21// operators (or a concurrent provision-tenant) cannot migrate the same database at once. 22package main 23 24import ( 25 "context" 26 "database/sql" 27 "flag" 28 "fmt" 29 "io/fs" 30 "log" 31 "os" 32 "sort" 33 "sync" 34 "time" 35 36 "github.com/google/uuid" 37 "github.com/jackc/pgx/v5" 38 "github.com/jackc/pgx/v5/pgxpool" 39 _ "github.com/jackc/pgx/v5/stdlib" 40 goose "github.com/pressly/goose/v3" 41 goosedb "github.com/pressly/goose/v3/database" 42 gooselock "github.com/pressly/goose/v3/lock" 43 "golang.org/x/sync/errgroup" 44 45 dbembed "github.com/missionworkforce/missionmatch/db" 46 "github.com/missionworkforce/missionmatch/internal/platform/database" 47 "github.com/missionworkforce/missionmatch/internal/platform/directory" 48) 49 50type target struct { 51 name string // human label (control plane, or the tenant database name) 52 dsn string // migrator DSN to this database 53 org *uuid.UUID // nil for the control plane; set for a tenant (to record bookkeeping) 54} 55 56type result struct { 57 name string 58 pubVersion int64 59 eeoVersion int64 60 pending bool // dry run: any pending migration 61 appliedPub int 62 appliedEEO int 63 err error 64} 65 66func main() { 67 controlDSN := flag.String("control-dsn", os.Getenv("MM_MIGRATOR_DATABASE_URL"), "migrator DSN to the control plane database (defaults to MM_MIGRATOR_DATABASE_URL)") 68 check := flag.Bool("check", false, "dry run: report each database's version and whether it is behind, make no changes") 69 concurrency := flag.Int("concurrency", 1, "migrate up to N databases at once (distinct databases only)") 70 flag.Parse() 71 72 if *controlDSN == "" { 73 log.Fatal("migrate-all-tenants: -control-dsn (or MM_MIGRATOR_DATABASE_URL) is required") 74 } 75 if *concurrency < 1 { 76 *concurrency = 1 77 } 78 ctx := context.Background() 79 80 // The control plane pool: it must be the BYPASSRLS migrator so it can read the directory and run DDL. 81 pool, err := pgxpool.New(ctx, *controlDSN) 82 if err != nil { 83 log.Fatalf("migrate-all-tenants: control plane pool: %v", err) 84 } 85 defer pool.Close() 86 if err := database.AssertMigratorRole(ctx, pool); err != nil { 87 log.Fatalf("migrate-all-tenants: %v", err) 88 } 89 dir := directory.New(pool) 90 91 // Targets: the control plane first (the foundation), then every active tenant database. A tenant DSN 92 // reuses the migrator credential and swaps only the database name (the migrator is a cluster role). 93 targets := []target{{name: "control plane", dsn: *controlDSN, org: nil}} 94 tenants, err := dir.List(ctx, directory.StatusActive) 95 if err != nil { 96 log.Fatalf("migrate-all-tenants: list tenants: %v", err) 97 } 98 for _, t := range tenants { 99 org := t.OrganizationID 100 targets = append(targets, target{name: t.DBName, dsn: swapDB(*controlDSN, t.DBName), org: &org}) 101 } 102 log.Printf("migrate-all-tenants: %d targets (control plane + %d active tenants), check=%v concurrency=%d", len(targets), len(tenants), *check, *concurrency) 103 104 results := make([]result, len(targets)) 105 var g errgroup.Group 106 g.SetLimit(*concurrency) 107 var mu sync.Mutex // guards the directory bookkeeping writes 108 for i := range targets { 109 i := i 110 g.Go(func() error { 111 r := migrateOne(ctx, targets[i], *check) 112 results[i] = r 113 // Record the applied version in the directory for a tenant that advanced (apply mode only). 114 if !*check && r.err == nil && targets[i].org != nil { 115 mu.Lock() 116 if e := dir.SetMigrated(ctx, *targets[i].org, r.pubVersion); e != nil { 117 log.Printf("migrate-all-tenants: WARNING could not record migration bookkeeping for %s: %v", targets[i].name, e) 118 } 119 mu.Unlock() 120 } 121 return nil // never abort the group on one database; the summary and exit code carry the verdict 122 }) 123 } 124 _ = g.Wait() 125 126 // Summary and a hard exit code. 127 failed, behind := report(results, *check) 128 if failed > 0 || (*check && behind > 0) { 129 os.Exit(1) 130 } 131} 132 133// migrateOne applies (or, in check mode, inspects) the two migration sets on one database using per 134// instance goose Providers and a session lock. 135func migrateOne(ctx context.Context, t target, check bool) result { 136 r := result{name: t.name} 137 sqlDB, err := sql.Open("pgx", t.dsn) 138 if err != nil { 139 r.err = fmt.Errorf("open: %w", err) 140 return r 141 } 142 defer func() { _ = sqlDB.Close() }() 143 sqlDB.SetMaxOpenConns(5) 144 sqlDB.SetConnMaxLifetime(5 * time.Minute) 145 146 locker, err := gooselock.NewPostgresSessionLocker() 147 if err != nil { 148 r.err = fmt.Errorf("locker: %w", err) 149 return r 150 } 151 152 pubFS, err := fs.Sub(dbembed.Migrations, "migrations") 153 if err != nil { 154 r.err = err 155 return r 156 } 157 pub, err := goose.NewProvider(goose.DialectPostgres, sqlDB, pubFS, goose.WithSessionLocker(locker)) 158 if err != nil { 159 r.err = fmt.Errorf("public provider: %w", err) 160 return r 161 } 162 defer func() { _ = pub.Close() }() 163 164 eeoStore, err := goosedb.NewStore(goosedb.DialectPostgres, "goose_db_version_eeo") 165 if err != nil { 166 r.err = fmt.Errorf("eeo store: %w", err) 167 return r 168 } 169 eeoFS, err := fs.Sub(dbembed.EEOMigrations, "migrations_eeo") 170 if err != nil { 171 r.err = err 172 return r 173 } 174 // Empty dialect because the table name is carried by the custom store. 175 eeo, err := goose.NewProvider("", sqlDB, eeoFS, goose.WithStore(eeoStore), goose.WithSessionLocker(locker)) 176 if err != nil { 177 r.err = fmt.Errorf("eeo provider: %w", err) 178 return r 179 } 180 defer func() { _ = eeo.Close() }() 181 182 if check { 183 pubPending, e1 := pub.HasPending(ctx) 184 eeoPending, e2 := eeo.HasPending(ctx) 185 if e1 != nil || e2 != nil { 186 r.err = firstErr(e1, e2) 187 return r 188 } 189 r.pending = pubPending || eeoPending 190 r.pubVersion, _ = pub.GetDBVersion(ctx) 191 r.eeoVersion, _ = eeo.GetDBVersion(ctx) 192 return r 193 } 194 195 // Apply public then eeo, so the public foundation exists before the segregated eeo schema. 196 rp, err := pub.Up(ctx) 197 if err != nil { 198 r.err = fmt.Errorf("public migrations: %w", err) 199 return r 200 } 201 re, err := eeo.Up(ctx) 202 if err != nil { 203 r.err = fmt.Errorf("eeo migrations: %w", err) 204 return r 205 } 206 r.appliedPub, r.appliedEEO = len(rp), len(re) 207 r.pubVersion, _ = pub.GetDBVersion(ctx) 208 r.eeoVersion, _ = eeo.GetDBVersion(ctx) 209 return r 210} 211 212// report prints a per database table and returns how many failed and how many are behind. 213func report(results []result, check bool) (failed, behind int) { 214 sorted := append([]result(nil), results...) 215 sort.SliceStable(sorted, func(i, j int) bool { return sorted[i].name < sorted[j].name }) 216 log.Printf("migrate-all-tenants: results") 217 for _, r := range sorted { 218 switch { 219 case r.err != nil: 220 failed++ 221 log.Printf(" FAILED %-28s public=v%d eeo=v%d %v", r.name, r.pubVersion, r.eeoVersion, r.err) 222 case check && r.pending: 223 behind++ 224 log.Printf(" BEHIND %-28s public=v%d eeo=v%d (has pending migrations)", r.name, r.pubVersion, r.eeoVersion) 225 case check: 226 log.Printf(" OK %-28s public=v%d eeo=v%d (at head)", r.name, r.pubVersion, r.eeoVersion) 227 default: 228 log.Printf(" APPLIED %-28s public=v%d (+%d) eeo=v%d (+%d)", r.name, r.pubVersion, r.appliedPub, r.eeoVersion, r.appliedEEO) 229 } 230 } 231 if failed > 0 { 232 log.Printf("migrate-all-tenants: %d database(s) FAILED", failed) 233 } 234 if check && behind > 0 { 235 log.Printf("migrate-all-tenants: %d database(s) BEHIND head", behind) 236 } 237 if failed == 0 && behind == 0 { 238 log.Printf("migrate-all-tenants: all databases OK") 239 } 240 return failed, behind 241} 242 243func firstErr(errs ...error) error { 244 for _, e := range errs { 245 if e != nil { 246 return e 247 } 248 } 249 return nil 250} 251 252// swapDB reuses a migrator DSN and swaps only the database name, so a tenant database is reached with the 253// same migrator credential (the migrator is a cluster role). Mirrors the helper in cmd/provision-tenant. 254func swapDB(dsn, name string) string { 255 cfg, err := pgx.ParseConfig(dsn) 256 if err != nil { 257 return dsn 258 } 259 out := fmt.Sprintf("host=%s port=%d user=%s dbname=%s", cfg.Host, cfg.Port, cfg.User, name) 260 if cfg.Password != "" { 261 out += " password=" + cfg.Password 262 } 263 if cfg.TLSConfig == nil { 264 out += " sslmode=disable" 265 } 266 return out 267} 268