discourse/lib/import/importer.rb

module Import

  class ImportDisabledError  < RuntimeError; end
  class FilenameMissingError < RuntimeError; end

  class Importer

    def initialize(user_id, filename, publish_to_message_bus = false)
      @user_id, @filename, @publish_to_message_bus = user_id, filename, publish_to_message_bus

      ensure_import_is_enabled
      ensure_no_operation_is_running
      ensure_we_have_a_user
      ensure_we_have_a_filename

      initialize_state
    end

    def run
      log "[STARTED]"
      log "'#{@user_info[:username]}' has started the restore!"

      mark_import_as_running

      listen_for_shutdown_signal

      enable_readonly_mode

      pause_sidekiq
      wait_for_sidekiq

      ensure_directory_exists(@tmp_directory)

      copy_archive_to_tmp_directory
      unzip_archive

      extract_metadata
      validate_metadata

      extract_dump
      restore_dump

      switch_schema!

      # TOFIX: MessageBus is busted...

      migrate_database
      reconnect_database

      extract_uploads
    rescue SystemExit
      log "Restore process was cancelled!"
      rollback
    rescue Exception => ex
      log "EXCEPTION: " + ex.message
      log ex.backtrace.join("\n")
      rollback
    else
      @success = true
    ensure
      notify_user
      clean_up
      @success ? log("[SUCCESS]") : log("[FAILED]")
    end

    protected

    def ensure_import_is_enabled
      raise Import::ImportDisabledError unless SiteSetting.allow_restore?
    end

    def ensure_no_operation_is_running
      raise BackupRestore::OperationRunningError if BackupRestore.is_operation_running?
    end

    def ensure_we_have_a_user
      user = User.where(id: @user_id).first
      raise Discourse::InvalidParameters.new(:user_id) unless user
      # keep some user data around to check them against the newly restored database
      @user_info = { id: user.id, username: user.username, email: user.email }
    end

    def ensure_we_have_a_filename
      raise Import::FilenameMissingError if @filename.nil?
    end

    def initialize_state
      @success = false
      @current_db = RailsMultisite::ConnectionManagement.current_db
      @current_version = BackupRestore.current_version
      @timestamp = Time.now.strftime("%Y-%m-%d-%H%M%S")
      @tmp_directory = File.join(Rails.root, "tmp", "restores", @current_db, @timestamp)
      @archive_filename = File.join(@tmp_directory, @filename)
      @tar_filename = @archive_filename[0...-3]
      @meta_filename = File.join(@tmp_directory, BackupRestore::METADATA_FILE)
      @dump_filename = File.join(@tmp_directory, BackupRestore::DUMP_FILE)
      @logs = []
      @readonly_mode_was_enabled = Discourse.readonly_mode?
    end

    def listen_for_shutdown_signal
      Thread.new do
        while BackupRestore.is_operation_running?
          exit if BackupRestore.should_shutdown?
          sleep 0.1
        end
      end
    end

    def mark_import_as_running
      log "Marking restore as running..."
      BackupRestore.mark_as_running!
    end

    def enable_readonly_mode
      return if @readonly_mode_was_enabled
      log "Enabling readonly mode..."
      Discourse.enable_readonly_mode
    end

    def pause_sidekiq
      log "Pausing sidekiq..."
      Sidekiq.pause!
    end

    def wait_for_sidekiq
      log "Waiting for sidekiq to finish running jobs..."
      iterations = 1
      while sidekiq_has_running_jobs?
        log "Waiting for sidekiq to finish running jobs... ##{iterations}"
        sleep 5
        iterations += 1
        raise "Sidekiq did not finish running all the jobs in the allowed time!" if iterations > 6
      end
    end

    def sidekiq_has_running_jobs?
      Sidekiq::Workers.new.each do |process_id, thread_id, worker|
        payload = worker.try(:payload)
        return true if payload.try(:all_sites)
        return true if payload.try(:current_site_id) == @current_db
      end

      false
    end

    def copy_archive_to_tmp_directory
      log "Copying archive to tmp directory..."
      source = File.join(Backup.base_directory, @filename)
      `cp #{source} #{@archive_filename}`
    end

    def unzip_archive
      log "Unzipping archive..."
      FileUtils.cd(@tmp_directory) { `gzip --decompress #{@archive_filename}` }
    end

    def extract_metadata
      log "Extracting metadata file..."
      FileUtils.cd(@tmp_directory) { `tar --extract --file #{@tar_filename} #{BackupRestore::METADATA_FILE}` }
      @metadata = Oj.load_file(@meta_filename)
    end

    def validate_metadata
      log "Validating metadata..."
      log "  Current version: #{@current_version}"
      log "  Restored version: #{@metadata["version"]}"

      error = "You're trying to import a more recent version of the schema. You should migrate first!"
      raise error if @metadata["version"] > @current_version
    end

    def extract_dump
      log "Extracting dump file..."
      FileUtils.cd(@tmp_directory) { `tar --extract --file #{@tar_filename} #{BackupRestore::DUMP_FILE}` }
    end

    def restore_dump
      log "Restoring dump file... (can be quite long)"

      logs = Queue.new
      psql_running = true
      has_error = false

      Thread.new do
        while psql_running
          message = logs.pop.strip
          has_error ||= (message =~ /ERROR:/)
          log(message) unless message.blank?
        end
      end

      IO.popen("#{psql_command} 2>&1") do |pipe|
        begin
          while line = pipe.readline
            logs << line
          end
        rescue EOFError
          # finished reading...
        ensure
          psql_running = false
          logs << ""
        end
      end

      # psql does not return a valid exit code when an error happens
      raise "psql failed" if has_error
    end

    def psql_command
      db_conf = BackupRestore.database_configuration

      password_argument = "PGPASSWORD=#{db_conf.password}" if db_conf.password.present?
      host_argument     = "--host=#{db_conf.host}"         if db_conf.host.present?
      username_argument = "--username=#{db_conf.username}" if db_conf.username.present?

      [ password_argument,                # pass the password to psql (if any)
        "psql",                           # the psql command
        "--dbname='#{db_conf.database}'", # connect to database *dbname*
        "--file='#{@dump_filename}'",     # read the dump
        "--single-transaction",           # all or nothing (also runs COPY commands faster)
        host_argument,                    # the hostname to connect to (if any)
        username_argument                 # the username to connect as (if any)
      ].join(" ")
    end

    def switch_schema!
      log "Switching schemas..."

      sql = [
        "BEGIN;",
        BackupRestore.move_tables_between_schemas_sql("public", "backup"),
        BackupRestore.move_tables_between_schemas_sql("restore", "public"),
        "COMMIT;"
      ].join("\n")

      User.exec_sql(sql)
    end

    def migrate_database
      log "Migrating the database..."
      Discourse::Application.load_tasks
      ENV["VERSION"] = @current_version.to_s
      Rake::Task["db:migrate"].invoke
    end

    def reconnect_database
      log "Reconnecting to the database..."
      ActiveRecord::Base.establish_connection
    end

    def extract_uploads
      log "Extracting uploads..."
      if `tar --list --file #{@tar_filename} | grep 'uploads/'`.present?
        FileUtils.cd(File.join(Rails.root, "public")) do
          `tar --extract --keep-newer-files --file #{@tar_filename} uploads/`
        end
      end
    end

    def rollback
      log "Trying to rollback..."
      if BackupRestore.can_rollback?
        log "Rolling back..."
        BackupRestore.move_tables_between_schemas("backup", "public")
      else
        log "There was no need to rollback"
      end
    end

    def notify_user
      if user = User.where(email: @user_info[:email]).first
        log "Notifying '#{user.username}' of the end of the restore..."
        # NOTE: will only notify if user != Discourse.site_contact_user
        if @success
          SystemMessage.create(user, :import_succeeded)
        else
          SystemMessage.create(user, :import_failed, logs: @logs.join("\n"))
        end
      else
        log "Could not send notification to '#{@user_info[:username]}' (#{@user_info[:email]}), because the user does not exists..."
      end
    end

    def clean_up
      log "Cleaning stuff up..."
      remove_tmp_directory
      unpause_sidekiq
      disable_readonly_mode
      mark_import_as_not_running
      log "Finished!"
    end

    def remove_tmp_directory
      log "Removing tmp '#{@tmp_directory}' directory..."
      FileUtils.rm_rf(@tmp_directory) if Dir[@tmp_directory].present?
    rescue
      log "Something went wrong while removing the following tmp directory: #{@tmp_directory}"
    end

    def unpause_sidekiq
      log "Unpausing sidekiq..."
      Sidekiq.unpause!
    end

    def disable_readonly_mode
      return if @readonly_mode_was_enabled
      log "Disabling readonly mode..."
      Discourse.disable_readonly_mode
    end

    def mark_import_as_not_running
      log "Marking restore as finished..."
      BackupRestore.mark_as_not_running!
    end

    def ensure_directory_exists(directory)
      log "Making sure #{directory} exists..."
      FileUtils.mkdir_p(directory)
    end

    def log(message)
      puts(message) rescue nil
      publish_log(message) rescue nil
      save_log(message)
    end

    def publish_log(message)
      return unless @publish_to_message_bus
      data = { timestamp: Time.now, operation: "restore", message: message }
      MessageBus.publish(BackupRestore::LOGS_CHANNEL, data, user_ids: [@user_id])
    end

    def save_log(message)
      @logs << "[#{Time.now}] #{message}"
    end

  end

end