cartodb/lib/tasks/sync_tables.rake

namespace :cartodb do
  # This rake retrieves all sync tables that should get synchronized, and puts the synchronization tasks at Resque
  # NOTE: This version does not mark the tables as "enqueued", should be done if planning to run multiple instances
  desc 'Runs the sync tables process'
  task :sync_tables, [:force_all_arg] => [:environment] do |task, args|
    puts '> Sync tables started' if ENV['VERBOSE']

    require_relative '../../services/synchronizer/lib/synchronizer/collection'
    collection = CartoDB::Synchronizer::Collection.new

    # This fetches and enqueues
    collection.fetch_and_enqueue(args[:force_all_arg].present? ? args[:force_all_arg] : false)

    puts '> Sync tables finished' if ENV['VERBOSE']
  end


  desc 'Adds visualization_id to every Synchronization'
  task :populate_synchronization_visualization_ids => [:environment] do |task, args|
    require_relative '../../services/synchronizer/lib/synchronizer/collection'
    collection = CartoDB::Synchronizer::Collection.new

    collection.fetch_all.each { |record|
      begin
        synchronization = CartoDB::Synchronization::Member.new(id: record[:id]).fetch
      rescue KeyError
        synchronization = nil
      end
      if synchronization
        begin
          table = UserTable.where({
              name: synchronization.name,
              user_id: synchronization.user_id
            }).first
          if table.nil?
            puts "\nSync id '#{record[:id]}' related table not found"
          else
            table = table.service
          end
        rescue StandardError => exception
          table = nil
          puts "\nSync id '#{record[:id]}' errored: #{exception.inspect}"
        end
        unless table.nil?
          if synchronization.visualization_id.nil?
            begin
              synchronization.visualization_id = table.table_visualization.id
            rescue StandardError => exception
              puts "\nSync id '#{record[:id]}' errored, canonical visualization not found"
            end
            begin
              synchronization.store
              printf '.'
            rescue StandardError => exception
              puts "\nSync id '#{record[:id]}' errored: #{exception.inspect}"
            end
          else
            printf 'S'
          end
        end
      else
        puts "\nSync id '#{record[:id]}' errored: missing synchronization entry"
      end
    }
    puts "\nFINISHED"

  end

  def report_incompatible_bq_parameters(parameters)
    valid_params = %w(provider connection table sql_query import_as project dataset billing_project)
    valid_conn_params = %w(billing_project service_account access_token refresh_token default_project default_dataset)
    invalid_params = parameters.keys - valid_params
    invalid_conn_params = (parameters['connection'] || {}).keys - valid_conn_params
    puts "  Invalid parameters: #{invalid_params.inspect}" if invalid_params.present?
    puts "  Invalid connection parameters: #{invalid_conn_params.inspect}" if invalid_conn_params.present?
  end

  def replicate_bq_config
    puts 'Replicating configuration'
    bigquery = Carto::ConnectorProvider.find_by(name: 'bigquery')
    bigquery_beta = Carto::ConnectorProvider.find_by(name: 'bigquery-beta')
    Carto::ConnectorConfiguration.where(connector_provider_id: bigquery.id).find_each do |bq_config|
      unless Carto::ConnectorConfiguration.where(
        connector_provider_id: bigquery_beta.id,
        user_id: bq_config.user_id,
        organization_id: bq_config.organization_id
      ).exists?
        Carto::ConnectorConfiguration.create!(
          connector_provider: bigquery_beta,
          user_id: bq_config.user_id,
          organization_id: bq_config.organization_id,
          enabled: bq_config.enabled,
          max_rows: bq_config.max_rows
        )
        puts "  configuration for #{bq_config.user&.username || bq_config.configuration.name} replicated"
      end
    end
  end

  def bq_syncs_query(user: nil, sync: nil)
    condition = ''
    condition += "AND user_id = '#{user.id}'" if user.present?
    condition += "AND id = '#{sync.id}'" if sync.present?
    %{
      service_name = 'connector'
      #{condition}
      AND (state IN (
            '#{Carto::Synchronization::STATE_SUCCESS}', '#{Carto::Synchronization::STATE_SYNCING}',
            '#{Carto::Synchronization::STATE_QUEUED}', '#{Carto::Synchronization::STATE_CREATED}')
        OR (state = '#{Carto::Synchronization::STATE_FAILURE}'
              AND retried_times < #{CartoDB::Synchronization::Member::MAX_RETRIES}))
      AND ((service_item_id::JSON)#>>'{provider}') = 'bigquery-beta'
    }
  end

  def bq_sync_blocked_states
    [
      Carto::Synchronization::STATE_CREATED,
      Carto::Synchronization::STATE_QUEUED,
      Carto::Synchronization::STATE_SYNCING
    ]
  end

  desc 'Port BQ syncs to beta connector'
  task port_bq_syncs_to_beta: [:environment] do
    dry_mode = ENV['DRY_RUN'] != 'NO'

    if dry_mode
      puts 'running in "dry" mode; set DRY_RUN=NO to make actual changes'
    else
      Rake::Task['cartodb:connectors:create_providers'].invoke
      replicate_bq_config
    end

    number_of_pending_syncs = 0
    Carto::Synchronization.where(bq_syncs_query).find_each do |synchronization|
      next unless synchronization.user.state == 'active'

      sleep 0.2
      synchronization.transaction do
        synchronization.reload
        parameters = JSON.parse(synchronization.service_item_id)
        if synchronization.state.in? bq_sync_blocked_states
          puts "Synchronization #{synchronization.id} could not be modifed; state: #{synchronization.state}"
          number_of_pending_syncs += 1
        elsif dry_mode
          puts "Synchronization #{synchronization.id} would be modified to use bigquery-beta"
          report_incompatible_bq_parameters(parameters)
        else
          begin
            puts "Modifying #{synchronization.id} to use bigquery-beta"
            run_at = synchronization.run_at
            synchronization.update! run_at: nil

            # Change the provider id
            parameters['provider'] = 'bigquery-beta'
            synchronization.update! service_item_id: parameters.to_json
            report_incompatible_bq_parameters(parameters)
          rescue
            raise
          ensure
            synchronization.update! run_at: run_at
          end
        end
      end
    end
    if number_of_pending_syncs.positive?
      puts "#{number_of_pending_syncs} syncs could not be modified. . Please try again later."
    end
  end

  desc 'Port BQ beta syncs to new connector'
  task :port_beta_bq_syncs_to_new, [:username_or_sync_id] => :environment do |_task, args|
    dry_mode = ENV['DRY_RUN'] != 'NO'

    puts 'running in "dry" mode; set DRY_RUN=NO to make actual changes' if dry_mode

    if args.username_or_sync_id != 'all-the-users'
      user = Carto::User.find_by(username: args.username_or_sync_id)
      if user.blank?
        sync = Carto::Synchronization.find_by(id: args.username_or_sync_id)
        raise "User/Sync not found: #{args.username_or_sync_id}" unless sync
      end
    end

    sql = bq_syncs_query(user: user, sync: sync)

    number_of_pending_syncs = 0
    Carto::Synchronization.where(sql).find_each do |synchronization|
      next unless synchronization.user.state == 'active'

      sleep 0.2
      synchronization.transaction do
        synchronization.reload
        parameters = JSON.parse(synchronization.service_item_id)
        if synchronization.state.in? bq_sync_blocked_states
          puts "Synchronization #{synchronization.id} could not be modifed; state: #{synchronization.state}"
          number_of_pending_syncs += 1
        elsif dry_mode
          puts "Synchronization #{synchronization.id} would be modified to use bigquery"
          puts '  parameter billing_project would be moved to connection' if parameters['billing_project'].present?
          report_incompatible_bq_parameters(parameters)
        else
          begin
            puts "Modifying #{synchronization.id} to use bigquery"
            run_at = synchronization.run_at
            synchronization.update! run_at: nil

            # Change the provider id
            parameters['provider'] = 'bigquery'
            # If passing the billing project out of the connection move it inside
            if parameters['billing_project'].present?
              puts '  Moving billing_project inside the connection parameter'
              billing_project = parameters.delete('billing_project')
              parameters['connection'] ||= {}
              parameters['connection']['billing_project'] = billing_project
            end
            synchronization.update! service_item_id: parameters.to_json
            report_incompatible_bq_parameters(parameters)
          rescue
            raise
          ensure
            synchronization.update! run_at: run_at
          end
        end
      end
    end
    if number_of_pending_syncs.positive?
      puts "#{number_of_pending_syncs} syncs could not be modified. . Please try again later."
    end
  end

  desc 'Migrate legacy sync tables to new connections'
  task :migrate_legacy_sync_tables_to_new_connections, [:username, :synchronization_ids, :flexible_password] => :environment do |_task, args|

    # NOTE: Managing user
    user = Carto::User.find_by(username: args[:username])
    if user.blank?
      puts "> ERROR: Missing user with username '#{args[:username]}'"
      next
    end

    # NOTE: Managing synchronizations
    sync_ids = (args[:synchronization_ids].try(:split, ' ') || []).uniq
    synchronizations = user.synchronizations

    if sync_ids.empty?
      puts "> INFO: No synchronizations were provided, so all '#{user.username}' synchronizations will be analyzed"
    else
      synchronizations = synchronizations.where(id: sync_ids)

      if synchronizations.count != sync_ids.count
        puts "> ERROR: #{sync_ids.count} synchronization/s provided, but only #{synchronizations.count} found"
        next
      end
      puts "> INFO: List of #{synchronizations.count} synchronization/s provided"
    end

    # NOTE: Looking for legacy synchronizations
    legacy_syncs = synchronizations.where(url: ['', nil]).select do |sync|
      data = JSON.parse(sync.service_item_id)
      data['provider'].present? && data['connection'].present? && data['connection_id'].blank?
    end

    puts "> INFO: #{legacy_syncs.count} legacy synchronization/s were found"
    next if legacy_syncs.count.zero?

    # NOTE: Finding/creating required connections
    connection_groups = legacy_syncs.group_by do |sync|
      data = JSON.parse(sync.service_item_id)
      [data['provider'], data['connection']]
    end

    puts "> INFO: #{connection_groups.count} unique connections needed"
    if args[:flexible_password] == 'true'
      puts "> INFO: Password parameter will be omitted while looking for existent connections"
    end

    connection_manager = Carto::ConnectionManager.new(user)
    syncs_with_connection = connection_groups.map.with_index do |((provider, connection_info), syncs), index|
      puts "> INFO: Connection #{index} - [#{provider}] #{connection_info}"

      connections = user.db_connections
      connection_info.each do |key, value|
        next if key == 'password' && args[:flexible_password] == 'true'
        connections = connections.where("parameters->>'#{key}' = ?", value)
      end
      connection = connections.find_by(connector: provider)

      if connection.present?
        puts "> INFO: Connection found [#{connection.id}]"
      else
        connection = connection_manager.create_db_connection(
          name: "#{provider} [#{SecureRandom.hex(6)}]",
          provider: provider,
          parameters: connection_info
        )
        puts "> INFO: Connection created [#{connection.id}]"
      end

      {
        connection: connection,
        syncs: syncs,
        password_changed: connection_info['password'] != connection.parameters['password']
      }
    rescue StandardError => exception
      puts "> WARNING: Error finding/creating connection #{index} - #{exception.message}"
      puts "> INFO: Skipping #{syncs.count} synchronization/s - Sample sync [#{syncs.first.id}]"

      { connection: nil, syncs: [] }
    end

    # NOTE: Updating and synchronizations
    syncs_with_connection.each.with_index do |data, index|
      next if data[:connection].nil?

      puts "> INFO: Linking #{data[:syncs].count} synchronization/s to connection #{index}"
      ActiveRecord::Base.transaction do
        data[:syncs].each do |sync|
          service_item_id = JSON.parse(sync.service_item_id).except('connection')
          service_item_id['connection_id'] = data[:connection].id

          sync.update!(service_item_id: service_item_id.to_json)
        end
      end

      # NOTE: Queuing synchronizations
      next unless data[:password_changed]

      data[:syncs].each do |sync|
        next if sync.state != 'failure' || sync.retried_times < CartoDB::Synchronization::Member::MAX_RETRIES

        puts "> INFO: Queueing failed sync [#{sync.id}]"
        CartoDB::Synchronization::Member.new(sync).enqueue
      end
    rescue StandardError => exception
      puts "> WARNING: Error updating synchronizations for connection #{index} - #{exception.message}"
      puts "> INFO: Skipping updating #{syncs.count} synchronization/s - Sample sync [#{data[:syncs].first.id}]"
    end
  rescue StandardError => exception
    puts "> ERROR: Something went wrong while migrating legacy synchronizations | #{exception.inspect}"
  end
end