package pd

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

var _ HttpPdClient = &pdClient{}

const (
	clusterIDPath               = "/pd/api/v1/cluster"
	keyspacePath                = "/pd/api/v2/keyspaces"
	keyspaceByNamePath          = "/pd/api/v2/keyspaces/%s"
	keyspaceByIDPath            = "/pd/api/v2/keyspaces/id/%d"
	keyspaceStatePath           = "/pd/api/v2/keyspaces/%s/state"
	keyspaceConfigPath          = "/pd/api/v2/keyspaces/%s/config"
	resourceGroupPath           = "/resource-manager/api/v1/config/group"
	resourceGroupByNamePath     = "/resource-manager/api/v1/config/group/%s"
	keyspaceRuVersionByNamePath = "/resource-manager/api/v1/config/controller/ru-version/%s"

	keyspaceMetaServiceGroup = "/pd/api/v2/meta-service-groups"

	keyspaceAlreadyExists = "keyspace already exists"
	keyspaceNotExists     = "keyspace does not exist"
)

type HttpPdClient interface {
	GetClusterID(ctx context.Context) (uint64, error)
	GetKeyspaceByID(ctx context.Context, id uint32) (*KeyspaceMeta, error)
	GetKeyspaceByName(ctx context.Context, keyspaceName string) (*KeyspaceMeta, error)
	CreateKeyspace(ctx context.Context, keyspace *KeyspaceMeta) (*KeyspaceMeta, error)
	UpdateKeyspaceState(ctx context.Context, keyspaceName string, state keyspacepb.KeyspaceState) error
	UpdateKeyspaceConfig(ctx context.Context, keyspaceName string, c *KeyspaceConfig) error
	// PostResourceGroup creates resource group.
	PostResourceGroup(ctx context.Context, group *ResourceGroup) error
	// PutResourceGroup updates resource group.
	PutResourceGroup(ctx context.Context, group *ResourceGroup) error
	DeleteResourceGroup(ctx context.Context, groupName string) error
	// SetKeyspaceRuVersion sets the RU version for a keyspace via the controller config API.
	SetKeyspaceRuVersion(ctx context.Context, keyspaceName string, req *SetKeyspaceRuVersionRequest) error
	GetMetaServiceGroupList(ctx context.Context) (*[]MetaServiceGroup, error)
}

// SetKeyspaceRuVersionRequest is the request body for setting the RU version of a keyspace
// via the /config/controller/ru_version/{keyspace_name} endpoint (tikv/pd#10334).
type SetKeyspaceRuVersionRequest struct {
	RuVersion int32 `json:"ru_version"`
}

type KeyspaceConfig struct {
	GCLifeTime               string `json:"gc_life_time,omitempty"`
	Encryption               string `json:"encryption,omitempty"`
	ZoneRedundant            string `json:"zone_redundant,omitempty"`
	TenantID                 string `json:"serverless_tenant_id,omitempty"`
	ProjectID                string `json:"serverless_project_id,omitempty"`
	ClusterID                string `json:"serverless_cluster_id,omitempty"`
	IsBranch                 string `json:"serverless_is_branch,omitempty"`
	IsBranchBootstrapped     string `json:"serverless_is_branch_bootstrapped,omitempty"`
	IsBootstrappedForRestore string `json:"serverless_is_bootstrapped_for_restore,omitempty"`
	MinRCU                   string `json:"min_rcu,omitempty"`
	MaxRCU                   string `json:"max_rcu,omitempty"`
	MainClusterId            string `json:"serverless_main_cluster_id,omitempty"`
	RuVersion                string `json:"ru_version,omitempty"`
	MetaServiceGroupID       string `json:"meta_service_group_id,omitempty"`
}

// KeyspaceMeta represents keyspace meta data.
type KeyspaceMeta struct {
	Id     uint32         `json:"id,omitempty"`
	Name   string         `json:"name"`
	State  string         `json:"state,omitempty"`
	Config KeyspaceConfig `json:"config,omitempty"`
}

// ResourceGroup the settings definitions.
type ResourceGroup struct {
	Name       string     `json:"name,omitempty"`
	Mode       int32      `json:"mode,omitempty"`
	RUSettings RUSettings `json:"r_u_settings,omitempty"`
}

type RUSettings struct {
	RU RU `json:"r_u,omitempty"`
}

