// Copyright 2016 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package main

import (
	"context"
	"database/sql"
	"fmt"
	"os"
	"strconv"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/ngaut/log"
	. "github.com/pingcap/check"
	"github.com/pingcap/tidb-enterprise-tools/pkg/filter"
	gmysql "github.com/siddontang/go-mysql/mysql"
	"github.com/siddontang/go-mysql/replication"
)

var _ = Suite(&testSyncerSuite{})

func TestSuite(t *testing.T) {
	TestingT(t)
}

type testSyncerSuite struct {
	db       *sql.DB
	syncer   *replication.BinlogSyncer
	streamer *replication.BinlogStreamer
	cfg      *Config
}

func (s *testSyncerSuite) SetUpSuite(c *C) {
	host := os.Getenv("MYSQL_HOST")
	if host == "" {
		host = "127.0.0.1"
	}
	port, _ := strconv.Atoi(os.Getenv("MYSQL_PORT"))
	if port == 0 {
		port = 3306
	}
	user := os.Getenv("MYSQL_USER")
	if user == "" {
		user = "root"
	}
	pswd := os.Getenv("MYSQL_PSWD")

	s.cfg = &Config{
		From: DBConfig{
			Host:     host,
			User:     user,
			Password: pswd,
			Port:     port,
		},
		ServerID: 101,
	}

	var err error
	dbAddr := fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8", s.cfg.From.User, s.cfg.From.Password, s.cfg.From.Host, s.cfg.From.Port)
	s.db, err = sql.Open("mysql", dbAddr)
	if err != nil {
		log.Fatal(err)
	}

	s.resetBinlogSyncer()

	_, err = s.db.Exec("SET GLOBAL binlog_format = 'ROW';")
	c.Assert(err, IsNil)
}

func (s *testSyncerSuite) resetBinlogSyncer() {
	var err error
	cfg := replication.BinlogSyncerConfig{
		ServerID:   uint32(s.cfg.ServerID),
		Flavor:     "mysql",
		Host:       s.cfg.From.Host,
		Port:       uint16(s.cfg.From.Port),
		User:       s.cfg.From.User,
		Password:   s.cfg.From.Password,
		UseDecimal: true,
	}
	if s.cfg.TimezoneStr != "" {
		timezone, err2 := time.LoadLocation(s.cfg.TimezoneStr)
		if err2 != nil {
			log.Fatal(err2)
		}
		cfg.TimestampStringLocation = timezone
	}
	pos := gmysql.Position{Name: "", Pos: 4}
	if s.syncer != nil {
		s.syncer.Close()
		pos = s.syncer.GetNextPosition()
	} else {
		s.resetMaster()
	}
	s.syncer = replication.NewBinlogSyncer(cfg)
	s.streamer, err = s.syncer.StartSync(pos)
	if err != nil {
		log.Fatal(err)
	}
}

func (s *testSyncerSuite) catchUpBinlog() {
	ch := make(chan interface{})
	ctx, cancel := context.WithCancel(context.Background())

	go func() {
		for {
			ev, _ := s.streamer.GetEvent(ctx)
			if ev == nil {
				return
			}
			ch <- struct{}{}
		}
	}()

	for {
		select {
		case <-ch:
			// do nothing
		case <-time.After(10 * time.Millisecond):
			cancel()
			return
		}
	}
}

func (s *testSyncerSuite) TearDownSuite(c *C) {
	s.db.Close()
}

func (s *testSyncerSuite) resetMaster() {
	s.db.Exec("reset master")
}

func (s *testSyncerSuite) clearRules() {
	s.cfg.DoDBs = nil
	s.cfg.DoTables = nil
	s.cfg.IgnoreDBs = nil
	s.cfg.IgnoreTables = nil
	s.cfg.SkipEvents = nil
	s.cfg.SkipDMLs = nil
}

func (s *testSyncerSuite) TestSelectDB(c *C) {
	s.cfg.DoDBs = []string{"~^b.*", "s1", "stest"}
	sqls := []string{
		"create database s1",
		"drop database s1",
		"create database s2",
		"drop database s2",
		"create database btest",
		"drop database btest",
		"create database b1",
		"drop database b1",
		"create database stest",
		"drop database stest",
		"create database st",
		"drop database st",
	}
	res := []bool{false, false, true, true, false, false, false, false, false, false, true, true}

	for _, sql := range sqls {
		s.db.Exec(sql)
	}

	parser, err := getParser(s.db, false)
	c.Assert(err, IsNil)

	syncer := NewSyncer(s.cfg)
	syncer.genRouter()
	syncer.genSkipDMLRules()
	i := 0
	for {
		if i >= len(sqls) {
			break
		}

		e, err := s.streamer.GetEvent(context.Background())
		c.Assert(err, IsNil)
		ev, ok := e.Event.(*replication.QueryEvent)
		if !ok {
			continue
		}
		sql := string(ev.Query)
		if syncer.skipQueryEvent(sql) {
			continue
		}

		tableNames, err := syncer.fetchDDLTableNames(sql, string(ev.Schema), parser)
		c.Assert(err, IsNil)
		r := syncer.skipQueryDDL(sql, tableNames[1])
		c.Assert(r, Equals, res[i])
		i++
	}
	s.clearRules()
	s.catchUpBinlog()
}

