rename export/import in favor of backup/restore for better consistency

lib/backup_restore/backup_restore.rb

@@ -0,0 +1,196 @@
+require "backup_restore/backuper"
+require "backup_restore/restorer"
+
+module BackupRestore
+
+  class OperationRunningError < RuntimeError; end
+
+  DUMP_FILE = "dump.sql"
+  METADATA_FILE = "meta.json"
+  LOGS_CHANNEL = "/admin/backups/logs"
+
+  def self.backup!(user_id, opts={})
+    start! BackupRestore::Backuper.new(user_id, opts)
+  end
+
+  def self.restore!(user_id, filename, publish_to_message_bus=false)
+    start! BackupRestore::Restorer.new(user_id, filename, publish_to_message_bus)
+  end
+
+  def self.rollback!
+    raise BackupRestore::OperationRunningError if BackupRestore.is_operation_running?
+    if can_rollback?
+      move_tables_between_schemas("backup", "public")
+      after_fork
+    end
+  end
+
+  def self.cancel!
+    set_shutdown_signal!
+    true
+  end
+
+  def self.mark_as_running!
+    $redis.setex(running_key, 60, "1")
+    save_start_logs_message_id
+    keep_it_running
+  end
+
+  def self.is_operation_running?
+    !!$redis.get(running_key)
+  end
+
+  def self.mark_as_not_running!
+    $redis.del(running_key)
+  end
+
+  def self.should_shutdown?
+    !!$redis.get(shutdown_signal_key)
+  end
+
+  def self.can_rollback?
+    backup_tables_count > 0
+  end
+
+  def self.operations_status
+    {
+      is_operation_running: is_operation_running?,
+      can_rollback: can_rollback?,
+      allow_restore: Rails.env.development? || SiteSetting.allow_restore
+    }
+  end
+
+  def self.logs
+    id = start_logs_message_id
+    MessageBus.backlog(LOGS_CHANNEL, id).map { |m| m.data }
+  end
+
+  def self.current_version
+    ActiveRecord::Migrator.current_version
+  end
+
+  def self.move_tables_between_schemas(source, destination)
+    User.exec_sql(move_tables_between_schemas_sql(source, destination))
+  end
+
+  def self.move_tables_between_schemas_sql(source, destination)
+    <<-SQL
+      DO $$DECLARE row record;
+      BEGIN
+        -- create <destination> schema if it does not exists already
+        -- NOTE: DROP & CREATE SCHEMA is easier, but we don't want to drop the public schema
+        -- ortherwise extensions (like hstore & pg_trgm) won't work anymore...
+        CREATE SCHEMA IF NOT EXISTS #{destination};
+        -- move all <source> tables to <destination> schema
+        FOR row IN SELECT tablename FROM pg_tables WHERE schemaname = '#{source}'
+        LOOP
+          EXECUTE 'DROP TABLE IF EXISTS #{destination}.' || quote_ident(row.tablename) || ' CASCADE;';
+          EXECUTE 'ALTER TABLE #{source}.' || quote_ident(row.tablename) || ' SET SCHEMA #{destination};';
+        END LOOP;
+        -- move all <source> views to <destination> schema
+        FOR row IN SELECT viewname FROM pg_views WHERE schemaname = '#{source}'
+        LOOP
+          EXECUTE 'DROP VIEW IF EXISTS #{destination}.' || quote_ident(row.viewname) || ' CASCADE;';
+          EXECUTE 'ALTER VIEW #{source}.' || quote_ident(row.viewname) || ' SET SCHEMA #{destination};';
+        END LOOP;
+      END$$;
+    SQL
+  end
+
+  DatabaseConfiguration = Struct.new(:host, :port, :username, :password, :database)
+
+  def self.database_configuration
+    config = Rails.env.production? ? ActiveRecord::Base.connection_pool.spec.config : Rails.configuration.database_configuration[Rails.env]
+    config = config.with_indifferent_access
+
+    DatabaseConfiguration.new(
+      config["host"],
+      config["port"],
+      config["username"] || ENV["USER"] || "postgres",
+      config["password"],
+      config["database"]
+    )
+  end
+
+  private
+
+  def self.running_key
+    "backup_restore_operation_is_running"
+  end
+
+  def self.keep_it_running
+    # extend the expiry by 1 minute every 30 seconds
+    Thread.new do
+      # this thread will be killed when the fork dies
+      while true
+        $redis.expire(running_key, 1.minute)
+        sleep 30.seconds
+      end
+    end
+  end
+
+  def self.shutdown_signal_key
+    "backup_restore_operation_should_shutdown"
+  end
+
+  def self.set_shutdown_signal!
+    $redis.set(shutdown_signal_key, "1")
+  end
+
+  def self.clear_shutdown_signal!
+    $redis.del(shutdown_signal_key)
+  end
+
+  def self.save_start_logs_message_id
+    id = MessageBus.last_id(LOGS_CHANNEL)
+    $redis.set(start_logs_message_id_key, id)
+  end
+
+  def self.start_logs_message_id
+    $redis.get(start_logs_message_id_key).to_i
+  end
+
+  def self.start_logs_message_id_key
+    "start_logs_message_id"
+  end
+
+  def self.start!(runner)
+    child = fork do
+      begin
+        after_fork
+        runner.run
+      rescue Exception => e
+        puts "--------------------------------------------"
+        puts "---------------- EXCEPTION -----------------"
+        puts e.message
+        puts e.backtrace.join("\n")
+        puts "--------------------------------------------"
+      ensure
+        begin
+          clear_shutdown_signal!
+        rescue Exception => e
+          puts "============================================"
+          puts "================ EXCEPTION ================="
+          puts e.message
+          puts e.backtrace.join("\n")
+          puts "============================================"
+        ensure
+          exit!(0)
+        end
+      end
+    end
+
+    Process.detach(child)
+
+    true
+  end
+
+  def self.after_fork
+    Discourse.after_fork
+  end
+
+  def self.backup_tables_count
+    User.exec_sql("SELECT COUNT(*) AS count FROM information_schema.tables WHERE table_schema = 'backup'")[0]['count'].to_i
+  end
+
+end

