package clusterproxy

import (
	"context"

	certmanagerv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1"
	"github.com/jetstack/cert-manager/pkg/apis/certmanager/v1alpha2"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless/dm"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	v1 "k8s.io/api/core/v1"
)

type AuthMethod string

const (
	AuthMethodPassword AuthMethod = "password"
	AuthMethodToken    AuthMethod = "token"
)

type ClusterOperator struct {
	Username   string
	AuthMethod AuthMethod
	Credential string // TOKEN, PASSWORD
}

type CdcInfo struct {
	ID    string   `json:"id"`
	State string   `json:"state"`
	CpTso uint64   `json:"checkpoint_tso"`
	Error CdcError `json:"error"`
}

type CdcError struct {
	Code    string `json:"code"`
	Message string `json:"message"`
}

type Safepoint struct {
	SP uint64 `json:"gc_safe_point"`
}

type AuditCheckModel struct {
	ExternalID     string `json:"externalID"`
	ServiceAccount string `json:"serviceAccount"`
	Audit          struct {
		BucketURL   string `json:"bucketUrl"`
		Region      string `json:"region"`
		RoleARN     string `json:"roleArn"`
		TidbRoleARN string `json:"tidbRoleArn"`
		SASToken    string `json:"sasToken"`
	} `json:"audit"`
}

// AuditCheckModelV2 is copied from tidbcloud/cluster-service-ng/internal/service-proxy/pkg/checker/core/types.go
// to avoid import.
type AuditCheckModelV2 struct {
	// TargetType should always be "audit".
	TargetType string       `json:"targetType"`
	Audit      *AuditParams `json:"audit,omitempty"`
}

type AuditParams struct {
	BucketURL string `json:"bucketUrl,omitempty"`
	Region    string `json:"region,omitempty"`

	AWS      *AWSAuth      `json:"aws,omitempty"`
	GCP      *GCPAuth      `json:"gcp,omitempty"`
	Azure    *AzureAuth    `json:"azure,omitempty"`
	AliCloud *AliCloudAuth `json:"alicloud,omitempty"`
}

type AWSAuth struct {
	RoleARN         string `json:"roleArn,omitempty"`
	ExternalID      string `json:"externalID,omitempty"`
	AccessKeyID     string `json:"access_key_id,omitempty"`
	SecretAccessKey string `json:"secret_access_key,omitempty"`
}

type GCPAuth struct {
	ServiceAccount string `json:"serviceAccount,omitempty"`
}

type AzureAuth struct {
	SASToken string `json:"sasToken,omitempty"`
}

type AliCloudAuth struct {
	RoleARN         string `json:"roleArn,omitempty"`
	ExternalID      string `json:"externalID,omitempty"`
	RoleSessionName string `json:"roleSessionName,omitempty"`
}

type AuditCheckResp struct {
	Pass    bool   `json:"pass"`
	Message string `json:"message"`
}

type KafkaItem struct {
	ReplicationFactor       uint32 `json:"replication_factor"`
	PartitionNum            uint32 `json:"partition_num"`
	SchemaRegistryEndpoints string `json:"schema_registry_endpoints"`
}

type S3Parameter struct {
	S3Uri           string `json:"s3_uri"`
	AccessKeyId     string `json:"access_key_id"`
	SecretAccessKey string `json:"secret_access_key"`
	RoleArn         string `json:"role_arn"`
	ExternalID      string `json:"external_id"`
}

type BigQueryParameter struct {
	ProjectID string `json:"project_id"`
	DatasetID string `json:"dataset_id"`
}

type GcsParameter struct {
	GcsUrl                    string `json:"gcs_url"`
	Region                    string `json:"region"`
	ImpersonateServiceAccount string `json:"impersonate_service_account"`
}

type MysqlParameter struct {
	// PEM encoded bytes
	SSLCABytes   []byte `json:"ssl_ca_bytes"`
	SSLCertBytes []byte `json:"ssl_cert_bytes"`
	SSLKeyBytes  []byte `json:"ssl_key_bytes"`
}

type ConnectionModel struct {
	Type       string            `json:"type"`
	User       string            `json:"user"`
	Password   string            `json:"password"`
	Url        string            `json:"url"`
	Topic      string            `json:"topic"`
	Database   string            `json:"database"`
	AuthType   string            `json:"auth_type"`
	EnableTls  bool              `json:"enable_tls"`
	MysqlParam MysqlParameter    `json:"mysql_param"`
	KafkaItem  KafkaItem         `json:"kafka_item"`
	S3Param    S3Parameter       `json:"s3_param"`
	GCSParam   GcsParameter      `json:"gcs_param"`
	BQParam    BigQueryParameter `json:"bigquery_param"`
}