type RU struct {
	Tokens   int64             `json:"tokens,omitempty"`
	Settings SpeedLimitSetting `json:"settings,omitempty"`
}

// SpeedLimitSetting is the setting to limit RU consume speed.
type SpeedLimitSetting struct {
	FillRate   uint64 `json:"fill_rate,omitempty"`
	BurstLimit int64  `json:"burst_limit,omitempty"`
}

type Cluster struct {
	Id uint64 `json:"id"`
}

type MetaServiceGroup struct {
	ID                string `json:"id"`
	Addresses         string `json:"addresses"`
	AssignedKeyspaces int    `json:"assigned_keyspaces"`
}

type pdClient struct {
	client      *http.Client
	serviceAddr string
}

func NewClient(serviceAddr string, tlsConfig *tls.Config) HttpPdClient {
	var (
		scheme = "http://"
		client = http.DefaultClient
	)
	if tlsConfig != nil {
		scheme = "https://"
		client.Transport = &http.Transport{
			TLSClientConfig:   tlsConfig,
			ForceAttemptHTTP2: true,
		}
	}
	return &pdClient{
		client:      client,
		serviceAddr: scheme + serviceAddr,
	}
}

func (p *pdClient) newRequest(
	ctx context.Context,
	method, path string,
	body io.Reader,
	headers ...[]string,
) (*http.Request, error) {
	req, err := http.NewRequestWithContext(ctx, method, p.serviceAddr+path, body)
	if err != nil {
		return nil, err
	}
	for _, header := range headers {
		req.Header.Set(header[0], header[1])
	}
	return req, nil
}

func (p *pdClient) do(r *http.Request) ([]byte, error) {
	resp, err := p.client.Do(r)
	if err != nil {
		return nil, fmt.Errorf("http request error: %w", err)
	}
	defer resp.Body.Close() //nolint:errcheck // ignore error
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("read body error: %w", err)
	}

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("http status: %s, body: %s, url: %s", resp.Status, body, r.URL)
	}

	return body, nil
}

func (p *pdClient) GetClusterID(ctx context.Context) (uint64, error) {
	req, err := p.newRequest(ctx, http.MethodGet, clusterIDPath, nil)
	if err != nil {
		return 0, err
	}
	resp, err := p.do(req)
	if err != nil {
		return 0, err
	}

	var cluster Cluster
	err = json.Unmarshal(resp, &cluster)
	if err != nil {
		return 0, fmt.Errorf("json unmarshal error: %w", err)
	}
	return cluster.Id, nil
}

func (p *pdClient) GetKeyspaceByID(ctx context.Context, id uint32) (*KeyspaceMeta, error) {
	return p.getKeyspace(ctx, fmt.Sprintf(keyspaceByIDPath, id))
}

func (p *pdClient) GetKeyspaceByName(ctx context.Context, keyspaceName string) (*KeyspaceMeta, error) {
	return p.getKeyspace(ctx, fmt.Sprintf(keyspaceByNamePath, keyspaceName))
}

func (p *pdClient) getKeyspace(ctx context.Context, path string) (*KeyspaceMeta, error) {
	req, err := p.newRequest(ctx, http.MethodGet, path, nil)
	if err != nil {
		return nil, err
	}
	resp, err := p.do(req)
	if err != nil {
		return nil, err
	}

	var keyspace KeyspaceMeta
	err = json.Unmarshal(resp, &keyspace)
	if err != nil {
		return nil, fmt.Errorf("json unmarshal error: %w", err)
	}
	return &keyspace, nil
}

func IsKeyspaceAlreadyExists(err error) bool {
	return strings.Contains(err.Error(), keyspaceAlreadyExists)
}

func IsKeyspaceNotExists(err error) bool {
	return strings.Contains(err.Error(), keyspaceNotExists)
}

func (p *pdClient) CreateKeyspace(ctx context.Context, keyspace *KeyspaceMeta) (*KeyspaceMeta, error) {
	buf := bytes.NewBuffer(nil)
	err := json.NewEncoder(buf).Encode(keyspace)
	if err != nil {
		return nil, err
	}
	req, err := p.newRequest(
		ctx,
		http.MethodPost,
		keyspacePath,
		buf,
		[]string{"Content-Type", "application/json"},
	)
	if err != nil {
		return nil, err
	}
	resp, err := p.do(req)
	if err != nil {
		return nil, fmt.Errorf("http request error: %w", err)
	}

	var k KeyspaceMeta
	err = json.Unmarshal(resp, &k)
	if err != nil {
		return nil, fmt.Errorf("json unmarshal error: %w", err)
	}

	return &k, nil
}