lib/backup_restore/backuper.rb

@@ -0,0 +1,341 @@
+module BackupRestore
+
+  class Backuper
+
+    attr_reader :success
+
+    def initialize(user_id, opts={})
+      @user_id = user_id
+      @publish_to_message_bus = opts[:publish_to_message_bus] || false
+      @with_uploads = opts[:with_uploads].nil? ? true : opts[:with_uploads]
+
+      ensure_no_operation_is_running
+      ensure_we_have_a_user
+
+      initialize_state
+    end
+
+    def run
+      log "[STARTED]"
+      log "'#{@user.username}' has started the backup!"
+
+      mark_backup_as_running
+
+      listen_for_shutdown_signal
+
+      ensure_directory_exists(@tmp_directory)
+      ensure_directory_exists(@archive_directory)
+
+      write_metadata
+
+      ### READ-ONLY / START ###
+      enable_readonly_mode
+
+      pause_sidekiq
+      wait_for_sidekiq
+
+      dump_public_schema
+
+      disable_readonly_mode
+      ### READ-ONLY / END ###
+
+      log "Finalizing backup..."
+
+      update_dump
+
+      create_archive
+
+      after_create_hook
+
+      remove_old
+    rescue SystemExit
+      log "Backup process was cancelled!"
+    rescue Exception => ex
+      log "EXCEPTION: " + ex.message
+      log ex.backtrace.join("\n")
+    else
+      @success = true
+      "#{@archive_basename}.tar.gz"
+    ensure
+      notify_user rescue nil
+      clean_up
+      @success ? log("[SUCCESS]") : log("[FAILED]")
+    end
+
+    protected
+
+    def ensure_no_operation_is_running
+      raise BackupRestore::OperationRunningError if BackupRestore.is_operation_running?
+    end
+
+    def ensure_we_have_a_user
+      @user = User.find_by(id: @user_id)
+      raise Discourse::InvalidParameters.new(:user_id) unless @user
+    end
+
+    def initialize_state
+      @success = false
+      @current_db = RailsMultisite::ConnectionManagement.current_db
+      @timestamp = Time.now.strftime("%Y-%m-%d-%H%M%S")
+      @tmp_directory = File.join(Rails.root, "tmp", "backups", @current_db, @timestamp)
+      @dump_filename = File.join(@tmp_directory, BackupRestore::DUMP_FILE)
+      @meta_filename = File.join(@tmp_directory, BackupRestore::METADATA_FILE)
+      @archive_directory = File.join(Rails.root, "public", "backups", @current_db)
+      @archive_basename = File.join(@archive_directory, "#{SiteSetting.title.parameterize}-#{@timestamp}")
+      @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_backup_as_running
+      log "Marking backup 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 |_, _, 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 write_metadata
+      log "Writing metadata to '#{@meta_filename}'..."
+      metadata = {
+        source: "discourse",
+        version: BackupRestore.current_version
+      }
+      File.write(@meta_filename, metadata.to_json)
+    end
+
+    def dump_public_schema
+      log "Dumping the public schema of the database..."
+
+      logs = Queue.new
+      pg_dump_running = true
+
+      Thread.new do
+        RailsMultisite::ConnectionManagement::establish_connection(db: @current_db)
+        while pg_dump_running
+          message = logs.pop.strip
+          log(message) unless message.blank?
+        end
+      end
+
+      IO.popen("#{pg_dump_command} 2>&1") do |pipe|
+        begin
+          while line = pipe.readline
+            logs << line
+          end
+        rescue EOFError
+          # finished reading...
+        ensure
+          pg_dump_running = false
+          logs << ""
+        end
+      end
+
+      raise "pg_dump failed" unless $?.success?
+    end
+
+    def pg_dump_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?
+      port_argument     = "--port=#{db_conf.port}"         if db_conf.port.present?
+      username_argument = "--username=#{db_conf.username}" if db_conf.username.present?
+
+      [ password_argument,            # pass the password to pg_dump (if any)
+        "pg_dump",                    # the pg_dump command
+        "--schema=public",            # only public schema
+        "--file='#{@dump_filename}'", # output to the dump.sql file
+        "--no-owner",                 # do not output commands to set ownership of objects
+        "--no-privileges",            # prevent dumping of access privileges
+        "--verbose",                  # specifies verbose mode
+        host_argument,                # the hostname to connect to (if any)
+        port_argument,                # the port to connect to (if any)
+        username_argument,            # the username to connect as (if any)
+        db_conf.database              # the name of the database to dump
+      ].join(" ")
+    end
+
+    def update_dump
+      log "Updating dump for more awesomeness..."
+
+      `#{sed_command}`
+    end
+
+    def sed_command
+      # in order to limit the downtime when restoring as much as possible
+      # we force the restoration to happen in the "restore" schema
+
+      # during the restoration, this make sure we
+      #  - drop the "restore" schema if it exists
+      #  - create the "restore" schema
+      #  - prepend the "restore" schema into the search_path
+
+      regexp = "SET search_path = public, pg_catalog;"
+
+      replacement = [ "DROP SCHEMA IF EXISTS restore CASCADE;",
+                      "CREATE SCHEMA restore;",
+                      "SET search_path = restore, public, pg_catalog;",
+                    ].join(" ")
+
+      # we only want to replace the VERY first occurence of the search_path command
+      expression = "1,/^#{regexp}$/s/#{regexp}/#{replacement}/"
+
+      # I tried to use the --in-place argument but it was SLOOOWWWWwwwwww
+      # so I output the result into another file and rename it back afterwards
+      [ "sed -e '#{expression}' < #{@dump_filename} > #{@dump_filename}.tmp",
+        "&&",
+        "mv #{@dump_filename}.tmp #{@dump_filename}",
+      ].join(" ")
+    end
+
+    def create_archive
+      log "Creating archive: #{File.basename(@archive_basename)}.tar.gz"
+
+      tar_filename = "#{@archive_basename}.tar"
+
+      log "Making sure archive does not already exist..."
+      `rm -f #{tar_filename}`
+      `rm -f #{tar_filename}.gz`
+
+      log "Creating empty archive..."
+      `tar --create --file #{tar_filename} --files-from /dev/null`
+
+      log "Archiving metadata..."
+      FileUtils.cd(File.dirname(@meta_filename)) do
+        `tar --append --dereference --file #{tar_filename} #{File.basename(@meta_filename)}`
+      end
+
+      log "Archiving data dump..."
+      FileUtils.cd(File.dirname(@dump_filename)) do
+        `tar --append --dereference --file #{tar_filename} #{File.basename(@dump_filename)}`
+      end
+
+      if @with_uploads
+        upload_directory = "uploads/" + @current_db
+
+        log "Archiving uploads..."
+        FileUtils.cd(File.join(Rails.root, "public")) do
+          `tar --append --dereference --file #{tar_filename} #{upload_directory}`
+        end
+      end
+
+      log "Gzipping archive..."
+      `gzip --best #{tar_filename}`
+    end
+
+    def after_create_hook
+      log "Executing the after_create_hook for the backup"
+      backup = Backup.create_from_filename("#{File.basename(@archive_basename)}.tar.gz")
+      backup.after_create_hook
+    end
+
+    def remove_old
+      log "Removing old backups..."
+      Backup.remove_old
+    end
+
+    def notify_user
+      log "Notifying '#{@user.username}' of the end of the backup..."
+      if @success
+        SystemMessage.create_from_system_user(@user, :backup_succeeded)
+      else
+        SystemMessage.create_from_system_user(@user, :backup_failed, logs: @logs.join("\n"))
+      end
+    end
+
+    def clean_up
+      log "Cleaning stuff up..."
+      remove_tmp_directory
+      unpause_sidekiq
+      disable_readonly_mode if Discourse.readonly_mode?
+      mark_backup_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!
+    rescue
+      log "Something went wrong while unpausing Sidekiq."
+    end
+
+    def disable_readonly_mode
+      return if @readonly_mode_was_enabled
+      log "Disabling readonly mode..."
+      Discourse.disable_readonly_mode
+    end
+
+    def mark_backup_as_not_running
+      log "Marking backup 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: "backup", message: message }
+      MessageBus.publish(BackupRestore::LOGS_CHANNEL, data, user_ids: [@user_id])
+    end
+
+    def save_log(message)
+      @logs << "[#{Time.now}] #{message}"
+    end
+
+  end
+
+end