func (s *testSyncerSuite) TestSelectTable(c *C) {
	s.cfg.DoDBs = []string{"t2"}
	s.cfg.DoTables = []*filter.Table{
		{Schema: "stest", Name: "log"},
		{Schema: "stest", Name: "~^t.*"},
		{Schema: "~^ptest*", Name: "~^t.*"},
	}
	sqls := []string{
		"create database s1",
		"create table s1.log(id int)",

		"create table mysql.test(id int)",
		"drop table mysql.test",
		"create database stest",
		"create table stest.log(id int)",
		"create table stest.t(id int)",
		"create table stest.log2(id int)",
		"insert into stest.t(id) values (10)",
		"insert into stest.log(id) values (10)",
		"insert into stest.log2(id) values (10)",
		"drop table stest.log,stest.t,stest.log2",
		"drop database stest",

		"create database t2",
		"create table t2.log(id int)",
		"create table t2.log1(id int)",
		"drop table t2.log",
		"drop database t2",
		"create database ptest1",
		"create table ptest1.t1(id int)",

		"# this is comment \ncreate table ptest1.t2(id int)",
		"# this is comment \ndrop database ptest1",
		"# this is comment \ncreate table s1.log2(id int)",
		"# this is comment \ndrop database s1",
	}
	res := [][]bool{
		{true},
		{true},

		{true},
		{true},
		{false},
		{false},
		{false},
		{true},
		{false},
		{false},
		{true},
		{false, false, true},
		{false},

		{false},
		{false},
		{false},
		{false},
		{false},
		{false},
		{false},

		{false},
		{false},
		{true},
		{true},
	}

	for _, sql := range sqls {
		s.db.Exec(sql)
	}

	syncer := NewSyncer(s.cfg)
	syncer.genRouter()
	syncer.genSkipDMLRules()
	i := 0
	for {
		if i >= len(sqls) {
			break
		}

		e, err := s.streamer.GetEvent(context.Background())
		if err != nil {
			log.Fatal(err)
		}
		switch ev := e.Event.(type) {
		case *replication.QueryEvent:
			query := string(ev.Query)
			log.Debugf("i: %d, query:%+v", i, query)
			if syncer.skipQueryEvent(query) {
				continue
			}

			parser, err := getParser(s.db, false)
			c.Assert(err, IsNil)

			querys, err := resolveDDLSQL(query, parser)
			if err != nil {
				log.Fatalf("ResolveDDlSQL failed %v", err)
			}
			if len(querys) == 0 {
				continue
			}
			log.Debugf("querys:%+v", querys)
			for j, q := range querys {
				tableNames, err := syncer.fetchDDLTableNames(q, string(ev.Schema), parser)
				c.Assert(err, IsNil)
				r := syncer.skipQueryDDL(q, tableNames[1])
				c.Assert(r, Equals, res[i][j])
			}
		case *replication.RowsEvent:
			r := syncer.skipRowEvent(string(ev.Table.Schema), string(ev.Table.Table), e.Header.EventType)
			c.Assert(r, Equals, res[i][0])

		default:
			continue
		}

		i++

	}
	s.clearRules()
	s.catchUpBinlog()
}