type UpdateStateParam struct {
	State string `json:"state"`
}

func (p *pdClient) UpdateKeyspaceState(ctx context.Context, keyspaceName string, state keyspacepb.KeyspaceState) error {
	update := UpdateStateParam{
		State: state.String(),
	}
	buf := bytes.NewBuffer(nil)
	err := json.NewEncoder(buf).Encode(update)
	if err != nil {
		return err
	}

	req, err := p.newRequest(
		ctx,
		http.MethodPut,
		fmt.Sprintf(keyspaceStatePath, keyspaceName),
		buf,
		[]string{"Content-Type", "application/json"},
	)
	if err != nil {
		return err
	}
	_, err = p.do(req)
	return err
}

// UpdateConfigParams represents parameters needed to modify target keyspace's configs.
type UpdateConfigParams struct {
	Config *KeyspaceConfig `json:"config"`
}

func (p *pdClient) UpdateKeyspaceConfig(ctx context.Context, keyspaceName string, c *KeyspaceConfig) error {
	update := &UpdateConfigParams{
		Config: c,
	}
	buf := bytes.NewBuffer(nil)
	err := json.NewEncoder(buf).Encode(update)
	if err != nil {
		return err
	}
	req, err := p.newRequest(ctx,
		http.MethodPatch,
		fmt.Sprintf(keyspaceConfigPath, keyspaceName),
		buf,
		[]string{"Content-Type", "application/json"},
	)
	if err != nil {
		return err
	}
	_, err = p.do(req)
	return err
}

func (p *pdClient) PostResourceGroup(ctx context.Context, group *ResourceGroup) error {
	return p.writeResourceGroup(ctx, http.MethodPost, group)
}

func (p *pdClient) PutResourceGroup(ctx context.Context, group *ResourceGroup) error {
	return p.writeResourceGroup(ctx, http.MethodPut, group)
}

func (p *pdClient) writeResourceGroup(ctx context.Context, method string, group *ResourceGroup) error {
	buf := bytes.NewBuffer(nil)
	err := json.NewEncoder(buf).Encode(group)
	if err != nil {
		return err
	}
	req, err := p.newRequest(ctx,
		method,
		resourceGroupPath,
		buf,
		[]string{"Content-Type", "application/json"},
	)
	if err != nil {
		return err
	}
	_, err = p.do(req)
	return err
}

func (p *pdClient) DeleteResourceGroup(ctx context.Context, groupName string) error {
	req, err := p.newRequest(
		ctx,
		http.MethodDelete,
		fmt.Sprintf(resourceGroupByNamePath, groupName),
		nil,
		[]string{"Content-Type", "application/json"},
	)
	if err != nil {
		return err
	}
	_, err = p.do(req)
	return err
}

func (p *pdClient) SetKeyspaceRuVersion(ctx context.Context, keyspaceName string, req *SetKeyspaceRuVersionRequest) error {
	buf := bytes.NewBuffer(nil)
	if err := json.NewEncoder(buf).Encode(req); err != nil {
		return err
	}
	r, err := p.newRequest(ctx,
		http.MethodPost,
		fmt.Sprintf(keyspaceRuVersionByNamePath, keyspaceName),
		buf,
		[]string{"Content-Type", "application/json"},
	)
	if err != nil {
		return err
	}
	_, err = p.do(r)
	return err
}

func (p *pdClient) GetMetaServiceGroupList(ctx context.Context) (*[]MetaServiceGroup, error) {
	req, err := p.newRequest(ctx, http.MethodGet, keyspaceMetaServiceGroup, nil)
	if err != nil {
		log.Info("new request to get meta service group list from PD error", zap.Error(err))
		return nil, err
	}
	resp, err := p.do(req)
	if err != nil {
		log.Info("do request to get meta service group list from PD error", zap.Error(err))
		return nil, err
	}

	var metaServiceGroupList []MetaServiceGroup
	err = json.Unmarshal(resp, &metaServiceGroupList)
	if err != nil {
		return nil, fmt.Errorf("json unmarshal error: %w", err)
	}
	return &metaServiceGroupList, nil
}