package dbctrl

import (
	"database/sql"
	"errors"
	"validate/util"
)

/*
 * Refer to :
 * - https://docs.pingcap.com/zh/tidb/v7.1/sql-statement-show-placement-for
 * - https://docs.pingcap.com/zh/tidb/v7.1/placement-rules-in-sql
 * - https://docs.pingcap.com/zh/tidb/v7.1/sql-statement-show-placement#%E8%AF%AD%E6%B3%95%E5%9B%BE
 */

type PlacementPolicy struct {
	POLICY_ID            int64
	CATALOG_NAME         string
	POLICY_NAME          string
	PRIMARY_REGION       string
	REGIONS              string
	CONSTRAINTS          string
	LEADER_CONSTRAINTS   string
	FOLLOWER_CONSTRAINTS string
	LEARNER_CONSTRAINTS  string
	SCHEDULE             string
	FOLLOWERS            int64
	LEARNERS             int64

	CreateStmt         string
	DatabaseList       []*DatabaseInfo
	TableList          []*TableInformationSchema
	PartitionTableList []*PartitionTableInformationSchema
}

const ShowCreatePlacementPolicyStmt = "SHOW CREATE PLACEMENT POLICY %s;"
const ShowAllPlacementPolicyStmt = "select * from information_schema.placement_policies"

const ShowAllDatabasePlacementPolicyStmt = "SELECT * FROM information_schema.SCHEMATA WHERE TIDB_PLACEMENT_POLICY_NAME IS NOT NULL;"
const ShowAllTablePlacementPolicyStmt = "SELECT * FROM information_schema.tables WHERE tidb_placement_policy_name IS NOT NULL;"
const ShowAllPartitionTablePlacementPolicyStmt = "SELECT * FROM information_schema.partitions WHERE tidb_placement_policy_name IS NOT NULL;"

func getPlacementPolicyCreateStmt(tx *sql.Tx, pp *PlacementPolicy) (*PlacementPolicy, error) {
	if tx == nil || pp == nil {
		return nil, errors.New("empty transaction object")
	}

	rows, err := tx.Query(ShowCreatePlacementPolicyStmt)
	if err != nil {
		return nil, err
	}
	for rows.Next() {
		var policyName string
		var policyCreateStmt string
		if err = rows.Scan(
			&policyName,
			&policyCreateStmt,
		); err != nil {
			util.Logger.Fatalf(err.Error())
		}

		pp.CreateStmt = policyCreateStmt
	}

	return pp, nil
}

func GetAllPlacementPolicy(tx *sql.Tx) ([]*PlacementPolicy, error) {
	if tx == nil {
		return nil, errors.New("empty transaction object")
	}

	var placementRuleList []*PlacementPolicy

	rows, err := tx.Query(ShowAllPlacementPolicyStmt)
	if err != nil {
		return nil, err
	}

	for rows.Next() {
		var policy = &PlacementPolicy{}
		if err = rows.Scan(
			&policy.POLICY_ID,
			&policy.CATALOG_NAME,
			&policy.POLICY_NAME,
			&policy.PRIMARY_REGION,
			&policy.REGIONS,
			&policy.CONSTRAINTS,
			&policy.LEADER_CONSTRAINTS,
			&policy.FOLLOWER_CONSTRAINTS,
			&policy.LEARNER_CONSTRAINTS,
			&policy.SCHEDULE,
			&policy.FOLLOWERS,
			&policy.LEARNERS,
		); err != nil {
			util.Logger.Fatalf(err.Error())
		}

		placementRuleList = append(placementRuleList, policy)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}

	for _, policy := range placementRuleList {
		policy, err = getPlacementPolicyCreateStmt(tx, policy)
	}

	return placementRuleList, nil
}

func GetAllDatabaseWithPlacementRule(tx *sql.Tx) ([]*DatabaseInfo, error) {
	return GetDatabaseByStmt(tx, ShowAllDatabasePlacementPolicyStmt)
}

func GetAllTableWithPlacementRule(tx *sql.Tx) ([]*TableInformationSchema, error) {
	return GetInformationSchemaTableByStmt(tx, ShowAllTablePlacementPolicyStmt)
}

func GetAllPartitionTableWithPlacementRule(tx *sql.Tx) ([]*PartitionTableInformationSchema, error) {
	return GetInformationSchemaPartitionTableByStmt(tx, ShowAllPartitionTablePlacementPolicyStmt)
}