package agent

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent"
	infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
	"google.golang.org/genproto/googleapis/rpc/status"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/types/known/emptypb"
)

// the timeout of frontend is 30s, so we set the timeout to 27s
const defaultTimeout = 27 * time.Second

var _ AgentClient = &AgentClientImpl{}

func GetAgentClient(ctx context.Context, cluster *core.Cluster) (AgentClient, error) {
	return NewRegionalAgentClient(ctx, cluster.Id, cluster.InfraProviderType, cluster.RegionName)
}

type AgentClient interface {
	// access
	CheckS3Access(ctx context.Context, req *agent.CheckS3AccessReq) (*agent.CheckS3AccessResp, error)
	CheckGcsAccess(ctx context.Context, req *agent.CheckGcsAccessReq) (*agent.CheckGcsAccessResp, error)
	CheckAzblobAccess(ctx context.Context, req *agent.CheckAzblobAccessReq) (*agent.CheckAzblobAccessResp, error)
	PrecheckImportJob(ctx context.Context, req *agent.PrecheckImportJobReq) (*agent.PrecheckImportJobResp, error)
	// lightning
	LightningPrecheck(ctx context.Context, req *agent.LightningPrecheckReq) (*agent.LightningPrecheckResp, error)

	// cluster or source
	ExecSQLOnCluster(ctx context.Context, req *agent.ExecSQLReq) (*agent.ExecSQLResp, error)
	LoadDataToCluster(ctx context.Context, req *agent.LoadDataReq) (*agent.LoadDataResp, error)
	CheckDBConnectivity(ctx context.Context, req *agent.CheckDBConnectivityReq) (*agent.CheckDBConnectivityResp, error)
	GetSourceTables(ctx context.Context, req *agent.CheckDBConnectivityReq) (*agent.GetSourceTablesResp, error)
	GetTiDBServerSecret(ctx context.Context, req *agent.GetTiDBServerSecretReq) (*agent.GetTiDBServerSecretResp, error)
	GetClusterClientSecret(ctx context.Context, req *agent.GetClusterClientSecretReq) (*agent.GetClusterClientSecretResp, error)
	GetClusterTables(ctx context.Context, req *agent.Cluster) (*agent.GetClusterTablesResp, error)
	PreviewCsvFile(ctx context.Context, req *agent.PreviewCsvFileReq) (*agent.PreviewCsvFileResp, error)
	SearchImportFiles(ctx context.Context, req *agent.SearchImportFilesReq) (*agent.SearchImportFilesResp, error)
	PrecheckImportFile(ctx context.Context, req *agent.PrecheckImportFileReq) (*emptypb.Empty, error)
	DropClusterTables(ctx context.Context, req *agent.DropClusterTablesReq) (*emptypb.Empty, error)

	// GetEngineClient get engine client
	GetEngineClient(engineAddr string) *EngineClient
	GetLightningClient(clusterId, importID uint64) LightningAPI
	GetDMOpenapiClient(clusterID uint64, dmAddr string) *DMOpenapiClient
	PrecheckDMJob(ctx context.Context, req *agent.PrecheckDMJobReq) (*agent.PrecheckDMJobResp, error)
	GetSourceBinlogStatus(ctx context.Context, req *agent.CheckDBConnectivityReq) (*agent.GetSourceBinlogStatusResp, error)
	SetHttpTimeout(time.Duration)
	GetUserFileContent(ctx context.Context, req *agent.GetUserFileReq) (*agent.GetUserFileResp, error)
	NsLookup(ctx context.Context, req *agent.NsLookupReq) (*agent.NsLookupResp, error)

	// DO Request
	FetchResponseWithHeader(ctx context.Context,
		method, path string,
		req any,
		extraHeaders map[string]string,
		errorUnwrapper func(logger *zap.Logger, bodyBytes []byte) error,
	) ([]byte, error)
}

type LightningAPI interface {
	GetTaskProgress(ctx context.Context) (*TaskProgress, error)
}

type AgentClientImpl struct {
	BaseURL      string
	ProviderType string

	Client HttpRequester
}

func (a *AgentClientImpl) SetHttpTimeout(t time.Duration) {
	a.Client.SetHttpTimeout(t)
}

func (a *AgentClientImpl) getSchema() string {
	if a.ProviderType == infraapi.ProviderTypeEKS || a.ProviderType == infraapi.ProviderTypeGKE {
		return "http"
	}
	return "https"
}

