package configchange

import (
	"context"
	"fmt"
	"os"
	"strconv"

	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/tidbcloud"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	config2 "github.com/tidbcloud/infra-provider/pkg/config"
)

var _ SQLExecutor = &clusterProxySQLExecutor{}

type clusterProxySQLExecutor struct {
	operator *clusterproxy.ClusterOperator
	proxy    clusterproxy.ClusterProxy
}

func newClusterProxySQLExecutor(ctx context.Context, cluster *cloudv1.Cluster, cloudAdminPasswd string, conf config2.RootConfig) (*clusterProxySQLExecutor, error) {
	proxy, err := getClusterProxy(ctx, cluster, conf)
	if err != nil {
		return nil, err
	}
	return &clusterProxySQLExecutor{
		operator: &clusterproxy.ClusterOperator{
			Username:   DBUserCloudAdmin,
			AuthMethod: clusterproxy.AuthMethodPassword,
			Credential: cloudAdminPasswd,
		},
		proxy: proxy,
	}, nil
}

func getClusterProxy(ctx context.Context, cluster *cloudv1.Cluster, conf config2.RootConfig) (clusterproxy.ClusterProxy, error) {
	clusterID, err := strconv.ParseUint(cluster.Name, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parse clusterID(%s): %w", cluster.Name, err)
	}

	proxyAuth0ClientId, ok1 := os.LookupEnv("PROXY_AUTH0_CLIENT_ID")
	if !ok1 {
		return nil, fmt.Errorf("clusterProxy auth0 clientID not found")
	}
	proxyAuth0ClientSecret, ok2 := os.LookupEnv("PROXY_AUTH0_CLIENT_SECRET")
	if !ok2 {
		return nil, fmt.Errorf("clusterProxy auth0 clientSecret not found")
	}
	proxyBaseUrl := cluster.Status.Proxy.Endpoint
	p, err := tidbcloud.NewClusterProxyV2(ctx, tidbcloud.ClusterProxyInput{
		PcaARN: "", // todo(chenfei): it seems not necessary
		Cfg: tidbcloud.ProxyConfig{
			Auth0Domain:       conf.Network.Proxy.Auth0Domain,
			Auth0ClientID:     proxyAuth0ClientId,
			Auth0ClientSecret: proxyAuth0ClientSecret,
			Auth0Audience:     conf.Network.Proxy.Auth0Audience,
		},
		ClusterID: clusterID,
		BaseURL:   proxyBaseUrl,
		Timeout:   10,
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "get cluster proxy failed")
	}
	return p, nil
}

func (e *clusterProxySQLExecutor) Execute(ctx context.Context, sql string) error {
	result, err := e.proxy.ExecuteSQLWithSqlError(ctx, e.operator, sql)
	if err != nil {
		return fmt.Errorf("execute sql: %s", err)
	}
	return wrapClusterProxyErr(result.SqlError)
}

func (e *clusterProxySQLExecutor) Query(ctx context.Context, sql string) ([]byte, error) {
	resp, err := e.proxy.QuerySQLWithSqlError(ctx, e.operator, sql)
	if err != nil {
		return nil, fmt.Errorf("query sql: %s", err)
	}
	return resp.Result, wrapClusterProxyErr(resp.SqlError)
}

type sqlExecuteError struct {
	Code    uint16
	Message string
}

func (e *sqlExecuteError) Error() string {
	return fmt.Sprintf("execute sql error, code: %d, message: %s", e.Code, e.Message)
}

func wrapClusterProxyErr(sqlError clusterproxy.SqlError) error {
	if sqlError.ErrNumber == 0 {
		return nil
	}
	return &sqlExecuteError{
		Code:    sqlError.ErrNumber,
		Message: sqlError.ErrMessage,
	}
}