lib/backup_restore/restorer.rb

@@ -0,0 +1,359 @@
+module BackupRestore
+
+  class RestoreDisabledError  < RuntimeError; end
+  class FilenameMissingError < RuntimeError; end
+
+  class Restorer
+
+    attr_reader :success
+
+    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_restore_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_restore_as_running
+
+      listen_for_shutdown_signal
+
+      ensure_directory_exists(@tmp_directory)
+
+      copy_archive_to_tmp_directory
+      unzip_archive
+
+      extract_metadata
+      validate_metadata
+
+      extract_dump
+      restore_dump
+
+      ### READ-ONLY / START ###
+      enable_readonly_mode
+
+      pause_sidekiq
+      wait_for_sidekiq
+
+      switch_schema!
+
+      # TOFIX: MessageBus is busted...
+
+      migrate_database
+      reconnect_database
+      reload_site_settings
+
+      disable_readonly_mode
+      ### READ-ONLY / END ###
+
+      extract_uploads
+    rescue SystemExit
+      log "Restore process was cancelled!"
+      rollback
+    rescue => ex
+      log "EXCEPTION: " + ex.message
+      log ex.backtrace.join("\n")
+      rollback
+    else
+      @success = true
+    ensure
+      notify_user rescue nil
+      clean_up
+      @success ? log("[SUCCESS]") : log("[FAILED]")
+    end
+
+    protected
+
+    def ensure_restore_is_enabled
+      raise Restore::RestoreDisabledError unless Rails.env.development? || 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.find_by(id: @user_id)
+      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 Restore::FilenameMissingError if @filename.nil?
+    end
+
+    def initialize_state
+      @success = false
+      @db_was_changed = 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_restore_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 |_, _, 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 restore 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
+        RailsMultisite::ConnectionManagement::establish_connection(db: @current_db)
+        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?
+      port_argument     = "--port=#{db_conf.port}"         if db_conf.port.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)
+        port_argument,                # the port 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")
+
+      @db_was_changed = true
+
+      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..."
+      RailsMultisite::ConnectionManagement::establish_connection(db: @current_db)
+    end
+
+    def reload_site_settings
+      log "Reloading site settings..."
+      SiteSetting.refresh!
+    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 @db_was_changed && 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.find_by(email: @user_info[:email])
+        log "Notifying '#{user.username}' of the end of the restore..."
+        if @success
+          SystemMessage.create_from_system_user(user, :restore_succeeded)
+        else
+          SystemMessage.create_from_system_user(user, :restore_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 if Discourse.readonly_mode?
+      mark_restore_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!
+    rescue
+      log "Something went wrong while unpausing Sidekiq."
+    end
+
+    def disable_readonly_mode
+      return if @readonly_mode_was_enabled
+      log "Disabling readonly mode..."
+      Discourse.disable_readonly_mode
+    end
+
+    def mark_restore_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