package configmanager

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

	"github.com/gogo/protobuf/proto"
	"github.com/pingcap/errors"
	"sigs.k8s.io/controller-runtime/pkg/log"

	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/utils"
)

const configPrefix = "pd/api/v1/config"
const storePrefix = "pd/api/v1/stores"
const storeUpStatePrefix = "pd/api/v1/store/%v/state?state=Up"
const schedulersPrefix = "pd/api/v1/schedulers"
const evictSchedulerLeader = "evict-leader-scheduler"

type PDAuth func(*http.Request) error

type Option func(p *pdClient)

func Auth(auth PDAuth) Option {
	return func(p *pdClient) {
		p.auth = auth
	}
}

func WithDryrun(dryrun bool) Option {
	return func(p *pdClient) {
		p.dryrun = dryrun
	}
}

func HttpClient(client *http.Client) Option {
	return func(p *pdClient) {
		p.client = client
	}
}

type pdClient struct {
	url    string
	client *http.Client
	auth   PDAuth
	dryrun bool
}

func NewDefaultPDClient(url string, opts ...Option) DynamicPDClient {
	p := &pdClient{
		url: url,
	}
	for _, opt := range opts {
		opt(p)
	}

	// We may set a client via an Option. If not, create a default one.
	if p.client == nil {
		p.client = &http.Client{
			Timeout: 10 * time.Second,
			Transport: &http.Transport{
				TLSClientConfig: &tls.Config{
					// nolint: gosec
					InsecureSkipVerify: true,
				},
			},
		}
	}
	return p
}

type StoreInfo struct {
	Store  *MetaStore   `json:"store"`
	Status *StoreStatus `json:"status"`
}

// StoresInfo is stores info returned from PD RESTful interface
type StoresInfo struct {
	Count  int          `json:"count"`
	Stores []*StoreInfo `json:"stores"`
}
type StoreState int32

var StoreState_name = map[int32]string{
	0: "Up",
	1: "Offline",
	2: "Tombstone",
}

var StoreState_value = map[string]int32{
	"Up":        0,
	"Offline":   1,
	"Tombstone": 2,
}

func (x StoreState) String() string {
	return proto.EnumName(StoreState_name, int32(x))
}

const (
	StoreState_Up        StoreState = 0
	StoreState_Offline   StoreState = 1
	StoreState_Tombstone StoreState = 2
)

type Store struct {
	Id uint64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
	// Address to handle client requests (kv, cop, etc.)
	Address string        `protobuf:"bytes,2,opt,name=address,proto3" json:"address,omitempty"`
	State   StoreState    `protobuf:"varint,3,opt,name=state,proto3,enum=metapb.StoreState" json:"state,omitempty"`
	Labels  []*StoreLabel `protobuf:"bytes,4,rep,name=labels,proto3" json:"labels,omitempty"`
	Version string        `protobuf:"bytes,5,opt,name=version,proto3" json:"version,omitempty"`
	// Address to handle peer requests (raft messages from other store).
	// Empty means same as address.
	PeerAddress string `protobuf:"bytes,6,opt,name=peer_address,json=peerAddress,proto3" json:"peer_address,omitempty"`
	// Status address provides the HTTP service for external components
	StatusAddress string `protobuf:"bytes,7,opt,name=status_address,json=statusAddress,proto3" json:"status_address,omitempty"`
	GitHash       string `protobuf:"bytes,8,opt,name=git_hash,json=gitHash,proto3" json:"git_hash,omitempty"`
	// The start timestamp of the current store
	StartTimestamp int64  `protobuf:"varint,9,opt,name=start_timestamp,json=startTimestamp,proto3" json:"start_timestamp,omitempty"`
	DeployPath     string `protobuf:"bytes,10,opt,name=deploy_path,json=deployPath,proto3" json:"deploy_path,omitempty"`
	// The last heartbeat timestamp of the store.
	LastHeartbeat int64 `protobuf:"varint,11,opt,name=last_heartbeat,json=lastHeartbeat,proto3" json:"last_heartbeat,omitempty"`
	// If the store is physically destroyed, which means it can never up again.
	PhysicallyDestroyed bool `protobuf:"varint,12,opt,name=physically_destroyed,json=physicallyDestroyed,proto3" json:"physically_destroyed,omitempty"`
	// NodeState is used to replace StoreState which will be deprecated in the future.
	NodeState NodeState `protobuf:"varint,13,opt,name=node_state,json=nodeState,proto3,enum=metapb.NodeState" json:"node_state,omitempty"`
}

type NodeState int32

const (
	NodeState_Preparing NodeState = 0
	NodeState_Serving   NodeState = 1
	NodeState_Removing  NodeState = 2
	NodeState_Removed   NodeState = 3
)

type StoreLabel struct {
	Key   string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
	Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
}

type MetaStore struct {
	Store
	StateName string `json:"state_name"`
}

// StoreStatus is TiKV store status returned from PD RESTful interface
type StoreStatus struct {
	Capacity           string `json:"capacity"`
	Available          string `json:"available"`
	LeaderCount        int    `json:"leader_count"`
	RegionCount        int    `json:"region_count"`
	SendingSnapCount   uint32 `json:"sending_snap_count"`
	ReceivingSnapCount uint32 `json:"receiving_snap_count"`
	ApplyingSnapCount  uint32 `json:"applying_snap_count"`
	IsBusy             bool   `json:"is_busy"`

	StartTS         time.Time `json:"start_ts"`
	LastHeartbeatTS time.Time `json:"last_heartbeat_ts"`
	Uptime          string    `json:"uptime"`
}

