require 'flydata/fluent-plugins/mysql/binlog_query_handler'

module Mysql

class DdlQueryHandler < BinlogQueryHandler
  DDL_TABLE_QUERY = /^(?:(?:ALTER|CREATE|DROP|RENAME) +(?:\w+ +)*TABLE +([^ ]+)|TRUNCATE +(?:TABLE +)?([^ ;]+))/i

  def emit_record(type, record)
    # ddl event record doesn't have "table_name"
    record['table_name'] = table_info(record)[:table_name]
    super do |opt|
      yield opt
    end
  end

  def acceptable_db?(record)
    supported_database == table_info(record)[:db_name]
  end

  def table_info(record)
    table_info = { db_name: record["db_name"], table_name: nil }
    if DDL_TABLE_QUERY =~ record["normalized_query"]
      table_name_in_query = ($1 ? $1 : $2).tr("`", "")

      if (idx = table_name_in_query.index("."))
        table_info[:db_name] = table_name_in_query[0...idx]
        table_info[:table_name] = table_name_in_query[idx+1..-1]
      else
        table_info[:table_name] = table_name_in_query
      end
    end
    table_info
  end
end

class TableDdlQueryHandler < DdlQueryHandler
end

class DatabaseDdlQueryHandler < DdlQueryHandler
  def emit_record(type, record)
    return unless acceptable_db?(record)

    check_empty_binlog

    opt = {}
    records = yield(opt) # The block may set options as necessary
    return if records.nil? # skip
    records = [records] unless records.kind_of?(Array)

    database = records.first[DB_NAME] || record['db_name']

    return unless acceptable_event?(type)

    # Add common information to each record
    records.each do |r|
      r[TYPE] = type
      r[RESPECT_ORDER] = true
      r[SRC_POS] = binlog_pos(record)
      r[V] = FlydataCore::Record::V2
    end

    # Use binlog's timestamp
    timestamp = record["timestamp"].to_i
    records.each do |row|
      Fluent::Engine.emit(@context.tag, timestamp, row)
    end
  end
end

end