class ActiveRecord::Migrator

def current_migration
  migrations.detect { |m| m.version == current_version }
end

def current_version
  MigrationContext.new(migrations_paths, SchemaMigration).current_version
end

def current_version
  migrated.max || 0
end

def ddl_transaction(migration)
  if use_transaction?(migration)
    Base.transaction { yield }
  else
    yield
  end
end

def down?
  @direction == :down
end

def execute_migration_in_transaction(migration)
  return if down? && !migrated.include?(migration.version.to_i)
  return if up?   &&  migrated.include?(migration.version.to_i)
  Base.logger.info "Migrating to #{migration.name} (#{migration.version})" if Base.logger
  ddl_transaction(migration) do
    migration.migrate(@direction)
    record_version_state_after_migrating(migration.version)
  end
rescue => e
  msg = +"An error has occurred, "
  msg << "this and " if use_transaction?(migration)
  msg << "all later migrations canceled:\n\n#{e}"
  raise StandardError, msg, e.backtrace
end

def finish
  migrations.index(target) || migrations.size - 1
end

def generate_migrator_advisory_lock_id
  db_name_hash = Zlib.crc32(Base.connection.current_database)
  MIGRATOR_SALT * db_name_hash
end

def initialize(direction, migrations, schema_migration, target_version = nil)
  @direction         = direction
  @target_version    = target_version
  @migrated_versions = nil
  @migrations        = migrations
  @schema_migration  = schema_migration
  validate(@migrations)
  @schema_migration.create_table
  ActiveRecord::InternalMetadata.create_table
end

def invalid_target?
  @target_version && @target_version != 0 && !target
end

def load_migrated
  @migrated_versions = Set.new(@schema_migration.all_versions.map(&:to_i))
end

def migrate
  if use_advisory_lock?
    with_advisory_lock { migrate_without_lock }
  else
    migrate_without_lock
  end
end

def migrate_without_lock
  if invalid_target?
    raise UnknownMigrationVersionError.new(@target_version)
  end
  result = runnable.each(&method(:execute_migration_in_transaction))
  record_environment
  result
end

def migrated
  @migrated_versions || load_migrated
end

def migrations
  down? ? @migrations.reverse : @migrations.sort_by(&:version)
end

def pending_migrations
  already_migrated = migrated
  migrations.reject { |m| already_migrated.include?(m.version) }
end

def ran?(migration)
  migrated.include?(migration.version.to_i)
end

def record_environment
  return if down?
  ActiveRecord::InternalMetadata[:environment] = ActiveRecord::Base.connection.migration_context.current_environment
end

def record_version_state_after_migrating(version)
  if down?
    migrated.delete(version)
    @schema_migration.delete_by(version: version.to_s)
  else
    migrated << version
    @schema_migration.create!(version: version.to_s)
  end
end

def run
  if use_advisory_lock?
    with_advisory_lock { run_without_lock }
  else
    run_without_lock
  end
end

def run_without_lock
  migration = migrations.detect { |m| m.version == @target_version }
  raise UnknownMigrationVersionError.new(@target_version) if migration.nil?
  result = execute_migration_in_transaction(migration)
  record_environment
  result
end

def runnable
  runnable = migrations[start..finish]
  if up?
    runnable.reject { |m| ran?(m) }
  else
    # skip the last migration if we're headed down, but not ALL the way down
    runnable.pop if target
    runnable.find_all { |m| ran?(m) }
  end
end

def start
  up? ? 0 : (migrations.index(current) || 0)
end

def target
  migrations.detect { |m| m.version == @target_version }
end

def up?
  @direction == :up
end

def use_advisory_lock?
  Base.connection.advisory_locks_enabled?
end

def use_transaction?(migration)
  !migration.disable_ddl_transaction && Base.connection.supports_ddl_transactions?
end

def validate(migrations)
  name, = migrations.group_by(&:name).find { |_, v| v.length > 1 }
  raise DuplicateMigrationNameError.new(name) if name
  version, = migrations.group_by(&:version).find { |_, v| v.length > 1 }
  raise DuplicateMigrationVersionError.new(version) if version
end

def with_advisory_lock
  lock_id = generate_migrator_advisory_lock_id
  with_advisory_lock_connection do |connection|
    got_lock = connection.get_advisory_lock(lock_id)
    raise ConcurrentMigrationError unless got_lock
    load_migrated # reload schema_migrations to be sure it wasn't changed by another process before we got the lock
    yield
  ensure
    if got_lock && !connection.release_advisory_lock(lock_id)
      raise ConcurrentMigrationError.new(
        ConcurrentMigrationError::RELEASE_LOCK_FAILED_MESSAGE
      )
    end
  end
end

def with_advisory_lock_connection
  pool = ActiveRecord::ConnectionAdapters::ConnectionHandler.new.establish_connection(
    ActiveRecord::Base.connection_db_config
  )
  pool.with_connection { |connection| yield(connection) }
ensure
  pool&.disconnect!
end