type TableInfo struct {
	Schema string `json:"schema"`
	Table  string `json:"table"`
}

type DispatchRule struct {
	Matcher   []string `json:"matcher,omitempty"`
	Topic     string   `json:"topic,omitempty"`
	Partition string   `json:"partition,omitempty"`
	IndexName string   `json:"index,omitempty"`
	Columns   []string `json:"columns,omitempty"`
}

type SinkConfig struct {
	Dispatchers    []DispatchRule `json:"dispatchers,omitempty"`
	Protocol       string         `json:"protocol,omitempty"`
	CaseSensitive  bool           `json:"case-sensitive,omitempty"`
	EnableOldValue bool           `json:"enable-old-value,omitempty"`
	SchemaRegistry string         `json:"schema-registry,omitempty"`
	Terminator     string         `json:"terminator,omitempty"`
	DateSeparator  string         `json:"date-separator,omitempty"`
	Csv            *CsvConfig     `json:"csv,omitempty"`
}

type CsvConfig struct {
	Delimiter            string `json:"delimiter,omitempty"`
	Quote                string `json:"quote,omitempty"`
	Null                 string `json:"null,omitempty"`
	IncludeCommitTs      bool   `json:"include-commit-ts,omitempty"`
	BinaryEncodingMethod string `json:"binary_encoding_method,omitempty"`
}

type CdcCreateModel struct {
	Sink                  string     `json:"sink_uri,omitempty"`
	StartTS               uint64     `json:"start_ts,omitempty"`
	ChangefeedID          string     `json:"changefeed_id,omitempty"`
	ForceReplicate        bool       `json:"force_replicate"`
	IgnoreIneligibleTable bool       `json:"ignore_ineligible_table,omitempty"`
	FilterRules           []string   `json:"filter_rules,omitempty"`
	MounterWorkerNum      int        `json:"mounter_worker_num,omitempty"`
	Config                SinkConfig `json:"sink_config,omitempty"`
}

type CdcUpdateModel struct {
	TargetTs         uint64     `json:"target_ts,omitempty"`
	SinkUri          string     `json:"sink_uri,omitempty"`
	FilterRules      []string   `json:"filter_rules,omitempty"`
	IgnoreTxnStartTs uint64     `json:"ignore_txn_start_ts,omitempty"`
	MounterWorkerNum int        `json:"mounter_worker_num,omitempty"`
	SinkConfig       SinkConfig `json:"sink_config,omitempty"`
}

// Ref: https://github.com/tikv/pd/blob/fa18b03821adc7aa19bfca45f346f12d23e164fb/server/api/cluster.go#L41-L43
type PdClusterModel struct {
	Id           uint64 `json:"id,omitempty"`
	MaxPeerCount uint32 `json:"max_peer_count,omitempty"`
}

type SqlError struct {
	ErrNumber  uint16 `json:"errNumber"` // the error number returned from tidb
	ErrMessage string `json:"errMessage"`
}

type QuerySQLResponse struct {
	SqlError

	Result []byte `json:"result"`
}

type ExecuteResponse struct {
	SqlError
}

type K8sResourceRequest struct {
	// Required: used by all operations to select Create, Get, GetDirectly,
	// List, Delete, Update, or UpdateStatus.
	OperateType string `json:"operateType"`
	// Required: used by all operations to identify the target resource version.
	APIVersion string `json:"apiVersion"`
	// Required: used by all operations to identify the target resource kind.
	Kind string `json:"kind"`
	// Required: used by all operations to scope the target resource.
	Namespace string `json:"namespace"`
	// Required by Get, GetDirectly, Delete, and UpdateStatus.
	// Optional for Create and Update: only used when object.metadata.name is
	// empty.
	Name string `json:"name"`
	// Optional: only used to filter List requests.
	Labels map[string]string `json:"labels"`
	// Optional: only used to filter List requests. It currently only supports
	// status.phase on cloud.pingcap.com/v1 Branch resources.
	FieldSelector map[string]string `json:"fieldSelector"`
	// Required by Create and Update: the full resource body.
	Object any `json:"object"`
	// Required by UpdateStatus: the status subresource payload.
	Status any `json:"status"`
}

