#!/usr/bin/env ruby

require 'bundler/setup'
require 'mysql2'
require 'socket'
require 'pp'
require "getoptlong"

Thread.abort_on_exception = false

opts = GetoptLong.new(
  ["--password",      "-p", GetoptLong::REQUIRED_ARGUMENT],
  ["--rehome-master",   "-r", GetoptLong::NO_ARGUMENT],
  ["--start-slave",   "-s", GetoptLong::NO_ARGUMENT],
)

opts.each do |opt, arg|
  case opt
  when '--password'
    $password = arg
  when '--rehome-master'
    $rehome_master = true
  when '--start-slave'
    $start_slave = true
    $rehome_master = true
  end
end

def usage
  puts "Usage: master_cut OLD_MASTER NEW_MASTER ADMIN_USERNAME"
  puts "                  [-p,--password PASSWORD]"
  puts "                  [-r,--rehome-master]"
  puts "                  [-s,--start-slave]"
  exit false
end

$old_master, $new_master, $username = *ARGV
unless $old_master && $new_master && $username
  usage
end


def open_cx(host)
  host, port = host.split(":")
  port = port.to_i if port
  Mysql2::Client.new(:host => host, :username => $username, :password => $password, :port => port)
end

def set_rw(cx)
  cx.query("SET GLOBAL READ_ONLY=0")
end

def set_ro(cx)
  cx.query("SET GLOBAL READ_ONLY=1")
end

$swapped_ok = false

def fail(reason)
  puts "Failed preflight check: #{reason}"
  exit false
end

def ask_for_password
  return unless $password.nil?

  $stdout.write("Password for #{$username}: ")
  begin
    system "stty -echo"
    $password = $stdin.gets.chomp
  ensure
    system "stty echo"
  end
end

def preflight_check
  cx = open_cx($old_master)
  rw = cx.query("select @@read_only as read_only").first['read_only']
  fail("old-master #{$old_master} is read-only!") if rw != 0

  slave_cx = open_cx($new_master)
  rw = slave_cx.query("select @@read_only as read_only").first['read_only']
  fail("new-master #{$old_master} is read-write!") if rw != 1

  slave_info = slave_cx.query("show slave status").first
  fail("no slave configured!") if slave_info.nil?
  fail("slave is stopped!") unless slave_info['Slave_IO_Running'] == 'Yes' && slave_info['Slave_SQL_Running'] == 'Yes'
  fail("slave is delayed") if slave_info['Seconds_Behind_Master'].nil? || slave_info['Seconds_Behind_Master'] > 0

  masters_slave_info = cx.query("show slave status").first
  if $rehome_master && (masters_slave_info.nil? || masters_slave_info['Master_User'] == 'test')
    fail("I can't rehome the original master -- it has no slave user or password.")
  end

  master_ip, slave_master_ip = [$old_master, slave_info['Master_Host']].map do |h|
    h = h.split(':').first
    Socket.gethostbyname(h)[3].unpack("CCCC")
  end

  if master_ip != slave_master_ip
    fail("slave does not appear to be replicating off master! (master: #{master_ip.join('.')}, slave's master: #{slave_master_ip.join('.')})")
  end
end

def process_kill_thread
  Thread.new do
    cx = open_cx($old_master)
    sleep 5
    while !$swapped_ok
      my_id = cx.query("SELECT CONNECTION_ID() as id").first['id']
      processlist = cx.query("show processlist")
      processlist.each do |process|
        next if process['Info'] =~ /SET GLOBAL READ_ONLY/
        next if process['Id'].to_i == my_id.to_i
        puts "killing #{process}"
        kill_query!(cx, process['Id'])
      end
      sleep 0.1
    end
  end
end

def kill_query!(cx, id)
  begin
    cx.query("kill #{id}")
  rescue Mysql2::Error => e
    raise e unless e.errno == 1094 # unknown thread id error
  end
end

def swap_thread
  Thread.new do
    master = open_cx($old_master)
    slave = open_cx($new_master)
    set_ro(master)
    slave.query("STOP SLAVE")
    new_master_info = slave.query("show master status").first
    set_rw(slave)
    $swapped_ok = true
    puts "Swapped #{$old_master} and #{$new_master}"
    puts "New master information at time of swap: "
    pp new_master_info
    if $rehome_master
      rehome_master(new_master_info, $start_slave)
    end
    exit
  end
end

def rehome_master(info, start_slave)
  puts "Reconfiguring #{$old_master} to be a slave of #{$new_master}..."
  host, port = $new_master.split(":")
  port_clause = port ? "master_port = #{port}," : ""
  cx = open_cx($old_master)
  cx.query("change master to master_host='#{host}', #{port_clause} master_log_file = '#{info['File']}', master_log_pos=#{info['Position']}")
  cx.query("START SLAVE") if start_slave
end

ask_for_password
preflight_check

threads = []
threads << swap_thread
threads << process_kill_thread
threads.each(&:join)

rehome_master