// Package ticdcservicecli provides an HTTP client for the ticdc-service REST API
// (exposed via grpc-gateway). The client pattern follows the regionalservercli package.
package ticdcservicecli

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"path/filepath"
	"time"

	spb "google.golang.org/genproto/googleapis/rpc/status"
	grpcstatus "google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"sigs.k8s.io/controller-runtime/pkg/log"

	ticdcpb "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
)

// Client exposes the ticdc-service operations needed for CDC node upgrade.
type Client interface {
	// ListTiCDCClusters returns all TiCDC clusters belonging to the given tidb cluster.
	ListTiCDCClusters(ctx context.Context, tidbClusterID string, ticdcClusters []string) (*ticdcpb.ListTiCDCClustersResponse, error)
	// UpdateTiCDCNodeVersion updates the node version on a single TiCDC cluster.
	UpdateTiCDCNodeVersion(ctx context.Context, ticdcClusterID, nodeVersion string) error
}

type client struct {
	endpoint     string
	httpClient   *http.Client
	marshalOpt   protojson.MarshalOptions
	unmarshalOpt protojson.UnmarshalOptions
}

// NewClient creates a new TiCDC service HTTP client.
func NewClient(endpoint string) Client {
	return &client{
		endpoint: endpoint,
		httpClient: &http.Client{
			Timeout: 10 * time.Second,
		},
		marshalOpt: protojson.MarshalOptions{
			UseProtoNames:   false,
			EmitUnpopulated: true,
		},
		unmarshalOpt: protojson.UnmarshalOptions{
			DiscardUnknown: true,
		},
	}
}

func (c *client) buildURL(format string, args ...any) string {
	path := fmt.Sprintf(format, args...)
	path = filepath.Join("/api/v1", path)
	return c.endpoint + path
}

func (c *client) processError(httpStatusCode int, data []byte) error {
	s := new(spb.Status)
	if err := c.unmarshalOpt.Unmarshal(data, s); err != nil {
		return fmt.Errorf("HTTP %d: %s", httpStatusCode, string(data))
	}
	s.Message = fmt.Sprintf("%s;HTTPSTATUSCODE:%v", s.Message, httpStatusCode)
	return grpcstatus.ErrorProto(s)
}

func (c *client) doGet(ctx context.Context, rawURL string) (data []byte, err error) {
	startTime := time.Now()
	logger := log.FromContext(ctx)
	defer func() {
		cost := time.Since(startTime).String()
		preview := string(data)
		if len(preview) > 200 {
			preview = preview[:200] + "..."
		}
		if err == nil {
			logger.Info("ticdc-service GET request succeeded", "cost", cost, "url", rawURL, "response", preview)
		} else {
			logger.Error(err, "ticdc-service GET request failed", "cost", cost, "url", rawURL)
		}
	}()

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
	if err != nil {
		return nil, err
	}
	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	data, err = io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode >= 300 {
		return nil, c.processError(resp.StatusCode, data)
	}
	return data, nil
}

func (c *client) doPost(ctx context.Context, rawURL string, body []byte) (data []byte, err error) {
	startTime := time.Now()
	logger := log.FromContext(ctx)
	defer func() {
		cost := time.Since(startTime).String()
		if err == nil {
			logger.Info("ticdc-service POST request succeeded", "cost", cost, "url", rawURL)
		} else {
			logger.Error(err, "ticdc-service POST request failed", "cost", cost, "url", rawURL)
		}
	}()

	req, err := http.NewRequestWithContext(ctx, http.MethodPost, rawURL, bytes.NewReader(body))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	data, err = io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	if resp.StatusCode >= 300 {
		return nil, c.processError(resp.StatusCode, data)
	}
	return data, nil
}

// ListTiCDCClusters queries the ticdc-service for all TiCDC clusters belonging to a TiDB cluster.
func (c *client) ListTiCDCClusters(ctx context.Context, tidbClusterID string, ticdcClusters []string) (*ticdcpb.ListTiCDCClustersResponse, error) {
	query := url.Values{}
	query.Set("tidbClusterId", tidbClusterID)
	for _, id := range ticdcClusters {
		query.Add("ticdcClusterId", id)
	}
	rawURL := c.buildURL("/ticdcClusters?%s", query.Encode())

	data, err := c.doGet(ctx, rawURL)
	if err != nil {
		return nil, fmt.Errorf("ListTiCDCClusters: %w", err)
	}

	out := new(ticdcpb.ListTiCDCClustersResponse)
	if err := c.unmarshalOpt.Unmarshal(data, out); err != nil {
		return nil, err
	}
	return out, nil
}

// UpdateTiCDCNodeVersion calls the ticdc-service to set the node version on a TiCDC cluster.
func (c *client) UpdateTiCDCNodeVersion(ctx context.Context, ticdcClusterID, nodeVersion string) error {
	rawURL := c.buildURL("/ticdcClusters/%s:updateNodeVersion", ticdcClusterID)
	req := &ticdcpb.UpdateTiCDCNodeVersionRequest{
		NodeVersion: nodeVersion,
	}
	bs, err := c.marshalOpt.Marshal(req)
	if err != nil {
		return err
	}
	if _, err := c.doPost(ctx, rawURL, bs); err != nil {
		return fmt.Errorf("UpdateTiCDCNodeVersion %s: %w", ticdcClusterID, err)
	}
	return nil
}