// ClusterProxy to make some operation on the cluster.
type ClusterProxy interface {
	// Deprecated: use ExecuteSQLV2 to instead
	ExecuteSQL(ctx context.Context, pwd encrypt.Secret, sql string) error
	ExecuteSQLEachTiDB(ctx context.Context, pwd encrypt.Secret, sql string) error
	// Deprecated: use ExecuteMultiSQLV2 to instead
	ExecuteMultiSQL(ctx context.Context, pwd encrypt.Secret, sqls []string) error
	CreateCloudAdmin(ctx context.Context, pwd encrypt.Secret) error
	// Deprecated: use ChangePasswordV2 to instead
	ChangePassword(ctx context.Context, pwd encrypt.Secret, newPwd string) error
	// Deprecated: use QuerySQLV2 to instead
	QuerySQL(ctx context.Context, pwd encrypt.Secret, sql string) ([]byte, error)

	ExecuteSQLV2(ctx context.Context, operator *ClusterOperator, sql string) error
	ExecuteMultiSQLV2(ctx context.Context, operator *ClusterOperator, sqls []string) error
	ChangePasswordV2(ctx context.Context, operator *ClusterOperator, username, newPwd string) error
	QuerySQLV2(ctx context.Context, operator *ClusterOperator, sql string) ([]byte, error)

	/* Note: The following methods which have 'WithSqlError' suffix, will DON'T raise an error if some errors happen but the connection between
	cluster-proxy and target tidb is OK.  For example, the 'operator' is unauthorized to connect to db, '1141' will be placed
	into ErrNumber. So, you should always BOTH check the 'error' and parse the response to ensure this request was executed successfully.
	*/
	ExecuteSQLWithSqlError(ctx context.Context, operator *ClusterOperator, sql string) (*ExecuteResponse, error)
	ExecuteMultiSQLWithSqlError(ctx context.Context, operator *ClusterOperator, sqls []string) (*ExecuteResponse, error)
	QuerySQLWithSqlError(ctx context.Context, operator *ClusterOperator, sql string) (*QuerySQLResponse, error)
	CallK8sResource(ctx context.Context, req K8sResourceRequest) (any, error)

	CreateCDCTask(ctx context.Context, model CdcCreateModel) error
	UpdateCDCTask(ctx context.Context, taskID string, model CdcUpdateModel) error
	GetCDCTaskInfo(ctx context.Context) ([]CdcInfo, error)
	PauseCDCTask(ctx context.Context, taskID string) error
	ResumeCDCTask(ctx context.Context, taskID string) error
	DeleteCDCTask(ctx context.Context, taskID string) error
	TestConnection(ctx context.Context, model *ConnectionModel) error
	GetSourceTables(ctx context.Context, model *ConnectionModel) ([]TableInfo, error)
	GetPDSafepoint(ctx context.Context) (*Safepoint, error)
	TestAuditConnection(ctx context.Context, input AuditCheckModel) (*AuditCheckResp, error)
	TestAuditConnectionV2(ctx context.Context, input AuditCheckModelV2) (*AuditCheckResp, error)
	CreateChangefeedV2(ctx context.Context, model ChangefeedConfig) error
	UpdateChangefeedV2(ctx context.Context, taskID string, model ChangefeedConfig) error
	GetChangefeedListV2(ctx context.Context) ([]ChangefeedCommonInfo, error)
	PauseChangefeedV2(ctx context.Context, taskID string) error
	ResumeChangefeedV2(ctx context.Context, taskID string, resumeConfig ResumeChangefeedConfig) error
	DeleteChangefeedV2(ctx context.Context, taskID string) error
	GetChangefeedV2(ctx context.Context, taskID string) (*ChangeFeedInfo, error)

	// Serverless methods:
	ListServerlessBackups(ctx context.Context, req serverless.ListBackupRequest) (*serverless.ListBackupResponse, error)
	RestoreServerlessCluster(ctx context.Context, req serverless.RestoreRequest) error
	GetServerlessRestoreProgress(ctx context.Context, token string, restoreId uint64) (*serverless.RestoreProgressResponse, error)
	// Called when restore task succeed or failed with no retry.
	CleanServerlessRestoreTask(ctx context.Context, token string, restoreId uint64) error
	IsServerlessTokenAllowedForBR(ctx context.Context, token string) (bool, error)
	// Serverless TiCDC
	CreateServerlessCDC(ctx context.Context, keyspaceId uint32) error
	DeleteServerlessCDC(ctx context.Context, keyspaceId uint32) error
	CreateServerlessChangefeed(ctx context.Context, keyspaceId uint32, req *serverless.ChangefeedConfig) (*serverless.Changefeed, error)
	GetServerlessChangefeed(ctx context.Context, keyspaceId uint32, changefeedId string) (*serverless.Changefeed, error)
	ListServerlessChangefeeds(ctx context.Context, keyspaceId uint32, state *serverless.ChangefeedState) (*serverless.ListChangefeedsResponse, error)
	DeleteServerlessChangefeed(ctx context.Context, keyspaceId uint32, changefeedId string) error
	UpdateServerlessChangefeed(ctx context.Context, keyspaceId uint32, changefeedId string, req *serverless.ChangefeedConfig) (*serverless.Changefeed, error)
	PauseServerlessChangefeed(ctx context.Context, keyspaceId uint32, changefeedId string) error
	ResumeServerlessChangefeed(ctx context.Context, keyspaceId uint32, changefeedId string, req *serverless.ResumeChangefeedConfig) error
	// Serverless PD
	GetServerlessGCSafePoint(ctx context.Context, keyspaceId uint32) (*serverless.GCSafePointV2, error)

	CreateServerlessDMSource(ctx context.Context, req *dm.CreateSourceRequest) (*dm.Source, error)
	GetServerlessDMSource(ctx context.Context, name string) (*dm.Source, error)
	DeleteServerlessDMSource(ctx context.Context, name string) error
	CreateServerlessDMTask(ctx context.Context, req *dm.CreateTaskRequest) (*dm.OperateTaskResponse, error)
	DeleteServerlessDMTask(ctx context.Context, name string) error
	StartServerlessDMTask(ctx context.Context, name string, reqg *dm.StartTaskRequest) error
	StopServerlessDMTask(ctx context.Context, name string, req *dm.StopTaskRequest) error
	QueryServerlessDMTaskStatus(ctx context.Context, name string, sourceNameList []string) (*dm.GetTaskStatusResponse, error)
	ListServerlessDMWorkers(ctx context.Context) (*dm.GetClusterWorkerListResponse, error)

	// open websocket connection
	OpenWebsocket(ctx context.Context, operator *ClusterOperator) (WebsocketConn, error)

	TiCDCClusterInterface
	ChangefeedInterface
}