func (a *AgentClientImpl) CheckS3Access(ctx context.Context, req *agent.CheckS3AccessReq) (*agent.CheckS3AccessResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/s3/check_access", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.CheckS3AccessResp
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) CheckGcsAccess(ctx context.Context, req *agent.CheckGcsAccessReq) (*agent.CheckGcsAccessResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/gcs/check_access", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.CheckGcsAccessResp
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, err
}

func (a *AgentClientImpl) CheckAzblobAccess(ctx context.Context, req *agent.CheckAzblobAccessReq) (*agent.CheckAzblobAccessResp, error) {
	return nil, ErrUnSupportedOperation
}

func (a *AgentClientImpl) PrecheckImportJob(ctx context.Context, req *agent.PrecheckImportJobReq) (*agent.PrecheckImportJobResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/orgs/%s/projects/%s/clusters/%s/imports:precheck", a.getSchema(), a.BaseURL, req.OrgId, req.ProjectId, req.ClusterId)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.PrecheckImportJobResp
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) LightningPrecheck(ctx context.Context, req *agent.LightningPrecheckReq) (*agent.LightningPrecheckResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/import/lightning-precheck", a.getSchema(), a.BaseURL)
	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.LightningPrecheckResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) getResponse(ctx context.Context, method, url string, req any) ([]byte, error) {
	return a.getResponseWithHeader(ctx, method, url, req, nil, defaultErrorUnwrapper)
}

func defaultErrorUnwrapper(logger *zap.Logger, bodyBytes []byte) error {
	var errResp status.Status
	if err := protojson.Unmarshal(bodyBytes, &errResp); err == nil {
		switch codes.Code(errResp.Code) {
		case codes.PermissionDenied:
			return errcode.NewForbiddenErr(errors.New(errResp.Message))
		case codes.DeadlineExceeded:
			return context.DeadlineExceeded
		default:
			return errors.New(errResp.Message)
		}
	} else {
		logger.Error("failed to unmarshal response body as grpc Status", zap.Error(err))
	}
	return nil
}

func (a *AgentClientImpl) FetchResponseWithHeader(ctx context.Context,
	method, path string,
	req any,
	extraHeaders map[string]string,
	errorUnwrapper func(logger *zap.Logger, bodyBytes []byte) error,
) ([]byte, error) {
	targetURL := fmt.Sprintf("%s://%s/%s", a.getSchema(), a.BaseURL, path)
	return a.getResponseWithHeader(ctx, method, targetURL, req, extraHeaders, errorUnwrapper)
}

func (a *AgentClientImpl) getResponseWithHeader(ctx context.Context,
	method, url string,
	req any,
	extraHeaders map[string]string,
	errorUnwrapper func(logger *zap.Logger, bodyBytes []byte) error,
) ([]byte, error) {
	logger := log.WithContextL(ctx)

	statusCode, bodyBytes, err := a.Client.RequestWithHeaders(ctx, method, url, req, extraHeaders)
	if err != nil {
		logger.Warn("failed to get response", zap.Error(err))
		return nil, err
	}

	var printedBodyBytes string
	if len(bodyBytes) > 200 {
		printedBodyBytes = string(bodyBytes[:200]) + "..."
	} else {
		printedBodyBytes = string(bodyBytes)
	}
	logger.Info("received response",
		zap.String("method", method),
		zap.String("url", url),
		zap.Int("status-code", statusCode),
		zap.String("resp", printedBodyBytes))

	if statusCode != http.StatusOK {
		if err2 := errorUnwrapper(logger, bodyBytes); err2 != nil {
			return nil, err2
		}
	}
	return bodyBytes, nil
}

