table.rb

require "memoist"
require "logging"
require_relative "log_helper"
require_relative "mysql_connection"

DEFAULT_CHUNK_SIZE = 4096

# Encapsulate comparison logic
class AbstractComparison
  attr_accessor :master, :replica

  def initialize(opts = {})
    @master = opts[:master]
    @replica = opts[:replica]
  end

  def master?
    !@master.nil?
  end

  def replica?
    !@replica.nil?
  end

  def primary_key
    @master&.primary_key || @replica&.primary_key
  end

  def compare
    master.equal?(replica)
  end
end

# Encapsulate chunk comparison logic
class ChunkComparison < AbstractComparison
  attr_accessor :min_row, :max_row, :table_name, :visible_min_row, :visible_max_row, :primary_key

  def initialize(opts = {})
    super
    @max_row = opts[:max_row]
    @min_row = opts[:min_row]
    @visible_max_row = opts.fetch(:visible_max_row) do
      @max_row
    end
    @visible_min_row = opts.fetch(:visible_min_row) do
      @min_row
    end
    @table_name = opts[:table_name]
    @primary_key = opts[:primary_key]
  end

  def debug_info
    {
      table_name: @table_name,
      min_row: [master.min, replica.min],
      max_row: [master.max, replica.max],
      visible_min_row: [master.visible_min, replica.visible_min],
      visible_max_row: [master.visible_max, replica.visible_max],
      count: [master.count, replica.count],
      crc32: [master.crc32, replica.crc32],
      mismatch: compare ? "false" : "true",
      primary_key: @primary_key
    }
  end
end

# Encapsulate row comparison logic
class RowComparison < AbstractComparison
  attr_accessor :row_id, :table_name, :visible_row_id

  def initialize(opts = {})
    super
    @row_id = opts[:row_id]
    @table_name = opts[:table_name]
    @visible_row_id = opts.fetch(:visible_row_id) do
      @row_id
    end
  end

  def self.from(opts = {})
    master = opts[:master]
    replica = opts[:replica]
    raise "must provide :master or :replica" unless master || replica

    opts[:row_id] = master&.row_id || replica.row_id
    opts[:table_name] = master&.table_name || replica.table_name
    # fall back to row_id if there's not a defined visible_row_id
    opts[:visible_row_id] = master&.visible_row_id || replica&.visible_row_id || opts[:row_id]
    RowComparison.new(opts)
  end
end

# Encapsulate per-table logic
class Table
  include LogHelper

  attr_accessor :conn, :name

  def initialize(table_name, connection, opts = {})
    @conn = connection
    @name = table_name
    @logger = opts.fetch(:logger) do
      logger
    end
    @primary_key = opts[:primary_key]
  end

  def max_row_id
    @conn.max_row_id(@name)
  end

  def min_row_id
    @conn.min_row_id(@name)
  end

  def row_count
    @conn.row_count(@name)
  end

  def chunk_checksum(opts = {})
    @conn.chunk_checksum(@name, opts)
  end

  def row_checksum(opts = {})
    @conn.row_checksum(@name, opts)
  end

  def primary_key_columns
    @conn.primary_key_columns(@name)
  end

  def primary_key
    if @primary_key
      @primary_key
    else
      @conn.primary_key(table_name: @name)
    end
  end

  def row_values(row_id)
    @conn.row_values(@name, row_id)
  end
end