func (c *pdClient) CancelDeleteStore(ctx context.Context, storeID string) error {
	path := fmt.Sprintf(storeUpStatePrefix, storeID)
	return c.HTTPPost(ctx, path, []byte{})
}

func (c *pdClient) GetEvictLeaderSchedulers(ctx context.Context) ([]string, error) {
	body, err := c.HTTPGet(ctx, schedulersPrefix)
	if err != nil {
		return nil, fmt.Errorf("failed to get schedulers from PD: %w", err)
	}

	var schedulers []string
	if err := json.Unmarshal(body, &schedulers); err != nil {
		return nil, fmt.Errorf("failed to unmarshal schedulers response: %w", err)
	}

	// Filter schedulers to only include evict-leader-schedulers
	var evictSchedulers []string
	for _, scheduler := range schedulers {
		if strings.HasPrefix(scheduler, evictSchedulerLeader) {
			evictSchedulers = append(evictSchedulers, scheduler)
		}
	}

	return evictSchedulers, nil
}

func (c *pdClient) RemoveEvictLeaderScheduler(ctx context.Context) error {
	path := fmt.Sprintf("%s/%s", schedulersPrefix, evictSchedulerLeader)
	if err := c.HTTPDelete(ctx, path, []byte{}); err != nil {
		return fmt.Errorf("failed to remove evict-leader-scheduler: %w", err)
	}
	return nil
}

func (c *pdClient) GetStores(ctx context.Context) (StoresInfo, error) {
	body, err := c.HTTPGet(ctx, storePrefix)
	if err != nil {
		return StoresInfo{}, err
	}
	stores := StoresInfo{}

	err = json.Unmarshal(body, &stores)
	if err != nil {
		log.FromContext(ctx).Error(err, "failed to unmarshal stores info", "body", string(body))
		return StoresInfo{}, err
	}
	return stores, nil
}

func (c *pdClient) GetConfig(ctx context.Context) (map[string]interface{}, error) {
	body, err := c.HTTPGet(ctx, configPrefix)
	if err != nil {
		return nil, err
	}
	config := map[string]interface{}{}
	err = json.Unmarshal(body, &config)
	if err != nil {
		return nil, fmt.Errorf("cannot unmarshal response(%s) to json: %w", string(body), err)
	}
	return config, nil
}

func (c *pdClient) HTTPGet(ctx context.Context, path string) ([]byte, error) {
	apiURL := fmt.Sprintf("%s/%s", c.url, path)
	req, err := http.NewRequest(http.MethodGet, apiURL, nil)
	if err != nil {
		return nil, err
	}
	if c.auth != nil {
		if err = c.auth(req); err != nil {
			return nil, err
		}
	}
	res, err := c.client.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()
	body, err := io.ReadAll(res.Body)
	if err != nil {
		return nil, err
	}
	return body, nil
}

func (c *pdClient) PutConfig(ctx context.Context, conf map[string]interface{}) error {
	// special treatment for PD's update config API
	body := utils.FlattenMap(conf)
	data, err := json.Marshal(body)
	if err != nil {
		return errors.Annotate(err, "encoding PD put config request body")
	}
	return c.HTTPPost(ctx, configPrefix, data)
}

func (c *pdClient) HTTPDo(ctx context.Context, path string, body []byte, method string) error {
	logger := log.FromContext(ctx).WithValues("client", "pd", "method", method, "path", path)
	apiURL := fmt.Sprintf("%s/%s", c.url, path)

	if c.dryrun {
		logger.Info("HTTP request rejected in dryrun mode", "url", apiURL, "bodySize", len(body))
		return fmt.Errorf("pdClient: HTTP %s request to %s rejected due to dryrun mode", method, apiURL)
	}

	logger.V(1).Info("Sending HTTP request to PD", "url", apiURL, "bodySize", len(body))

	req, err := http.NewRequest(method, apiURL, bytes.NewBuffer(body))
	if err != nil {
		return errors.Annotatef(err, "failed to build HTTP %s request to PD API %s", method, apiURL)
	}
	if c.auth != nil {
		if err = c.auth(req); err != nil {
			return errors.Annotatef(err, "failed to apply authentication for HTTP %s request to %s", method, apiURL)
		}
	}
	res, err := c.client.Do(req)
	if err != nil {
		return errors.Annotatef(err, "failed to execute HTTP %s request to PD API %s", method, apiURL)
	}
	defer res.Body.Close()

	if res.StatusCode != http.StatusOK {
		errorBody, readErr := io.ReadAll(res.Body)
		if readErr != nil {
			logger.Error(readErr, "Failed to read error response body", "statusCode", res.StatusCode)
			return errors.Annotatef(readErr, "HTTP %s request to PD API %s failed with status %d, unable to read error body", method, apiURL, res.StatusCode)
		}
		logger.Error(nil, "PD API returned non-OK status", "statusCode", res.StatusCode, "errorBody", string(errorBody))
		return errors.Errorf("HTTP %s request to PD API %s failed with status %d: %s", method, apiURL, res.StatusCode, string(errorBody))
	}

	logger.V(1).Info("HTTP request to PD completed successfully", "statusCode", res.StatusCode)
	return nil
}

func (c *pdClient) HTTPPost(ctx context.Context, path string, body []byte) error {
	return c.HTTPDo(ctx, path, body, http.MethodPost)
}

func (c *pdClient) HTTPDelete(ctx context.Context, path string, body []byte) error {
	return c.HTTPDo(ctx, path, body, http.MethodDelete)
}