// Copyright 2015 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,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package ddl_test

import (
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/tidb/ddl"
	"github.com/pingcap/tidb/meta"
	"github.com/pingcap/tidb/parser/model"
	"github.com/pingcap/tidb/sessionctx"
	"github.com/pingcap/tidb/testkit"
	"github.com/pingcap/tidb/util"
	"github.com/stretchr/testify/require"
)

const testLease = 5 * time.Second

func TestCheckOwner(t *testing.T) {
	_, dom := testkit.CreateMockStoreAndDomainWithSchemaLease(t, testLease)

	time.Sleep(testLease)
	require.Equal(t, dom.DDL().OwnerManager().IsOwner(), true)
	require.Equal(t, dom.DDL().GetLease(), testLease)
}

func TestInvalidDDLJob(t *testing.T) {
	store, dom := testkit.CreateMockStoreAndDomainWithSchemaLease(t, testLease)

	job := &model.Job{
		SchemaID:   0,
		TableID:    0,
		Type:       model.ActionNone,
		BinlogInfo: &model.HistoryInfo{},
		Args:       []interface{}{},
	}
	ctx := testNewContext(store)
	ctx.SetValue(sessionctx.QueryString, "skip")
	err := dom.DDL().DoDDLJob(ctx, job)
	require.Equal(t, err.Error(), "[ddl:8204]invalid ddl job type: none")
}

func TestAddBatchJobError(t *testing.T) {
	store, dom := testkit.CreateMockStoreAndDomainWithSchemaLease(t, testLease)
	ctx := testNewContext(store)

	require.Nil(t, failpoint.Enable("github.com/pingcap/tidb/ddl/mockAddBatchDDLJobsErr", `return(true)`))
	// Test the job runner should not hang forever.
	job := &model.Job{SchemaID: 1, TableID: 1}
	ctx.SetValue(sessionctx.QueryString, "skip")
	err := dom.DDL().DoDDLJob(ctx, job)
	require.Error(t, err)
	require.Equal(t, err.Error(), "mockAddBatchDDLJobsErr")
	require.Nil(t, failpoint.Disable("github.com/pingcap/tidb/ddl/mockAddBatchDDLJobsErr"))
}

func TestParallelDDL(t *testing.T) {
	store, dom := testkit.CreateMockStoreAndDomainWithSchemaLease(t, testLease)

	tk := testkit.NewTestKit(t, store)
	tk.MustExec("use test")

	/*
		build structure:
			DBs -> {
			 db1: test_parallel_ddl_1
			 db2: test_parallel_ddl_2
			}
			Tables -> {
			 db1.t1 (c1 int, c2 int)
			 db1.t2 (c1 int primary key, c2 int, c3 int)
			 db2.t3 (c1 int, c2 int, c3 int, c4 int)
			}
	*/
	tk.MustExec("create database test_parallel_ddl_1")
	tk.MustExec("create database test_parallel_ddl_2")
	tk.MustExec("create table test_parallel_ddl_1.t1(c1 int, c2 int, key db1_idx2(c2))")
	tk.MustExec("create table test_parallel_ddl_1.t2(c1 int primary key, c2 int, c3 int)")
	tk.MustExec("create table test_parallel_ddl_2.t3(c1 int, c2 int, c3 int, c4 int)")

	// set hook to execute jobs after all jobs are in queue.
	jobCnt := 11
	tc := &ddl.TestDDLCallback{Do: dom}
	once := sync.Once{}
	var checkErr error
	tc.OnJobRunBeforeExported = func(job *model.Job) {
		// TODO: extract a unified function for other tests.
		once.Do(func() {
			for {
				tk1 := testkit.NewTestKit(t, store)
				tk1.MustExec("begin")
				txn, err := tk1.Session().Txn(true)
				require.NoError(t, err)
				jobs, err := ddl.GetAllDDLJobs(tk1.Session(), meta.NewMeta(txn))
				require.NoError(t, err)
				tk1.MustExec("rollback")
				var qLen1, qLen2 int
				for _, job := range jobs {
					if !job.MayNeedReorg() {
						qLen1++
					} else {
						qLen2++
					}
				}
				if checkErr != nil {
					break
				}
				if qLen1+qLen2 == jobCnt {
					if qLen2 != 5 {
						checkErr = errors.Errorf("add index jobs cnt %v != 6", qLen2)
					}
					break
				}
				time.Sleep(5 * time.Millisecond)
			}
		})
	}

	once1 := sync.Once{}
	tc.OnGetJobBeforeExported = func(string) {
		once1.Do(func() {
			for {
				tk := testkit.NewTestKit(t, store)
				tk.MustExec("begin")
				txn, err := tk.Session().Txn(true)
				require.NoError(t, err)
				jobs, err := ddl.GetAllDDLJobs(tk.Session(), meta.NewMeta(txn))
				require.NoError(t, err)
				tk.MustExec("rollback")
				if len(jobs) == jobCnt {
					break
				}
				time.Sleep(time.Millisecond * 20)
			}
		})
	}
	dom.DDL().SetHook(tc)

	/*
		prepare jobs:
		/	job no.	/	database no.	/	table no.	/	action type	 /
		/     1		/	 	1			/		1		/	add index	 /
		/     2		/	 	1			/		1		/	add column	 /
		/     3		/	 	1			/		1		/	add index	 /
		/     4		/	 	1			/		2		/	drop column	 /
		/     5		/	 	1			/		1		/	drop index 	 /
		/     6		/	 	1			/		2		/	add index	 /
		/     7		/	 	2			/		3		/	drop column	 /
		/     8		/	 	2			/		3		/	rebase autoID/
		/     9		/	 	1			/		1		/	add index	 /
		/     10	/	 	2			/		null   	/	drop schema  /
		/     11	/	 	2			/		2		/	add index	 /
	*/
	var wg util.WaitGroupWrapper

	seqIDs := make([]int, 11)

	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_1.t1 add index db1_idx1(c1)")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[0], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_1.t1 add column c3 int")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[1], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_1.t1 add index db1_idxx(c1)")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[2], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_1.t2 drop column c3")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[3], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_1.t1 drop index db1_idx2")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[4], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_1.t2 add index db1_idx2(c2)")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[5], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_2.t3 drop column c4")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[6], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_2.t3 auto_id_cache 1024")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[7], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("alter table test_parallel_ddl_1.t1 add index db1_idx3(c2)")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[8], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		tk.MustExec("drop database test_parallel_ddl_2")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[9], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})
	time.Sleep(5 * time.Millisecond)
	wg.Run(func() {
		tk := testkit.NewTestKit(t, store)
		_ = tk.ExecToErr("alter table test_parallel_ddl_2.t3 add index db3_idx1(c2)")
		rs := tk.MustQuery("select json_extract(@@tidb_last_ddl_info, '$.seq_num')")
		seqIDs[10], _ = strconv.Atoi(rs.Rows()[0][0].(string))
	})

	wg.Wait()

	// Table 1 order.
	require.Less(t, seqIDs[0], seqIDs[1])
	require.Less(t, seqIDs[1], seqIDs[2])
	require.Less(t, seqIDs[2], seqIDs[4])
	require.Less(t, seqIDs[4], seqIDs[8])

	// Table 2 order.
	require.Less(t, seqIDs[3], seqIDs[5])

	// Table 3 order.
	require.Less(t, seqIDs[6], seqIDs[7])
	require.Less(t, seqIDs[7], seqIDs[9])
}