func (s *testSyncerSuite) TestIgnoreDB(c *C) {
	s.cfg.IgnoreDBs = []string{"~^b.*", "s1", "stest"}
	sqls := []string{
		"create database s1",
		"drop database s1",
		"create database s2",
		"drop database s2",
		"create database btest",
		"drop database btest",
		"create database b1",
		"drop database b1",
		"create database stest",
		"drop database stest",
		"create database st",
		"drop database st",
	}
	res := []bool{true, true, false, false, true, true, true, true, true, true, false, false}

	for _, sql := range sqls {
		s.db.Exec(sql)
	}

	parser, err := getParser(s.db, false)
	c.Assert(err, IsNil)

	syncer := NewSyncer(s.cfg)
	syncer.genRouter()
	syncer.genSkipDMLRules()
	i := 0
	for {
		if i >= len(sqls) {
			break
		}

		e, err := s.streamer.GetEvent(context.Background())
		if err != nil {
			log.Fatal(err)
		}
		ev, ok := e.Event.(*replication.QueryEvent)
		if !ok {
			continue
		}
		sql := string(ev.Query)
		if syncer.skipQueryEvent(sql) {
			continue
		}
		tableNames, err := syncer.fetchDDLTableNames(sql, string(ev.Schema), parser)
		c.Assert(err, IsNil)
		r := syncer.skipQueryDDL(sql, tableNames[1])
		c.Assert(r, Equals, res[i])
		i++
	}
	s.clearRules()
	s.catchUpBinlog()
}

func (s *testSyncerSuite) TestIgnoreTable(c *C) {
	s.cfg.IgnoreDBs = []string{"t2"}
	s.cfg.IgnoreTables = []*filter.Table{
		{Schema: "stest", Name: "log"},
		{Schema: "stest", Name: "~^t.*"},
	}
	sqls := []string{
		"create database s1",
		"create table s1.log(id int)",
		"drop database s1",

		"create table mysql.test(id int)",
		"drop table mysql.test",
		"create database stest",
		"create table stest.log(id int)",
		"create table stest.t(id int)",
		"create table stest.log2(id int)",
		"insert into stest.t(id) values (10)",
		"insert into stest.log(id) values (10)",
		"insert into stest.log2(id) values (10)",
		"drop table stest.log,stest.t,stest.log2",
		"drop database stest",

		"create database t2",
		"create table t2.log(id int)",
		"create table t2.log1(id int)",
		"drop table t2.log",
		"drop database t2",
	}
	res := [][]bool{
		{false},
		{false},
		{false},

		{true},
		{true},
		{true},
		{true},
		{true},
		{false},
		{true},
		{true},
		{false},
		{true, true, false},
		{true},

		{true},
		{true},
		{true},
		{true},
		{true},
	}

	for _, sql := range sqls {
		s.db.Exec(sql)
	}

	syncer := NewSyncer(s.cfg)
	syncer.genRouter()
	syncer.genSkipDMLRules()
	i := 0
	for {
		if i >= len(sqls) {
			break
		}
		e, err := s.streamer.GetEvent(context.Background())
		if err != nil {
			log.Fatal(err)
		}
		switch ev := e.Event.(type) {
		case *replication.QueryEvent:
			query := string(ev.Query)
			if syncer.skipQueryEvent(query) {
				continue
			}

			parser, err := getParser(s.db, false)
			c.Assert(err, IsNil)

			querys, err := resolveDDLSQL(query, parser)
			if err != nil {
				log.Fatalf("ResolveDDlSQL failed %v", err)
			}
			if len(querys) == 0 {
				continue
			}

			for j, q := range querys {
				tableNames, err := syncer.fetchDDLTableNames(q, string(ev.Schema), parser)
				c.Assert(err, IsNil)
				r := syncer.skipQueryDDL(q, tableNames[1])
				c.Assert(r, Equals, res[i][j])
			}
		case *replication.RowsEvent:
			r := syncer.skipRowEvent(string(ev.Table.Schema), string(ev.Table.Table), e.Header.EventType)
			c.Assert(r, Equals, res[i][0])

		default:
			continue
		}

		i++

	}
	s.clearRules()
	s.catchUpBinlog()
}

