package deploypkg

import (
	"context"
	"database/sql"
	"fmt"
	"net/url"
	"strings"
	"time"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/log"
	logsearch "github.com/pingcap/test-infra/sdk/pkg/logs"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

type Cluster interface {
	GetDB(dbName string) (*sql.DB, error)
	ServiceURL(typ resource.TiDBAddrType) (*url.URL, error)
	PeerURLs(typ resource.TiDBAddrType) ([]*url.URL, error)
	SearchLog(ctx context.Context, opts resource.TiDBClusterSearchLogOptions) ([]string, error)
}

var auditSqls = [][]string{
	{"drop database if exists `audit`", "DropDatabase"},
	{"create database if not exists `audit`", "CreateDatabase"},
	{"use `audit`", "Use"},
	{"Create table t1(a int, b varchar(20))", "CreateTable"},
	{"Insert into t1 values (10, 'a'), (11, 'b')", "Insert"},
	{"Select * from t1", "Select"},
	{"Select * from t1 where a = 10", "Select"},
	{"Update t1 set b = 'c' where a = 11", "Update"},
	{"Delete from t1 where b = 'a'", "Delete"},
	{"Alter table t1 add column c date", "AlterTable"},
	{"Create Index a on t1(a)", "CreateIndex"},
	{"Create view v as select * from t1", "CreateView"},
	{"Analyze table t1", "AnalyzeTable"},
}

func RunSearchLog(ctx context.Context, tc Cluster, startTime time.Time, sqls [][]string, audited bool, redacted bool) {
	// wait for audit log record
	log.Info("waiting 10s to check the audit log.")
	time.Sleep(10 * time.Second)
	// 10m is long enough to execute sql and do not limit the end time for search log.
	endTime := startTime.Add(10 * time.Minute)

	var fetchLog []string
	for _, sql := range sqls {
		if sql[1] == "Use" {
			// only check table_access
			// fetchLog = "SQL_TEXT=\"use `audit`\""
			continue
		}

		if redacted {
			fetchLog = []string{fmt.Sprintf("SQL_STATEMENTS=%s", sql[1])}
			if strings.Contains(sql[0], "t1") {
				fetchLog = append(fetchLog, "t1")
			}
		} else {
			s := sql[0]
			s = strings.Replace(s, "(", `\(`, -1)
			s = strings.Replace(s, ")", `\)`, -1)
			s = strings.Replace(s, `*`, `\*`, -1)
			fetchLog = []string{fmt.Sprintf("%s", s), "SQL_TEXT"}
		}

		log.Info("execute and check audit log:", zap.Bool("audited", audited), zap.Bool("redacted", redacted), zap.String("sql", sql[0]), zap.Strings("pattern", fetchLog), zap.Time("start_time", startTime), zap.Time("end_time", endTime))
		logopt := resource.TiDBClusterSearchLogOptions{
			Target:    resource.TiDB,
			Levels:    []logsearch.LogLevel{logsearch.LogLevelInfo},
			StartTime: startTime,
			EndTime:   endTime,
			Patterns:  fetchLog,
			Limit:     10,
		}

		// FIXME: seems search log may return nil since there are audit log record.
		if res, err := tc.SearchLog(ctx, logopt); err == nil {
			log.Info("query result:", zap.Strings("res", res))
			Expect(len(res) > 0).Should(Equal(audited))
		} else {
			log.Error("Faild to get audit log: ", zap.Error(err))
		}
	}
}

func RunAndCheckAuditLog(ctx context.Context, tc Cluster, sqls [][]string, db *sql.DB, audited bool, redacted bool) {
	// sleep for Eliminate the impact of other operations on the log
	log.Info("waiting 5s to get the start time.")
	time.Sleep(5 * time.Second)
	startTime := time.Now()
	log.Info("waiting 5s to start execute sql.")
	time.Sleep(5 * time.Second)
	for _, sql := range sqls {
		s := strings.TrimSpace(sql[0])
		log.Info("execute:", zap.String("sql", s))
		Try(db.Exec(s))
	}
	RunSearchLog(ctx, tc, startTime, sqls, audited, redacted)
}

func GetDBWithUser(tc resource.Addressable, dbName, user string) (*sql.DB, error) {
	addr, err := tc.ServiceURL(resource.DBAddr)
	if err != nil {
		return nil, err
	}
	addr.User = url.User(user)

	db, err := sql.Open("mysql", resource.BuildDSN(addr, dbName, "timeout=2m"))
	if err != nil {
		return nil, err
	}
	if err := db.Ping(); err != nil {
		return nil, err
	}
	return db, nil
}

func DownloadPlugin(node resource.WorkloadNode, plugin_url string) {
	if strings.HasSuffix(plugin_url, "tar.gz") {
		downloadplugin := fmt.Sprintf("mkdir -p /root/plugins && cd /root/plugins && curl -LO %s", plugin_url)
		if _, _, _, err := node.Exec(resource.WorkloadNodeExecOptions{Command: downloadplugin}); err != nil {
			log.L().Fatal("Faild to download plugins", zap.Error(err))
		}

		files := strings.Split(plugin_url, "/")
		filename := files[len(files)-1]
		if _, _, _, err := node.Exec(resource.WorkloadNodeExecOptions{Command: fmt.Sprintf("cd /root/plugins && tar zxf %s", filename)}); err != nil {
			log.L().Fatal("Faild to extract plugins", zap.Error(err))
		}
	} else {
		for _, plugin := range [2]string{"audit-1", "whitelist-1"} {
			downloadplugin := fmt.Sprintf("mkdir -p /root/plugins/bin/  && cd /root/plugins/bin && curl -LO %s%s.so", plugin_url, plugin)
			log.Info("Execute command:", zap.String("command", downloadplugin))
			_, _, _, err := node.Exec(resource.WorkloadNodeExecOptions{Command: downloadplugin})
			if err != nil {
				log.Info("Failed to download plugin: %v", zap.String("plugin", err.Error()))
			}
		}
	}
}

func CheckClusterStatus(tc Cluster) {
	log.Info("Start check cluster status")
	time.Sleep(3 * time.Second)
	db, err := tc.GetDB("test")
	log.Info("connect the db.")
	if err != nil {
		log.Error("error getting db.", zap.String("GetDB", err.Error()))
	}
	err = db.Ping()
	Expect(err).Should(Succeed())
	log.Info("cluster started successfully.")
}

func CheckAuditLogForMatcherUser(ctx context.Context, tc Cluster, db *sql.DB) {
	Try(db.Exec("insert into mysql.tidb_audit_table_access (user, db, tbl, access_type) values ('test', '.*', '.*', ''), ('^qa$', '.*', '.*', '');"))
	Try(db.Exec("flush tidb plugins audit;"))

	db0, _ := GetDBWithUser(tc, "test", "test1")
	RunAndCheckAuditLog(ctx, tc, auditSqls, db0, true, true)

	db1, _ := GetDBWithUser(tc, "test", "test2")
	RunAndCheckAuditLog(ctx, tc, auditSqls, db1, true, true)

	db3, _ := GetDBWithUser(tc, "test", "qa")
	RunAndCheckAuditLog(ctx, tc, auditSqls, db3, true, true)
}

func CheckAuditLogDisable(ctx context.Context, tc Cluster, db *sql.DB) {
	Try(db.Exec("insert into mysql.tidb_audit_table_access (user, db, tbl, access_type) values ('.*', '.*', '.*', '');"))
	Try(db.Exec("flush tidb plugins audit;"))
	Try(db.Exec("admin plugins disable audit;"))
	db0, _ := GetDBWithUser(tc, "test", "test1")
	RunAndCheckAuditLog(ctx, tc, auditSqls, db0, false, true)
}

func CheckAuditLogForNoMatcherUser(ctx context.Context, tc Cluster, db *sql.DB) {

	Try(db.Exec("insert into mysql.tidb_audit_table_access (user, db, tbl, access_type) values ('^qa$', '.*', '.*', '');"))
	Try(db.Exec("flush tidb plugins audit;"))

	db0, _ := GetDBWithUser(tc, "test", "test1")
	RunAndCheckAuditLog(ctx, tc, auditSqls, db0, false, true)

	db3, _ := GetDBWithUser(tc, "test", "qa1")
	RunAndCheckAuditLog(ctx, tc, auditSqls, db3, false, true)
}

func CheckRedactAuditLog(ctx context.Context, tc Cluster, db *sql.DB) {

	Try(db.Exec("insert into mysql.tidb_audit_table_access (user, db, tbl, access_type) values ('.*', '.*', '.*', '');"))
	log.Info("set global tidb_audit_redact_log = off;")
	Try(db.Exec("set global tidb_audit_redact_log = off;"))
	Try(db.Exec("flush tidb plugins audit;"))
	RunAndCheckAuditLog(ctx, tc, auditSqls, db, true, false)

	log.Info("set global tidb_audit_redact_log = on;")
	Try(db.Exec("set global tidb_audit_redact_log = on;"))
	Try(db.Exec("flush tidb plugins audit;"))
	RunAndCheckAuditLog(ctx, tc, auditSqls, db, true, true)
}

func Must(err error) {
	if err != nil {
		Fail(err.Error())
	}
}

func ShowPluginInfo(db *sql.DB, pluginName string, status string) {
	var Name, Status, Type, Library, License, Version string
	sql := fmt.Sprintf("show plugins like '%s';", pluginName)
	rows, err := db.Query(sql)
	if err != nil {
		log.Info("show plugins failed", zap.Error(err))
	}
	for rows.Next() {
		rows.Scan(&Name, &Status, &Type, &Library, &License, &Version)
		log.Info("plugin status", zap.String("plugin_name", Name), zap.String("status", Status))
		Expect(Status).Should(Equal(status))
	}
}

func AdminPlugin(db *sql.DB) {
	db.Exec("admin plugins disable audit;")
	db.Exec("admin plugins disable whitelist;")
	time.Sleep(1 * time.Second)
	ShowPluginInfo(db, "whitelist", "Ready-disable")
	ShowPluginInfo(db, "audit", "Ready-disable")

	db.Exec("admin plugins enable audit;")
	db.Exec("admin plugins enable whitelist;")
	time.Sleep(1 * time.Second)
	ShowPluginInfo(db, "whitelist", "Ready-enable")
	ShowPluginInfo(db, "audit", "Ready-enable")
}

func CheckFlushAudigPluginInTxn(ctx context.Context, tc Cluster, db *sql.DB) {
	txn, err := db.Begin()
	Expect(err).Should(BeNil())
	Try(txn.Exec("insert into mysql.tidb_audit_table_access (user, db, tbl, access_type) values ('.*', '.*', '.*', '');"))
	Try(txn.Exec("flush tidb plugins audit;"))
	Try(txn.Commit())
	RunAndCheckAuditLog(ctx, tc, auditSqls, db, true, true)
}

func CheckFlushAudigPluginOutTxn(ctx context.Context, tc Cluster, db *sql.DB) {
	txn, err := db.Begin()
	Expect(err).Should(BeNil())
	Try(txn.Exec("insert into mysql.tidb_audit_table_access (user, db, tbl, access_type) values ('.*', '.*', '.*', '');"))
	Try(txn.Commit())
	Try(db.Exec("flush tidb plugins audit;"))
	RunAndCheckAuditLog(ctx, tc, auditSqls, db, true, true)
}

func CheckGeneralAuditLog(ctx context.Context, tc Cluster, event string) {

	db, _ := tc.GetDB("test")
	Try(db.Exec("create database if not exists test02;"))
	Try(db.Exec("use test02"))
	Try(db.Exec("drop table if exists t1;"))
	Try(db.Exec("drop view if exists v1;"))
	Try(db.Exec("create sequence if not exists seq;"))
	Try(db.Exec("create table t1(a int, b int, unique index (b));"))
	Try(db.Exec("create view v1 as select * from t1 where b > 1;"))
	// to make the log search easier.
	Try(db.Exec("set global tidb_audit_redact_log=false"))
	Try(db.Exec(fmt.Sprintf("set global tidb_audit_log_general_events = '%s'", event)))
	Try(db.Exec("flush tidb plugins audit;"))
	db.Close()
	time.Sleep(10 * time.Second)
	db1, _ := tc.GetDB("test")
	defer db1.Close()
	generalLogs := []string{
		// "flush status;", //table_access
		// "FLUSH TABLES;",             //table_access
		// "flush tidb plugins audit;", // table_access
		// "SHOW BACKUPS;",              //table_access
		// "SHOW RESTORES;",             //table_access
		// "SHOW CONFIG;",               // table_access
		// "SHOW CREATE SEQUENCE seq;",  // table_access
		// "SHOW STATS_HEALTHY;",        //table_access
		// "show stats_histograms;",     //table_access
		// "show stats_meta;",           //table_access
		// "show table t1 next_row_id;", //table_access
		"use `test02`",
		"show tables",
		"drop stats t1",
		"create user if not exists gu1",
		"create role if not exists gr1",
		"grant gr1 to gu1", // table_access
		"set default role gr1 to gu1",
		"drop user gu1",
		"drop role gr1",
		"BEGIN;",
		"START TRANSACTION;",
		"COMMIT;",
		"ROLLBACK;",
		"SHOW GRANTS;",
		"SHOW PROCESSLIST;",
		"show analyze status;",
		"SHOW SESSION BINDINGS;",
		"SHOW BUILTINS;",
		"SHOW CHARACTER SET;",
		"SHOW COLLATION;",
		"show columns from v1;",
		"show fields from v1;",
		"SHOW CREATE USER 'root';",
		"SHOW DATABASES;",
		"SHOW ENGINES;",
		"SHOW ERRORS;",
		"SHOW INDEXES FROM t1;",
		"SHOW MASTER STATUS;",
		"SHOW PLUGINS;",
		"show privileges;",
		"SHOW PROFILES;",
		"SHOW PUMP STATUS;",
		"SHOW SCHEMAS;",
		"show status;",
		"SHOW TABLE STATUS LIKE 't1';",
		"SHOW TABLES;",
		"SHOW VARIABLES;",
		"SHOW WARNINGS;",
		"SET @number = 5;",
		"SET NAMES utf8;",
		"SET CHARACTER SET utf8mb4;",
		"SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;",
		"SET SESSION sql_mode = 'STRICT_TRANS_TABLES'",
		"DO SLEEP(1);",
		"SELECT 1;",
		"EXPLAIN SELECT 1;",
		"DESC SELECT 1;",
		"DESCRIBE SELECT 1;",
		"show table t1 regions;",
		"EXPLAIN ANALYZE SELECT a, b FROM t1 WHERE a = 1;",
		"EXPLAIN SELECT a, b FROM t1;",
		"PREPARE mystmt FROM 'SELECT ? as num FROM DUAL';",
		"EXECUTE mystmt USING @number;",
		"DEALLOCATE PREPARE mystmt;",
		"split table t1 between (0) and (10000000) regions 16",
		"split table t1 between (0) and (10000000) regions 16", // let STARTING can serch the split table log
	}

	for id, sql := range generalLogs {
		startTime := time.Now()
		Try(db1.Exec(sql))
		log.Info("Execute", zap.String("sql", sql))
		var fetchSql string
		if event == "STARTING" {
			if id > 0 {
				fetchSql = generalLogs[id-1]
			}
		} else {
			fetchSql = sql
		}
		// prepare and execute mystmt
		if strings.Contains(fetchSql, "mystmt ") {
			fetchSql = "FROM DUAL"
		}

		// grant is table_access, for set default role
		if strings.HasPrefix(fetchSql, "grant") {
			continue
		}
		res := searchGeneralLog(ctx, tc, fetchSql, startTime)
		switch event {
		case "STARTING":
			Expect(len(res) > 0).Should(BeTrue(), event, fetchSql)
		case "COMPLETED":
			Expect(len(res) > 0).Should(BeTrue(), event, fetchSql)
		case "ERROR":
			Expect(len(res) > 0).Should(BeFalse(), event, fetchSql)
		}
	}
}

func CheckGeneralAuditLogFail(ctx context.Context, tc Cluster) {
	db, _ := tc.GetDB("test")
	Try(db.Exec("set global tidb_audit_log_general_events = 'ERROR';"))
	Try(db.Exec("create database if not exists test01"))
	db.Close()
	db, _ = tc.GetDB("test01")
	errGeneralLogs := []string{
		"use `test1`",

		// user and role do not exists
		"create user gu2",
		"create role gr2",
		"set default role gr2 to gu2",
		"drop user gu2",
		"drop role gr2",

		// table and view do not exists
		"drop stats t1",
		"show columns from v1;",
		"show fields from v1;",
		"SHOW CREATE USER 'root1';",
		"SHOW INDEXES FROM t1;",
		"SHOW TABLE STATUS LIKE 't1';",
		"SET SESSION sql_mode = 'STRICT_TRANS_TABLE'",
		"show table t1 regions;",
		"EXPLAIN ANALYZE SELECT a, b FROM t1 WHERE a = 1;",
		"EXPLAIN SELECT a, b FROM t1;",
		"PREPARE mystmt1 FROM 'SELECT ? as num FROM t1';",
		"EXECUTE mystmt1 USING @number;",
		"DEALLOCATE PREPARE mystmt1;",
		"split table t1 between (0) and (10000000) regions 16",
	}
	for _, sql := range errGeneralLogs {
		startTime := time.Now()
		if _, err := db.Exec(sql); err != nil {
			if strings.Contains(sql, "mystmt1 ") {
				sql = "as num FROM t1"
			}
			res := searchGeneralLog(ctx, tc, sql, startTime)
			Expect(len(res) > 0).Should(BeTrue(), "ERROR", sql)
		}
	}
}

func searchGeneralLog(ctx context.Context, tc Cluster, fetchSql string, startTime time.Time) []string {
	var fetchLog []string
	fetchSql = strings.Replace(fetchSql, "(", `\(`, -1)
	fetchSql = strings.Replace(fetchSql, ")", `\)`, -1)
	fetchSql = strings.Replace(fetchSql, `*`, `\*`, -1)
	fetchLog = []string{fetchSql, "EVENT_CLASS=GENERAL"}
	logopt := resource.TiDBClusterSearchLogOptions{
		Target:    resource.TiDB,
		Levels:    []logsearch.LogLevel{logsearch.LogLevelInfo},
		StartTime: startTime,
		EndTime:   startTime.Add(10 * time.Minute),
		Patterns:  fetchLog,
		Limit:     10,
	}
	log.Info("search log", zap.Strings("log", fetchLog), zap.Time("start_time", startTime))
	res, err := tc.SearchLog(ctx, logopt)
	if err != nil {
		log.Error("Faild to get audit log: ", zap.Error(err))
	}
	log.Info("query result:", zap.Strings("res", res))
	return res
}