type TiCDCClusterInterface interface {
	// cdc cluster API:
	CreateTiCDCCR(ctx context.Context, cr *v1alpha1.TidbCluster) error
	UpdateTiCDCCR(ctx context.Context, cr *v1alpha1.TidbCluster) error
	DeleteTiCDCCR(ctx context.Context, name string, namespace string) error
	QueryTiCDCCR(ctx context.Context, name string, namespace string) (*v1alpha1.TidbCluster, error)
	ListTiCDCNode(ctx context.Context, label map[string]string) (*v1.NodeList, error)
	ListTiCDCPod(ctx context.Context, label map[string]string) (*v1.PodList, error)
	DeleteTiCDCPvc(ctx context.Context, name string, namespace string, label map[string]string) error
	ListTiCDCPvc(ctx context.Context, namespace string, label map[string]string) (*v1.PersistentVolumeClaimList, error)
	CreateTiCDCCertV1alpha2(ctx context.Context, cert *v1alpha2.Certificate) error
	DeleteTiCDCCertV1alpha2(ctx context.Context, name string, namespace string) error
	QueryTiCDCCertV1alpha2(ctx context.Context, name string, namespace string) (*v1alpha2.Certificate, error)
	CreateTiCDCCertV1(ctx context.Context, cert *certmanagerv1.Certificate) error
	DeleteTiCDCCertV1(ctx context.Context, name string, namespace string) error
	QueryTiCDCCertV1(ctx context.Context, name string, namespace string) (*certmanagerv1.Certificate, error)
}

type ChangefeedInterface interface {
	CreateChangefeed(ctx context.Context, crName string, model CdcCreateModel) error
	UpdateChangefeed(ctx context.Context, crName string, taskID string, model CdcUpdateModel) error
	GetChangefeedList(ctx context.Context, crName string) ([]CdcInfo, error)
	PauseChangefeed(ctx context.Context, crName string, changefeedID string) error
	ResumeChangefeed(ctx context.Context, crName string, changefeedID string) error
	DeleteChangefeed(ctx context.Context, crName string, changefeedID string) error

	CreateChangefeed_v2(ctx context.Context, crName string, model ChangefeedConfig) error
	UpdateChangefeed_v2(ctx context.Context, crName string, changefeedID string, model ChangefeedConfig) error
	GetChangefeedList_v2(ctx context.Context, crName string) ([]ChangefeedCommonInfo, error)
	PauseChangefeed_v2(ctx context.Context, crName string, changefeedID string) error
	ResumeChangefeed_v2(ctx context.Context, crName string, changefeedID string, resumeConfig ResumeChangefeedConfig) error
	DeleteChangefeed_v2(ctx context.Context, crName string, changefeedID string) error
	GetChangefeed_v2(ctx context.Context, crName string, changefeedID string) (*ChangeFeedInfo, error)
}

// A websocket connection is aligned with a database connection.
// The interface is not concurrent safe.
type WebsocketConn interface {
	Query(ctx context.Context, sql string) ([]byte, error)
	BatchQuery(ctx context.Context, sqls []string) ([][]byte, error)
	Exec(ctx context.Context, sql string) error
	Close() error
}