package repository

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"time"

	"github.com/tidbcloud/serverless-metering/pkg/config"
)

// response from pd meta store
type MetaData struct {
	KeyspaceID   string
	KeyspaceName string
	ClusterId    string
	TenantID     string
	ProjectID    string
	Encrypted    bool
	IsBranch     bool
	RUVersion    string
	MinRCU       int64
	MaxRCU       int64
}

type PDMetaStoreClient struct {
	pdMetaStoreHost string
	pdMetaStorePort int
	client          http.Client
}

func NewPDMetaStoreClient() *PDMetaStoreClient {
	cfg := config.Global().Repo
	client := http.Client{
		Timeout: time.Second * 3,
	}
	return &PDMetaStoreClient{
		client:          client,
		pdMetaStoreHost: cfg.PdMetaStoreHost,
		pdMetaStorePort: cfg.PdMetaStorePort,
	}
}

func (r *PDMetaStoreClient) GetClusterByID(ctx context.Context, clusterID string) (*ClusterBranch, error) {
	url := fmt.Sprintf("http://%s:%d/api/v1/meta?cluster_id=%s", r.pdMetaStoreHost, r.pdMetaStorePort, clusterID)
	return r.getCluster(ctx, url)
}

func (r *PDMetaStoreClient) GetClusterByKeyspaceID(ctx context.Context, keyspaceID string) (*ClusterBranch, error) {
	url := fmt.Sprintf("http://%s:%d/api/v1/meta?keyspace_id=%s", r.pdMetaStoreHost, r.pdMetaStorePort, keyspaceID)
	return r.getCluster(ctx, url)
}

func (r *PDMetaStoreClient) ListEssentialPlanClusters(ctx context.Context) ([]*ClusterBranch, error) {
	url := fmt.Sprintf("http://%s:%d/api/v1/meta?billing_plan=essential", r.pdMetaStoreHost, r.pdMetaStorePort)
	metaDatas, err := r.requestPDMetaStore(ctx, url)
	if err != nil {
		return nil, err
	}
	return metaDatas, nil
}

func (r *PDMetaStoreClient) getCluster(ctx context.Context, url string) (*ClusterBranch, error) {
	metaDatas, err := r.requestPDMetaStore(ctx, url)
	if err != nil {
		return nil, err
	}
	if len(metaDatas) == 0 {
		return nil, fmt.Errorf("cluster not found")
	}
	if len(metaDatas) > 1 {
		return nil, fmt.Errorf("multiple clusters found")
	}

	return metaDatas[0], nil
}

func (r *PDMetaStoreClient) requestPDMetaStore(_ context.Context, url string) ([]*ClusterBranch, error) {
	resp, err := r.client.Get(url)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode == http.StatusNotFound {
		return nil, fmt.Errorf("cluster branch not found")
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("http response not ok, status: %d", resp.StatusCode)
	}

	data, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}

	var metaDatas []MetaData
	err = json.Unmarshal(data, &metaDatas)
	if err != nil {
		return nil, err
	}

	result := make([]*ClusterBranch, 0, len(metaDatas))
	for _, md := range metaDatas {
		if md.IsBranch {
			continue
		}
		clusterID, err := strconv.ParseUint(md.ClusterId, 10, 64)
		if err != nil {
			continue
		}
		tenantID, err := strconv.ParseUint(md.TenantID, 10, 64)
		if err != nil {
			continue
		}
		projectID, err := strconv.ParseUint(md.ProjectID, 10, 64)
		if err != nil {
			continue
		}
		keyspaceID, err := strconv.ParseUint(md.KeyspaceID, 10, 32)
		if err != nil {
			continue
		}
		result = append(result, &ClusterBranch{
			Type:            CBTypeCluster,
			PrimaryID:       md.ClusterId,
			ClusterID:       clusterID,
			TenantID:        tenantID,
			ProjectID:       projectID,
			KeyspaceID:      uint32(keyspaceID),
			KeyspaceToken:   md.KeyspaceName,
			EncryptKeyspace: md.Encrypted,
			RUVersion:       md.RUVersion,
			MaxRcu:          md.MaxRCU,
			MinRcu:          md.MinRCU,
		})
	}
	return result, nil
}