package common

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

	"github.com/pingcap/kvproto/pkg/keyspacepb"
)

const (
	metaServerListKeyspacesPath  = "/api/v2/meta:list"
	metaServerListKeyspacesLimit = 1000
)

type metaServerListKeyspacesResponse struct {
	Items  []*metaServerKeyspace `json:"items"`
	Total  int                   `json:"total"`
	Limit  int                   `json:"limit"`
	Offset int                   `json:"offset"`
}

type metaServerKeyspace struct {
	KeyspaceID     string                   `json:"KeyspaceID"`
	KeyspaceName   string                   `json:"KeyspaceName"`
	State          keyspacepb.KeyspaceState `json:"State"`
	CreatedAt      int64                    `json:"CreatedAt"`
	StateChangedAt int64                    `json:"StateChangedAt"`
	Config         map[string]string        `json:"Config"`
}

func GetKeyspaceByState(ctx context.Context, metaServerAddr string, state keyspacepb.KeyspaceState) ([]*keyspacepb.KeyspaceMeta, error) {
	if strings.TrimSpace(metaServerAddr) == "" {
		return nil, fmt.Errorf("meta server address is empty")
	}

	httpClient := &http.Client{Timeout: time.Duration(httpTimeout) * time.Second}
	var allKeyspaces []*keyspacepb.KeyspaceMeta
	offset := 0
	for {
		resp, err := listKeyspacesFromMetaServer(ctx, httpClient, metaServerAddr, metaServerListKeyspacesLimit, offset, state)
		if err != nil {
			return nil, err
		}
		if len(resp.Items) == 0 {
			break
		}
		for _, item := range resp.Items {
			keyspaceMeta, err := item.toKeyspaceMeta()
			if err != nil {
				return nil, err
			}
			allKeyspaces = append(allKeyspaces, keyspaceMeta)
		}

		offset += len(resp.Items)
		if offset >= resp.Total {
			break
		}
	}
	return allKeyspaces, nil
}

func listKeyspacesFromMetaServer(
	ctx context.Context,
	httpClient *http.Client,
	metaServerAddr string,
	limit int,
	offset int,
	state keyspacepb.KeyspaceState,
) (*metaServerListKeyspacesResponse, error) {
	requestURL, err := buildMetaServerListKeyspacesURL(metaServerAddr, limit, offset, state)
	if err != nil {
		return nil, err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL, nil)
	if err != nil {
		return nil, err
	}
	res, err := httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()
	if res.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(io.LimitReader(res.Body, 4096))
		return nil, fmt.Errorf("list keyspaces from meta server failed: status=%d body=%s", res.StatusCode, string(body))
	}

	var resp metaServerListKeyspacesResponse
	if err = json.NewDecoder(res.Body).Decode(&resp); err != nil {
		return nil, err
	}
	return &resp, nil
}

func buildMetaServerListKeyspacesURL(metaServerAddr string, limit int, offset int, state keyspacepb.KeyspaceState) (string, error) {
	addr := strings.TrimRight(strings.TrimSpace(metaServerAddr), "/")
	if !strings.HasPrefix(addr, "http://") && !strings.HasPrefix(addr, "https://") {
		addr = "http://" + addr
	}
	u, err := url.Parse(addr + metaServerListKeyspacesPath)
	if err != nil {
		return "", err
	}
	query := u.Query()
	query.Set("limit", strconv.Itoa(limit))
	query.Set("offset", strconv.Itoa(offset))
	query.Set("state", state.String())
	u.RawQuery = query.Encode()
	return u.String(), nil
}

func (m *metaServerKeyspace) toKeyspaceMeta() (*keyspacepb.KeyspaceMeta, error) {
	id, err := strconv.ParseUint(m.KeyspaceID, 10, 32)
	if err != nil {
		return nil, fmt.Errorf("invalid keyspace id %q from meta server: %w", m.KeyspaceID, err)
	}
	return &keyspacepb.KeyspaceMeta{
		Id:             uint32(id),
		Name:           m.KeyspaceName,
		State:          m.State,
		CreatedAt:      m.CreatedAt,
		StateChangedAt: m.StateChangedAt,
		Config:         maps.Clone(m.Config),
	}, nil
}