func (a *AgentClientImpl) ExecSQLOnCluster(ctx context.Context, req *agent.ExecSQLReq) (*agent.ExecSQLResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/cluster/execsql", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		log.Error(ctx, "execute sql failed", zap.Error(err))
		return nil, err
	}
	var resp agent.ExecSQLResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) CheckDBConnectivity(ctx context.Context, req *agent.CheckDBConnectivityReq) (*agent.CheckDBConnectivityResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/db-connectivity", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.CheckDBConnectivityResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) GetClusterClientSecret(ctx context.Context, req *agent.GetClusterClientSecretReq) (*agent.GetClusterClientSecretResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/cluster-client-secret", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.GetClusterClientSecretResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) GetTiDBServerSecret(ctx context.Context, req *agent.GetTiDBServerSecretReq) (*agent.GetTiDBServerSecretResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/tidb-server-secret", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.GetTiDBServerSecretResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) GetEngineClient(engineAddr string) *EngineClient {
	// timeout of AgentClientImpl maybe changed by SetHttpTimeout,
	// but engine need change to a default one, so we copy it
	agentClient := *a
	agentClient.Client = a.Client.Clone()
	agentClient.Client.SetHttpTimeout(defaultTimeout)
	return &EngineClient{
		AgentClient: &agentClient,
		engineAddr:  engineAddr,
	}
}

func (a *AgentClientImpl) GetDMOpenapiClient(clusterID uint64, dmAddr string) *DMOpenapiClient {
	return nil
}

func (a *AgentClientImpl) GetLightningClient(clusterId, importID uint64) LightningAPI {
	agentClient := *a
	agentClient.Client = a.Client.Clone()
	agentClient.Client.SetHttpTimeout(defaultTimeout)

	scheme := "https"
	ns := fmt.Sprintf("tidb%d", clusterId)
	serviceName := fmt.Sprintf("import-%d", importID)

	lightningAddr := fmt.Sprintf("%s://%s.%s.svc.cluster.local:8289", scheme, serviceName, ns)

	return &LightningClient{
		AgentClient:   &agentClient,
		lightningAddr: lightningAddr,
	}
}

func (a *AgentClientImpl) PrecheckDMJob(ctx context.Context, req *agent.PrecheckDMJobReq) (*agent.PrecheckDMJobResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/dm/jobprecheck", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.PrecheckDMJobResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) GetSourceTables(ctx context.Context, req *agent.CheckDBConnectivityReq) (*agent.GetSourceTablesResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/source-tables", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.GetSourceTablesResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) GetSourceBinlogStatus(ctx context.Context, req *agent.CheckDBConnectivityReq) (*agent.GetSourceBinlogStatusResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/source/binlog-status", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.GetSourceBinlogStatusResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) GetClusterTables(ctx context.Context, req *agent.Cluster) (*agent.GetClusterTablesResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/cluster-tables", a.getSchema(), a.BaseURL)
	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.GetClusterTablesResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) GetUserFileContent(ctx context.Context, req *agent.GetUserFileReq) (*agent.GetUserFileResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/user-file", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.GetUserFileResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) NsLookup(ctx context.Context, req *agent.NsLookupReq) (*agent.NsLookupResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/nslookup", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.NsLookupResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) LoadDataToCluster(ctx context.Context, req *agent.LoadDataReq) (*agent.LoadDataResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/cluster/load-data", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.LoadDataResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) PreviewCsvFile(ctx context.Context, req *agent.PreviewCsvFileReq) (*agent.PreviewCsvFileResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/import/preview-csv-file", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.PreviewCsvFileResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) SearchImportFiles(ctx context.Context, req *agent.SearchImportFilesReq) (*agent.SearchImportFilesResp, error) {
	url := fmt.Sprintf("%s://%s/api/v1/import/search-import-files", a.getSchema(), a.BaseURL)

	respBytes, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}
	var resp agent.SearchImportFilesResp
	// use protojson to unmarshal the JSON, so that the enum value could be correctly parsed
	if err := protojson.Unmarshal(respBytes, &resp); err != nil {
		return nil, errors.Trace(err)
	}
	return &resp, nil
}

func (a *AgentClientImpl) PrecheckImportFile(ctx context.Context, req *agent.PrecheckImportFileReq) (*emptypb.Empty, error) {
	url := fmt.Sprintf("%s://%s/api/v1/import/precheck-import-file", a.getSchema(), a.BaseURL)

	_, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}

	return &emptypb.Empty{}, nil
}

func (a *AgentClientImpl) DropClusterTables(ctx context.Context, req *agent.DropClusterTablesReq) (*emptypb.Empty, error) {
	url := fmt.Sprintf("%s://%s/api/v1/import/drop-cluster-tables", a.getSchema(), a.BaseURL)

	_, err := a.getResponse(ctx, http.MethodPost, url, req)
	if err != nil {
		return nil, err
	}

	return &emptypb.Empty{}, nil
}