cartodb/lib/carto/ghost_tables_manager.rb

require_relative 'bolt.rb'

module Carto
  class GhostTablesManager
    MUTEX_REDIS_KEY = 'ghost_tables_working'.freeze
    MUTEX_TTL_MS = 600000
    MAX_TABLES_FOR_SYNC_RUN = 8

    def initialize(user_id)
      @user_id = user_id
    end

    def user
      @user ||= ::User[@user_id]
    end

    def link_ghost_tables
      return if user_tables_synced_with_db?

      if should_run_synchronously?
        link_ghost_tables_synchronously
      else
        ::Resque.enqueue(::Resque::UserDBJobs::UserDBMaintenance::LinkGhostTables, @user_id)
      end
    end

    def link_ghost_tables_synchronously
      sync_user_tables_with_db unless user_tables_synced_with_db?
    end

    # determine linked tables vs cartodbfied tables consistency; i.e.: needs to run
    def user_tables_synced_with_db?
      user_tables = fetch_user_tables
      cartodbfied_tables = fetch_cartodbfied_tables

      user_tables.length == cartodbfied_tables.length &&
        (user_tables - cartodbfied_tables).empty? &&
        (cartodbfied_tables - user_tables).empty?
    end

    def get_bolt
      Carto::Bolt.new("#{user.username}:#{MUTEX_REDIS_KEY}", ttl_ms: MUTEX_TTL_MS)
    end

    # run a block of code exclusively with GhostTablesManager (using Bolt lock)
    # if warning_params is provided (with paramters for Logger.warning) then
    # the code is executed even if the lock is not acquired (in which case
    # a warning is emmitted)
    def self.run_synchronized(user_id, attempts: 10, timeout: 30000, **warning_params)
      gtm = new(user_id)
      bolt = gtm.get_bolt
      rerun_func = lambda { gtm.send(:sync) }
      lock_acquired = bolt.run_locked(attempts: attempts, timeout: timeout, rerun_func: rerun_func) do
        yield
      end
      if !lock_acquired && warning_params.present?
        # run even if lock wasn't aquired
        CartoDB::Logger.warning(warning_params)
        yield
      end
    end

    private

    # It's nice to run sync if any unsafe stale (dropped or renamed) tables will be shown to the user but we can't block
    # the workers for more that 180 seconds
    def should_run_synchronously?
      cartodbfied_tables = fetch_cartodbfied_tables

      dropped_and_stale_tables = find_dropped_tables(cartodbfied_tables) + find_stale_tables(cartodbfied_tables)
      total_tables_to_be_linked = dropped_and_stale_tables + find_new_tables(cartodbfied_tables)

      dropped_and_stale_tables.count != 0 && total_tables_to_be_linked.count < MAX_TABLES_FOR_SYNC_RUN
    end

    def sync_user_tables_with_db
      got_locked = get_bolt.run_locked(rerun_func: lambda { sync }) { sync }

      CartoDB::Logger.info(message: 'Ghost table race condition avoided', user: user) unless got_locked
    end

    def sync
      cartodbfied_tables = fetch_cartodbfied_tables

      # Update table_id on UserTables with physical tables with changed oid. Should go first.
      find_regenerated_tables(cartodbfied_tables).each(&:regenerate_user_table)

      # Relink tables that have been renamed through the SQL API
      find_renamed_tables(cartodbfied_tables).each(&:rename_user_table_vis)

      # Create UserTables for non linked Tables
      find_new_tables(cartodbfied_tables).each(&:create_user_table)

      # Unlink tables that have been created through the SQL API. Should go last.
      find_dropped_tables(cartodbfied_tables).each(&:drop_user_table)
    end

    # Any UserTable that has been renamed or regenerated.
    def find_stale_tables(cartodbfied_tables)
      find_regenerated_tables(cartodbfied_tables) | find_renamed_tables(cartodbfied_tables)
    end

    # UserTables that coincide with a cartodbfied table in name but not id
    def find_renamed_tables(cartodbfied_tables)
      user_tables = fetch_user_tables

      cartodbfied_tables.select do |cartodbfied_table|
        user_tables.any?{|t| t.name != cartodbfied_table.name && t.id == cartodbfied_table.id}
      end
    end

    # UserTables that coincide with a cartodbfied table in id but not in name
    def find_regenerated_tables(cartodbfied_tables)
      user_tables = fetch_user_tables

      cartodbfied_tables.select do |cartodbfied_table|
        user_tables.any?{|t| t.name == cartodbfied_table.name && t.id != cartodbfied_table.id}
      end
    end

    # Cartodbfied tables that are not stale and are not linked as UserTables yet
    def find_new_tables(cartodbfied_tables)
      cartodbfied_tables - fetch_user_tables - find_stale_tables(cartodbfied_tables)
    end

    # UserTables that are not stale and have no cartodbfied table associated to it
    def find_dropped_tables(cartodbfied_tables)
      fetch_user_tables - cartodbfied_tables - find_stale_tables(cartodbfied_tables)
    end

    # Fetches all currently linked user tables
    def fetch_user_tables
      Carto::UserTable.select([:name, :table_id]).where(user_id: @user_id).map do |record|
        Carto::TableFacade.new(record[:table_id], record[:name], @user_id)
      end
    end

    # Fetches all linkable tables: non raster cartodbfied + raster
    def fetch_cartodbfied_tables
      (fetch_non_raster_cartodbfied_tables + fetch_raster_tables).uniq
    end

    # this method searchs for tables with all the columns needed in a cartodb table.
    # it does not check column types, and only the latest cartodbfication trigger attached (test_quota_per_row)
    def fetch_non_raster_cartodbfied_tables
      cartodb_columns = (Table::CARTODB_REQUIRED_COLUMNS + [Table::THE_GEOM_WEBMERCATOR]).map { |col| "'#{col}'" }
                                                                                         .join(',')

      sql = %{
        WITH cartodbfied_tables as (
          SELECT c.table_name,
                 tg.tgrelid reloid,
                 count(column_name::text) cdb_columns_count
          FROM information_schema.columns c, pg_tables t, pg_trigger tg
          WHERE
            t.tablename !~ '^importer_' AND
            t.tablename = c.table_name AND
            t.schemaname = c.table_schema AND
            c.table_schema = '#{user.database_schema}' AND
            t.tableowner in ('#{user.get_database_roles.join('\',\'')}') AND
            column_name IN (#{cartodb_columns}) AND
            tg.tgrelid = (quote_ident(t.schemaname) || '.' || quote_ident(t.tablename))::regclass::oid AND
            tg.tgname = 'test_quota_per_row'
            GROUP BY reloid, 1)
        SELECT table_name, reloid FROM cartodbfied_tables WHERE cdb_columns_count = #{cartodb_columns.split(',').length}
      }

      user.in_database(as: :superuser)[sql].all.map do |record|
        Carto::TableFacade.new(record[:reloid], record[:table_name], @user_id)
      end
    end

    # Find raster tables which won't appear as cartodbfied but MUST be linked
    def fetch_raster_tables
      sql = %{
        WITH cartodbfied_tables as (
          SELECT c.table_name,
                 tg.tgrelid reloid,
                 count(column_name::text) cdb_columns_count
          FROM information_schema.columns c, pg_tables t, pg_trigger tg
          WHERE
            t.tablename = c.table_name AND
            t.schemaname = c.table_schema AND
            c.table_schema = '#{user.database_schema}' AND
            t.tableowner in ('#{user.get_database_roles.join('\',\'')}') AND
            column_name IN ('cartodb_id', 'the_raster_webmercator') AND
            tg.tgrelid = (quote_ident(t.schemaname) || '.' || quote_ident(t.tablename))::regclass::oid AND
            tg.tgname = 'test_quota_per_row'
            GROUP BY reloid, 1)
        SELECT table_name, reloid FROM cartodbfied_tables WHERE cdb_columns_count = 2;
      }

      user.in_database(as: :superuser)[sql].all.map do |record|
        Carto::TableFacade.new(record[:reloid], record[:table_name], @user_id)
      end
    end
  end

  class TableFacade
    attr_reader :id, :name, :user_id

    def initialize(id, name, user_id)
      @id = id
      @name = name
      @user_id = user_id
    end

    def user
      @user ||= ::User[@user_id]
    end

    def user_table_with_matching_id
      user.tables.where(table_id: id).first
    end

    def user_table_with_matching_name
      user.tables.where(name: name).first
    end

    def create_user_table
      CartoDB::Logger.debug(message: 'ghost tables',
                            action: 'linking new table',
                            user: user,
                            table_name: name,
                            table_id: id)
      user_table = Carto::UserTable.new
      user_table.user_id = user.id
      user_table.table_id = id
      user_table.name = name
      new_table = ::Table.new(user_table: user_table)

      new_table.register_table_only = true
      new_table.keep_user_database_table = true

      new_table.save
    rescue StandardError => exception
      CartoDB::Logger.error(message: 'Ghost tables: Error creating UserTable',
                            exception: exception,
                            user: user,
                            table_name: name,
                            table_id: id)
    end

    def rename_user_table_vis
      CartoDB::Logger.debug(message: 'ghost tables',
                            action: 'relinking renamed table',
                            user: user,
                            table_name: name,
                            table_id: id)

      user_table_vis = user_table_with_matching_id.table_visualization

      user_table_vis.register_table_only = true
      user_table_vis.name = name

      user_table_vis.store
    rescue StandardError => exception
      CartoDB::Logger.error(message: 'Ghost tables: Error renaming Visualization',
                            exception: exception,
                            user: user,
                            table_name: name,
                            table_id: id)
    end

    def drop_user_table
      CartoDB::Logger.debug(message: 'ghost tables',
                            action: 'unlinking dropped table',
                            user: user,
                            table_name: name,
                            table_id: id)

      user_table_to_drop = user.tables.where(table_id: id, name: name).first
      return unless user_table_to_drop # The table has already been deleted

      table_to_drop = ::Table.new(user_table: user_table_to_drop)
      table_to_drop.keep_user_database_table = true
      table_to_drop.destroy
    rescue StandardError => exception
      CartoDB::Logger.error(message: 'Ghost tables: Error dropping Table',
                            exception: exception,
                            user: user,
                            table_name: name,
                            table_id: id)
    end

    def regenerate_user_table
      CartoDB::Logger.debug(message: 'ghost tables',
                            action: 'regenerating table_id',
                            user: user,
                            table_name: name,
                            table_id: id)

      user_table_to_regenerate = user_table_with_matching_name

      user_table_to_regenerate.table_id = id
      user_table_to_regenerate.save
    rescue StandardError => exception
      CartoDB::Logger.error(message: 'Ghost tables: Error syncing table_id for UserTable',
                            exception: exception,
                            user: user,
                            table_name: name,
                            table_id: id)
    end

    def eql?(other)
      id.eql?(other.id) && name.eql?(other.name) && user.id.eql?(other.user_id)
    end

    def ==(other)
      eql?(other)
    end

    def hash
      [id, name, user_id].hash
    end
  end
end