Refactor database implementation to use PostgreSQL instead of MongoDB (#289)
Fixes #228 ## Motivation and Context See #19 ## How Has This Been Tested? - Run locally - Run in docker compose - Run infra in minikube - Unit and integration tests all passing :) ## Breaking Changes Yes: users will not be able to access any data they used to have in custom mongodb. We are accepting this breaking change, given the registry is in experimental development status. ## Types of changes <!-- What types of changes does your code introduce? Put an `x` in all the boxes that apply: --> - [ ] Bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [x] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update ## Checklist <!-- Go over all the following points, and put an `x` in all the boxes that apply. --> - [x] I have read the [MCP Documentation](https://modelcontextprotocol.io) - [x] My code follows the repository's style guidelines - [x] New and existing tests pass locally - [x] I have added appropriate error handling - [x] I have added or updated documentation as needed ## Additional context I have not implemented connection pooling support for simplicity, and because cloudnative-pg already provides built-in connection pooling with pgBouncer so if we want it later we can switch to this endpoint easily.
This commit is contained in:
adam jones
GitHub
parent
ee910ca286
commit
ebaf4c9c99
1 changed files with 846 additions and 564 deletions
192
internal/database/migrate.go
Normal file
192
internal/database/migrate.go
Normal file
|
|
@ -0,0 +1,192 @@
|
|||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"embed"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
//go:embed migrations/*.sql
|
||||
var migrationFiles embed.FS
|
||||
|
||||
// Migration represents a database migration
|
||||
type Migration struct {
|
||||
Version int
|
||||
Name string
|
||||
SQL string
|
||||
}
|
||||
|
||||
// Migrator handles database migrations
|
||||
type Migrator struct {
|
||||
conn *pgx.Conn
|
||||
}
|
||||
|
||||
// NewMigrator creates a new migrator instance
|
||||
func NewMigrator(conn *pgx.Conn) *Migrator {
|
||||
return &Migrator{conn: conn}
|
||||
}
|
||||
|
||||
// ensureMigrationsTable creates the migrations tracking table if it doesn't exist
|
||||
func (m *Migrator) ensureMigrationsTable(ctx context.Context) error {
|
||||
query := `
|
||||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||
version INTEGER PRIMARY KEY,
|
||||
name VARCHAR(255) NOT NULL,
|
||||
applied_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
|
||||
)
|
||||
`
|
||||
_, err := m.conn.Exec(ctx, query)
|
||||
return err
|
||||
}
|
||||
|
||||
// getAppliedMigrations returns a map of already applied migration versions
|
||||
func (m *Migrator) getAppliedMigrations(ctx context.Context) (map[int]bool, error) {
|
||||
query := "SELECT version FROM schema_migrations ORDER BY version"
|
||||
rows, err := m.conn.Query(ctx, query)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query applied migrations: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
applied := make(map[int]bool)
|
||||
for rows.Next() {
|
||||
var version int
|
||||
if err := rows.Scan(&version); err != nil {
|
||||
return nil, fmt.Errorf("failed to scan migration version: %w", err)
|
||||
}
|
||||
applied[version] = true
|
||||
}
|
||||
|
||||
return applied, rows.Err()
|
||||
}
|
||||
|
||||
// loadMigrations loads all migration files from the embedded filesystem
|
||||
func (m *Migrator) loadMigrations() ([]Migration, error) {
|
||||
entries, err := migrationFiles.ReadDir("migrations")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read migrations directory: %w", err)
|
||||
}
|
||||
|
||||
var migrations []Migration
|
||||
for _, entry := range entries {
|
||||
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".sql") {
|
||||
continue
|
||||
}
|
||||
|
||||
// Parse version from filename (e.g., "001_initial_schema.sql" -> version 1)
|
||||
name := entry.Name()
|
||||
parts := strings.SplitN(name, "_", 2)
|
||||
if len(parts) != 2 {
|
||||
log.Printf("Skipping migration file with invalid name format: %s", name)
|
||||
continue
|
||||
}
|
||||
|
||||
version, err := strconv.Atoi(parts[0])
|
||||
if err != nil {
|
||||
log.Printf("Skipping migration file with invalid version: %s", name)
|
||||
continue
|
||||
}
|
||||
|
||||
// Read the migration SQL
|
||||
content, err := migrationFiles.ReadFile(filepath.Join("migrations", name))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read migration file %s: %w", name, err)
|
||||
}
|
||||
|
||||
migrations = append(migrations, Migration{
|
||||
Version: version,
|
||||
Name: strings.TrimSuffix(name, ".sql"),
|
||||
SQL: string(content),
|
||||
})
|
||||
}
|
||||
|
||||
// Sort migrations by version
|
||||
sort.Slice(migrations, func(i, j int) bool {
|
||||
return migrations[i].Version < migrations[j].Version
|
||||
})
|
||||
|
||||
return migrations, nil
|
||||
}
|
||||
|
||||
// Migrate runs all pending migrations
|
||||
func (m *Migrator) Migrate(ctx context.Context) error {
|
||||
// Ensure the migrations table exists
|
||||
if err := m.ensureMigrationsTable(ctx); err != nil {
|
||||
return fmt.Errorf("failed to create migrations table: %w", err)
|
||||
}
|
||||
|
||||
// Get applied migrations
|
||||
applied, err := m.getAppliedMigrations(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get applied migrations: %w", err)
|
||||
}
|
||||
|
||||
// Load all migration files
|
||||
migrations, err := m.loadMigrations()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to load migrations: %w", err)
|
||||
}
|
||||
|
||||
// Find pending migrations
|
||||
var pending []Migration
|
||||
for _, migration := range migrations {
|
||||
if !applied[migration.Version] {
|
||||
pending = append(pending, migration)
|
||||
}
|
||||
}
|
||||
|
||||
if len(pending) == 0 {
|
||||
log.Println("No pending migrations")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Printf("Applying %d pending migrations", len(pending))
|
||||
|
||||
// Apply each pending migration in a transaction
|
||||
for _, migration := range pending {
|
||||
if err := m.applyMigration(ctx, migration); err != nil {
|
||||
return fmt.Errorf("failed to apply migration %d (%s): %w", migration.Version, migration.Name, err)
|
||||
}
|
||||
log.Printf("Applied migration %d: %s", migration.Version, migration.Name)
|
||||
}
|
||||
|
||||
log.Println("All migrations applied successfully")
|
||||
return nil
|
||||
}
|
||||
|
||||
// applyMigration applies a single migration in a transaction
|
||||
func (m *Migrator) applyMigration(ctx context.Context, migration Migration) error {
|
||||
tx, err := m.conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to begin transaction: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
if err := tx.Rollback(ctx); err != nil && !errors.Is(err, pgx.ErrTxClosed) {
|
||||
log.Printf("Failed to rollback migration transaction: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
// Execute the migration SQL
|
||||
_, err = tx.Exec(ctx, migration.SQL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to execute migration SQL: %w", err)
|
||||
}
|
||||
|
||||
// Record the migration as applied
|
||||
_, err = tx.Exec(ctx,
|
||||
"INSERT INTO schema_migrations (version, name) VALUES ($1, $2)",
|
||||
migration.Version, migration.Name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to record migration: %w", err)
|
||||
}
|
||||
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
Loading…
Reference in a new issue