monorepo/cloud/maplepress-backend/pkg/storage/database/migration.go

package database

import (
	"fmt"

	"github.com/gocql/gocql"
	"github.com/golang-migrate/migrate/v4"
	_ "github.com/golang-migrate/migrate/v4/database/cassandra"
	_ "github.com/golang-migrate/migrate/v4/source/file"
	"go.uber.org/zap"

	"codeberg.org/mapleopentech/monorepo/cloud/maplepress-backend/config"
)

// silentGocqlLogger filters out noisy "invalid peer" warnings from gocql
type silentGocqlLogger struct{}

func (l *silentGocqlLogger) Print(v ...interface{}) {
	// Silently discard all gocql logs including "invalid peer" warnings
}

func (l *silentGocqlLogger) Printf(format string, v ...interface{}) {
	// Silently discard all gocql logs including "invalid peer" warnings
}

func (l *silentGocqlLogger) Println(v ...interface{}) {
	// Silently discard all gocql logs including "invalid peer" warnings
}

// Migrator handles database schema migrations
// This encapsulates all migration logic and makes it testable
type Migrator struct {
	config *config.Config
	logger *zap.Logger
}

// NewMigrator creates a new migration manager
func NewMigrator(cfg *config.Config, logger *zap.Logger) *Migrator {
	if logger == nil {
		// Create a no-op logger if none provided (for backward compatibility)
		logger = zap.NewNop()
	}
	return &Migrator{
		config: cfg,
		logger: logger,
	}
}

// Up runs all pending migrations with dirty state recovery
func (m *Migrator) Up() error {
	// Ensure keyspace exists before running migrations
	m.logger.Debug("Ensuring keyspace exists...")
	if err := m.ensureKeyspaceExists(); err != nil {
		return fmt.Errorf("failed to ensure keyspace exists: %w", err)
	}

	m.logger.Debug("Creating migrator...")
	migrateInstance, err := m.createMigrate()
	if err != nil {
		return fmt.Errorf("failed to create migrator: %w", err)
	}
	defer migrateInstance.Close()

	m.logger.Debug("Checking migration version...")
	version, dirty, err := migrateInstance.Version()
	if err != nil && err != migrate.ErrNilVersion {
		return fmt.Errorf("failed to get migration version: %w", err)
	}

	if dirty {
		m.logger.Warn("Database is in dirty state, attempting to force clean state",
			zap.Uint("version", uint(version)))
		if err := migrateInstance.Force(int(version)); err != nil {
			return fmt.Errorf("failed to force clean migration state: %w", err)
		}
	}

	// Run migrations
	if err := migrateInstance.Up(); err != nil && err != migrate.ErrNoChange {
		return fmt.Errorf("failed to run migrations: %w", err)
	}

	// Get final version
	finalVersion, _, err := migrateInstance.Version()
	if err != nil && err != migrate.ErrNilVersion {
		m.logger.Warn("Could not get final migration version", zap.Error(err))
	} else if err != migrate.ErrNilVersion {
		m.logger.Debug("Database migrations completed successfully",
			zap.Uint("version", uint(finalVersion)))
	} else {
		m.logger.Debug("Database migrations completed successfully (no migrations applied)")
	}

	return nil
}

// Down rolls back the last migration
// Useful for development and rollback scenarios
func (m *Migrator) Down() error {
	migrateInstance, err := m.createMigrate()
	if err != nil {
		return fmt.Errorf("failed to create migrator: %w", err)
	}
	defer migrateInstance.Close()

	if err := migrateInstance.Steps(-1); err != nil {
		return fmt.Errorf("failed to rollback migration: %w", err)
	}

	return nil
}

// Version returns the current migration version
func (m *Migrator) Version() (uint, bool, error) {
	migrateInstance, err := m.createMigrate()
	if err != nil {
		return 0, false, fmt.Errorf("failed to create migrator: %w", err)
	}
	defer migrateInstance.Close()

	return migrateInstance.Version()
}

// ForceVersion forces the migration version (useful for fixing dirty states)
func (m *Migrator) ForceVersion(version int) error {
	migrateInstance, err := m.createMigrate()
	if err != nil {
		return fmt.Errorf("failed to create migrator: %w", err)
	}
	defer migrateInstance.Close()

	if err := migrateInstance.Force(version); err != nil {
		return fmt.Errorf("failed to force version %d: %w", version, err)
	}

	m.logger.Info("Successfully forced migration version", zap.Int("version", version))
	return nil
}

// createMigrate creates a migrate instance with proper configuration
func (m *Migrator) createMigrate() (*migrate.Migrate, error) {
	// Set global gocql logger to suppress "invalid peer" warnings
	// This affects the internal gocql connections used by golang-migrate
	gocql.Logger = &silentGocqlLogger{}

	// Build Cassandra connection string
	// Format: cassandra://host:port/keyspace?consistency=level
	databaseURL := fmt.Sprintf("cassandra://%s/%s?consistency=%s",
		m.config.Database.Hosts[0], // Use first host for migrations
		m.config.Database.Keyspace,
		m.config.Database.Consistency,
	)

	// Create migrate instance
	migrateInstance, err := migrate.New(m.config.Database.MigrationsPath, databaseURL)
	if err != nil {
		return nil, fmt.Errorf("failed to initialize migrate: %w", err)
	}

	return migrateInstance, nil
}

// ensureKeyspaceExists creates the keyspace if it doesn't exist
// This must be done before running migrations since golang-migrate requires the keyspace to exist
func (m *Migrator) ensureKeyspaceExists() error {
	// Create cluster configuration without keyspace
	cluster := gocql.NewCluster(m.config.Database.Hosts...)
	cluster.Port = 9042
	cluster.Consistency = gocql.Quorum
	cluster.ProtoVersion = 4

	// Suppress noisy "invalid peer" warnings from gocql
	// Use a minimal logger that discards these harmless Docker networking warnings
	cluster.Logger = &silentGocqlLogger{}

	// Create session to system keyspace
	session, err := cluster.CreateSession()
	if err != nil {
		return fmt.Errorf("failed to connect to Cassandra: %w", err)
	}
	defer session.Close()

	// Create keyspace if it doesn't exist
	replicationFactor := m.config.Database.Replication
	createKeyspaceQuery := fmt.Sprintf(`
		CREATE KEYSPACE IF NOT EXISTS %s
		WITH replication = {'class': 'NetworkTopologyStrategy', 'datacenter1': %d}
		AND durable_writes = true
	`, m.config.Database.Keyspace, replicationFactor)

	m.logger.Debug("Creating keyspace if it doesn't exist",
		zap.String("keyspace", m.config.Database.Keyspace))
	if err := session.Query(createKeyspaceQuery).Exec(); err != nil {
		return fmt.Errorf("failed to create keyspace: %w", err)
	}

	m.logger.Debug("Keyspace is ready", zap.String("keyspace", m.config.Database.Keyspace))
	return nil
}