package clustergateway

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"strings"

	"github.com/go-resty/resty/v2"
	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	pdapiv2 "github.com/tikv/pd/server/apiv2/handlers"
)

// ErrKeyspaceNotFound is returned when PD reports that the requested keyspace does not exist.
var ErrKeyspaceNotFound = errors.New("keyspace not found")

// providerTypeToKmsVendor maps cloud provider type to the KMS vendor name
// expected by PD/TiKV. Notably, ALICLOUD maps to "aliyun".
func providerTypeToKmsVendor(providerType string) string {
	v := strings.ToLower(providerType)
	if v == "alicloud" {
		return "aliyun"
	}
	return v
}

type keyspaceEncryptionConfig struct {
	Enabled           bool              `json:"enabled"`
	CmekId            string            `json:"cmek_id"`
	Vendor            string            `json:"vendor"`
	Region            string            `json:"region"`
	Endpoint          string            `json:"endpoint,omitempty"`
	EncryptionContext map[string]string `json:"encryption_context,omitempty"`
}

func filterEncryptionContext(encryptionContext map[string]string) map[string]string {
	if externalID := encryptionContext["ExternalId"]; externalID != "" {
		return map[string]string{"ExternalId": externalID}
	}
	return nil
}

// PD API

func (c *clusterGatewayImpl) GetKeySpace(ctx context.Context, sharedPool *entity.SharedPool, keySpace string) (*pdapiv2.KeyspaceMeta, error) {
	resp, err := c.callClusterProxy(ctx, sharedPool, http.MethodGet, operationMap["GetKeySpace"], func(request *resty.Request) *resty.Request {
		request.SetPathParam(sharedPoolIdParameter, sharedPool.GetId())
		request.SetPathParam(keySpaceParameter, keySpace)
		return request
	})
	if err != nil {
		if resp != nil {
			if resp.StatusCode() == http.StatusNotFound ||
				strings.Contains(string(resp.Body()), "keyspace does not exist") {
				return nil, fmt.Errorf("get keyspace %s: %w", keySpace, ErrKeyspaceNotFound)
			}
		}
		return nil, fmt.Errorf("get keyspace %s: %w", keySpace, err)
	}

	var out pdapiv2.KeyspaceMeta
	if err := json.Unmarshal(resp.Body(), &out); err != nil {
		return nil, fmt.Errorf("unmarshal keyspace %s response failed: %w", keySpace, err)
	}

	return &out, nil
}

func (c *clusterGatewayImpl) UpdateKeySpaceState(ctx context.Context, sharedPool *entity.SharedPool, keySpace string, state keyspacepb.KeyspaceState) error {
	resp, err := c.callClusterProxy(ctx, sharedPool, http.MethodPut, operationMap["UpdateKeySpaceState"], func(request *resty.Request) *resty.Request {
		request.SetPathParam(sharedPoolIdParameter, sharedPool.GetId())
		request.SetPathParam(keySpaceParameter, keySpace)
		request.SetBody(
			pdapiv2.UpdateStateParam{
				State: state.String(),
			},
		)
		return request
	})
	if err != nil {
		if resp != nil && resp.StatusCode() == http.StatusConflict {
			return nil
		}
		return err
	}

	return nil
}

func (c *clusterGatewayImpl) CreateKeySpace(ctx context.Context, sharedPool *entity.SharedPool, keySpace, cmekId string, encryptionContext map[string]string) error {
	resp, err := c.callClusterProxy(ctx, sharedPool, http.MethodPost, operationMap["CreateKeySpace"], func(request *resty.Request) *resty.Request {
		request.SetPathParam(sharedPoolIdParameter, sharedPool.GetId())

		body := pdapiv2.CreateKeyspaceParams{
			Name:   keySpace,
			Config: map[string]string{},
		}
		if cmekId != "" {
			config := &keyspaceEncryptionConfig{
				Enabled:           true,
				CmekId:            cmekId,
				Vendor:            providerTypeToKmsVendor(sharedPool.GetProvider().GetType().String()),
				Region:            strings.ToLower(sharedPool.GetProvider().GetRegion()),
				EncryptionContext: filterEncryptionContext(encryptionContext),
			}
			jsonConfig, _ := json.Marshal(config)
			body.Config["encryption"] = string(jsonConfig)
		}
		request.SetBody(body)
		return request
	})
	if err != nil {
		if resp != nil && resp.StatusCode() == http.StatusConflict {
			return nil
		}

		// pd hard code
		// keyspace already exists
		if resp != nil && resp.StatusCode() == http.StatusInternalServerError &&
			strings.Contains(string(resp.Body()), "keyspace already exists") {

			// use get to double check
			keyspace, err := c.GetKeySpace(ctx, sharedPool, keySpace)
			if err != nil {
				if IsNotFoundError(err) {
					return fmt.Errorf("create api return keyspace %s already exists, but not found in PD: %w", keySpace, err)
				}
				// if get keyspace failed, return the original error
				return fmt.Errorf("get keyspace %s failed: %w", keySpace, err)
			}

			if keyspace.GetState() != keyspacepb.KeyspaceState_ENABLED {
				return fmt.Errorf("keyspace %s already exists, but state is not enabled: %s", keySpace, keyspace.GetState())
			}
			return nil
		}

		return err
	}

	return nil
}

func (c *clusterGatewayImpl) ListKeySpaces(ctx context.Context, sharedPool *entity.SharedPool) (*pdapiv2.LoadAllKeyspacesResponse, error) {
	resp, err := c.callClusterProxy(ctx, sharedPool, http.MethodGet, operationMap["ListKeySpaces"], func(request *resty.Request) *resty.Request {
		request.SetPathParam(sharedPoolIdParameter, sharedPool.GetId())
		return request
	})
	if err != nil {
		return nil, fmt.Errorf("list keyspaces failed: %w", err)
	}

	var out pdapiv2.LoadAllKeyspacesResponse
	if err := json.Unmarshal(resp.Body(), &out); err != nil {
		return nil, fmt.Errorf("unmarshal keyspaces response failed: %w", err)
	}

	return &out, nil
}