# Encapsulate table-pair (master:slave) logic
class TablePair
  extend Memoist
  include LogHelper

  attr_accessor :database_name, :master, :replica, :table_name

  def initialize(table_name, master_connection, replica_connection, opts = {})
    @table_name = table_name
    @master = Table.new(table_name, master_connection, opts)
    @replica = Table.new(table_name, replica_connection, opts)
    @chunk_size = opts.fetch(:chunk_size, DEFAULT_CHUNK_SIZE)
    @database_name = opts.fetch(:database_name, nil)
    @logger = opts.fetch(:logger) do
      logger
    end
  end

  def min_row_id
    mmin = @master.min_row_id
    rmin = @replica.min_row_id
    mmin <= rmin ? mmin : rmin
  end

  def max_row_id
    mmax = @master.max_row_id
    rmax = @replica.max_row_id
    mmax >= rmax ? mmax : rmax
  end

  def delta(opts = {})
    # First check the row count on either side
    row_count_diff = self.row_count_diff
    @logger.debug("row_count_diff: #{row_count_diff.inspect}")
    @logger.info("row count mismatch: #{row_count_diff}") unless row_count_diff[:diff] == 0

    row_mismatch = row_count_diff[:diff] != 0 ||
      row_count_diff[:rmax] <=> row_count_diff[:mmax] ||
      row_count_diff[:rmin] <=> row_count_diff[:mmin]
    @logger.debug("row_mismatch: #{row_mismatch}, table_name: #{@table_name}")

    ccs = compare_chunks(opts)
    if ccs.empty? and not row_mismatch
      @logger.debug("no chunk diff on #{@table_name}, skipping row diff")
      return Hash[]
    end

    row_diff = Hash[]
    ccs.each do |cc|
      rd = compare_rows(min: cc.master.min, max: cc.master.max)
      row_diff.merge! rd
    end

    # There are additional rows on the replica, probably with higher id values than the master has.
    if row_count_diff[:diff] < 0
      rd = compare_rows(min: @replica.min_row_id, max: @replica.max_row_id)
      row_diff.merge! rd
    end

    # Replica min is "lower" than Master min, so we need a deletion
    if (row_count_diff[:rmin] <=> row_count_diff[:mmin]) < 0
      @logger.debug("mismatch between rmin/mmin")
      rd = compare_rows(min: @replica.min_row_id, max: @master.min_row_id)
      row_diff.merge! rd
    end

    # Replica max is "greater" than Master max, so we need a deletion
    if (row_count_diff[:rmax] <=> row_count_diff[:mmax]) > 0
      @logger.debug("mismatch between mmax/rmax ")
      rd = compare_rows(min: @master.max_row_id, max: @replica.max_row_id)
      row_diff.merge! rd
    end

    @logger.debug("returning diff: #{row_diff}")

    row_diff
  rescue Exception => err
    @logger.error(%{skipping Table "#{@table_name}" because of error ("#{err.message}"): #{err.backtrace} })
    Hash[]
  end

  def compare_rows(opts = {})
    @logger.debug("comparing rows, opts: #{opts.inspect}")

    diff = Hash[]
    @master.row_checksum(opts).each do |cs|
      diff[cs.row_id] = RowComparison.from(master: cs)
    end

    @replica.row_checksum(opts).each do |cs|
      mcs = diff[cs.row_id]

      if mcs.nil?
        diff[cs.row_id] = RowComparison.from(replica: cs)
        next
      end

      if mcs.master.equal?(cs)
        diff.delete(cs.row_id)
        next
      end

      diff[cs.row_id].replica = cs
    end

    @logger.debug("Calculated RowChecksum diff: #{diff.inspect}")
    diff
  end

  def compare_chunks(opts = {})
    diff = []
    primary_key = @master.primary_key
    master_chunks = master_chunks(opts)
    @logger.debug("found #{master_chunks.count} master chunks")

    master_chunks.each do |mch|
      # This should only be one chunk, but it's technically a list, so we'll treat it like a list
      @logger.debug("master chunk: #{mch.inspect}")

      @replica.chunk_checksum(min: mch.min, max: mch.max).each do |rch|
        @logger.debug("replica chunk: #{rch.inspect}")
        @logger.debug("chunk counts: table => #{@table_name}, master => #{mch.count}, replica => #{rch.count}")

        next if rch.equal?(mch) # only keep checksums that are mismatched
        @logger.debug("chunk checksums misaligned on #{mch.inspect} and #{rch.inspect}")

        diff << ChunkComparison.new(
          master: mch,
          replica: rch,
          table_name: mch.table_name,
          primary_key: primary_key,
          min_row: mch.min,
          max_row: mch.max,
          visible_min_row: mch.visible_min,
          visible_max_row: mch.visible_max
        )
      end
    end

    if @logger.debug?
      summary = {
        table_name: @table_name,
        max_row_id: @master.max_row_id,
        min_row_id: @master.min_row_id,
        primary_key_columns: @master.primary_key,
        diff: diff
      }

      @logger.debug("Calculated ChunkComparison diff: #{summary.inspect}")
    end

    diff
  end

  def generate_update(row_id)
    delta = row_diff(row_id)

    return "" if delta.empty?

    table_name = if @database_name
      "#{@database_name}.#{@table_name}"
    else
      @table_name
    end

    # TODO - from here we need to call into the Table's QueryStrategy object to ensure our queries are consistent.
    primary_key = @master.primary_key
    @logger.debug("master Primary Key: '#{primary_key}'")

    # query_strategy = @master.conn.query_strategy(table_name: @table_name)
    # @logger.debug("master Query Strategy: '#{query_strategy}'")

    pairs = delta.filter { |k, v| !k.eql?(primary_key) }.map do |k, v|
      val = if v.nil?
        "NULL"
      else
        %('#{v}')
      end
      %(#{k} = #{val})
    end

    where_clause = @master.conn.where_clause(@table_name, row_id)
    @logger.debug("where_clause: #{where_clause}")

    %(UPDATE #{table_name} SET #{pairs.join(",\n\t\t")} #{where_clause}\n;)
  end

  private

  def row_count_diff
    mrc = @master.row_count
    rrc = @replica.row_count
    diff = "#{mrc}".to_i - "#{ rrc }".to_i
    @logger.debug("row counts: master => #{mrc}, replica => #{rrc}, diff => #{diff}")

    {
      table_name: @table_name,
      master: mrc,
      replica: rrc,
      diff: diff,
      mmax: @master.max_row_id,
      mmin: @master.min_row_id,
      rmax: @replica.max_row_id,
      rmin: @replica.min_row_id
    }
  end

  def master_chunks(_opts = {})
    chunks = []
    row_id = @master.min_row_id
    @logger.info("Min Row ID #{@master.min_row_id}; Max Row ID: #{@master.max_row_id}")
    loop do
      @master.chunk_checksum(min: row_id, limit: @chunk_size).each do |mch|
        chunks.unshift(mch)
      end
      @logger.debug("chunk count: #{chunks.count}")
      row_id = chunks.first.max
      @logger.debug("master chunks first max: #{chunks.first.max}, count: #{chunks.first.count}, @master.max_row_id: #{@master.max_row_id}")

      break if chunks.first.max.to_s == @master.max_row_id.to_s # stop if the max row ids match
    end

    chunks.reverse
  end

  # Find a per-column diff for a table pair
  def row_diff(row_id)
    @logger.debug("row_id: #{row_id}")
    master_row = @master.row_values(row_id).first
    replica_row = @replica.row_values(row_id).first
    delta = Hash[]
    master_row.each do |k, v|
      next if replica_row[k] == v
      delta[k] = v
    end
    delta
  end
end