func TestJobNeedGC(t *testing.T) {
	job := &model.Job{Type: model.ActionAddIndex, State: model.JobStateCancelled}
	require.False(t, ddl.JobNeedGCForTest(job))

	job = &model.Job{Type: model.ActionAddColumn, State: model.JobStateDone}
	require.False(t, ddl.JobNeedGCForTest(job))
	job = &model.Job{Type: model.ActionAddIndex, State: model.JobStateDone}
	require.True(t, ddl.JobNeedGCForTest(job))
	job = &model.Job{Type: model.ActionAddPrimaryKey, State: model.JobStateDone}
	require.True(t, ddl.JobNeedGCForTest(job))
	job = &model.Job{Type: model.ActionAddIndex, State: model.JobStateRollbackDone}
	require.True(t, ddl.JobNeedGCForTest(job))
	job = &model.Job{Type: model.ActionAddPrimaryKey, State: model.JobStateRollbackDone}
	require.True(t, ddl.JobNeedGCForTest(job))

	job = &model.Job{Type: model.ActionMultiSchemaChange, State: model.JobStateDone, MultiSchemaInfo: &model.MultiSchemaInfo{
		SubJobs: []*model.SubJob{
			{Type: model.ActionAddColumn, State: model.JobStateDone},
			{Type: model.ActionRebaseAutoID, State: model.JobStateDone},
		}}}
	require.False(t, ddl.JobNeedGCForTest(job))
	job = &model.Job{Type: model.ActionMultiSchemaChange, State: model.JobStateDone, MultiSchemaInfo: &model.MultiSchemaInfo{
		SubJobs: []*model.SubJob{
			{Type: model.ActionAddIndex, State: model.JobStateDone},
			{Type: model.ActionAddColumn, State: model.JobStateDone},
			{Type: model.ActionRebaseAutoID, State: model.JobStateDone},
		}}}
	require.True(t, ddl.JobNeedGCForTest(job))
	job = &model.Job{Type: model.ActionMultiSchemaChange, State: model.JobStateDone, MultiSchemaInfo: &model.MultiSchemaInfo{
		SubJobs: []*model.SubJob{
			{Type: model.ActionAddIndex, State: model.JobStateDone},
			{Type: model.ActionDropColumn, State: model.JobStateDone},
			{Type: model.ActionRebaseAutoID, State: model.JobStateDone},
		}}}
	require.True(t, ddl.JobNeedGCForTest(job))
	job = &model.Job{Type: model.ActionMultiSchemaChange, State: model.JobStateRollbackDone, MultiSchemaInfo: &model.MultiSchemaInfo{
		SubJobs: []*model.SubJob{
			{Type: model.ActionAddIndex, State: model.JobStateRollbackDone},
			{Type: model.ActionAddColumn, State: model.JobStateRollbackDone},
			{Type: model.ActionRebaseAutoID, State: model.JobStateCancelled},
		}}}
	require.True(t, ddl.JobNeedGCForTest(job))
}