func (s *testSyncerSuite) TestSkipDML(c *C) {
	s.cfg.SkipDMLs = []*SkipDML{
		{
			Schema: "",
			Table:  "",
			Type:   "update",
		}, {
			Schema: "foo",
			Table:  "",
			Type:   "delete",
		}, {
			Schema: "foo1",
			Table:  "bar1",
			Type:   "delete",
		},
	}

	sqls := []struct {
		sql     string
		isDML   bool
		skipped bool
	}{
		{"drop database if exists foo", false, false},
		{"create database foo", false, false},
		{"create table foo.bar(id int)", false, false},
		{"insert into foo.bar values(1)", true, false},
		{"update foo.bar set id=2", true, true},
		{"delete from foo.bar where id=2", true, true},
		{"drop database if exists foo1", false, false},
		{"create database foo1", false, false},
		{"create table foo1.bar1(id int)", false, false},
		{"insert into foo1.bar1 values(1)", true, false},
		{"update foo1.bar1 set id=2", true, true},
		{"delete from foo1.bar1 where id=2", true, true},
		{"drop database if exists foo", false, false},
		{"drop database if exists foo1", false, false},
	}

	res := make([]bool, 0, len(sqls))
	for _, t := range sqls {
		if t.isDML {
			res = append(res, t.skipped)
		}
		s.db.Exec(t.sql)
	}

	syncer := NewSyncer(s.cfg)
	syncer.genRouter()
	syncer.genSkipDMLRules()

	c.Logf("skip rules %+v", syncer.skipDMLRules)

	i := 0
	for {
		if i >= len(res) {
			break
		}
		e, err := s.streamer.GetEvent(context.Background())
		if err != nil {
			log.Fatal(err)
		}
		switch ev := e.Event.(type) {
		case *replication.QueryEvent:
		case *replication.RowsEvent:
			r := syncer.skipRowEvent(string(ev.Table.Schema), string(ev.Table.Table), e.Header.EventType)
			c.Logf("%s, ev %v", buildTableName(string(ev.Table.Schema), string(ev.Table.Table)), e.Header.EventType)
			c.Assert(r, Equals, res[i])
			i++
		default:
		}
	}
	s.clearRules()
	s.catchUpBinlog()
}

func (s *testSyncerSuite) TestTimezone(c *C) {
	s.cfg.DoDBs = []string{"~^tztest_.*"}

	createSQLs := []string{
		"create database if not exists tztest_1",
		"create table if not exists tztest_1.t_1(id int, a timestamp)",
	}

	testCases := []struct {
		sqls     []string
		timezone string
	}{
		{
			[]string{
				"insert into tztest_1.t_1(id, a) values (1, '1990-04-15 01:30:12')",
				"insert into tztest_1.t_1(id, a) values (2, '1990-04-15 02:30:12')",
				"insert into tztest_1.t_1(id, a) values (3, '1990-04-15 03:30:12')",
			},
			"Asia/Shanghai",
		},
		{
			[]string{
				"insert into tztest_1.t_1(id, a) values (4, '1990-04-15 01:30:12')",
				"insert into tztest_1.t_1(id, a) values (5, '1990-04-15 02:30:12')",
				"insert into tztest_1.t_1(id, a) values (6, '1990-04-15 03:30:12')",
			},
			"America/Phoenix",
		},
	}
	queryTs := "select unix_timestamp(a) from `tztest_1`.`t_1` where id = ?"

	dropSQLs := []string{
		"drop table tztest_1.t_1",
		"drop database tztest_1",
	}

	for _, sql := range createSQLs {
		s.db.Exec(sql)
	}

	for _, testCase := range testCases {
		s.cfg.TimezoneStr = testCase.timezone
		syncer := NewSyncer(s.cfg)
		syncer.genRouter()
		s.resetBinlogSyncer()

		// we should not use `sql.DB.Exec` to do query which depends on session variables
		// because `sql.DB.Exec` will choose a underlying Conn for every query from the connection pool
		// and different Conn using different session
		// ref: `sql.DB.Conn`
		// and `set @@global` is also not reasonable, because it can not affect sessions already exist
		// if we must ensure multi queries use the same session, we should use a transaction
		txn, err := s.db.Begin()
		c.Assert(err, IsNil)
		txn.Exec("set @@session.time_zone = ?", testCase.timezone)
		txn.Exec("set @@session.sql_mode = ''")
		for _, sql := range testCase.sqls {
			_, err = txn.Exec(sql)
			c.Assert(err, IsNil)
		}
		err = txn.Commit()
		c.Assert(err, IsNil)

		location, err := time.LoadLocation(testCase.timezone)
		c.Assert(err, IsNil)

		idx := 0
		for {
			if idx >= len(testCase.sqls) {
				break
			}
			e, err := s.streamer.GetEvent(context.Background())
			c.Assert(err, IsNil)
			switch ev := e.Event.(type) {
			case *replication.RowsEvent:
				rowid := ev.Rows[0][0].(int32)
				var ts sql.NullInt64
				err2 := s.db.QueryRow(queryTs, rowid).Scan(&ts)
				c.Assert(err2, IsNil)
				c.Assert(ts.Valid, IsTrue)

				raw := ev.Rows[0][1].(string)
				data, err := time.ParseInLocation("2006-01-02 15:04:05", raw, location)
				c.Assert(err, IsNil)
				c.Assert(data.Unix(), DeepEquals, ts.Int64)
				idx++
			default:
				continue
			}
		}
	}

	for _, sql := range dropSQLs {
		s.db.Exec(sql)
	}
	s.catchUpBinlog()
}