package service

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

	"github.com/tidbcloud/pkgv2/auth/auth0"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/tidbcloud"
	"github.com/tidbcloud/pkgv2/infra/aws"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
)

// newArchTiCDCClient is the HTTP client serverless-service uses to
// talk to a shared TiCDC cluster's `/api/v2/changefeeds` surface.
// The cluster is reached via the regional dataplane's clusterproxy
// (`/proxy/namespace/<ns>/cluster/<svc>/ticdc/<cr-name>/...`), with
// mTLS terminated via the PCA-issued cert.
//
// All call sites pass `keyspace` as the keyspace NAME (the TiDB
// cluster's user_prefix), NOT the numeric keyspace ID — see
// cdcService.getNewArchKeyspaceName for the lookup.
type newArchTiCDCClient struct {
	baseURL    string
	httpClient *http.Client
	authCfg    tidbcloud.ProxyConfig
}

const newArchTiCDCRequestTimeout = 45 * time.Second

func newNewArchTiCDCClient(ctx context.Context, clusterProxyURL string) (*newArchTiCDCClient, error) {
	cfg := config.Global()
	rawURL := clusterProxyURL
	if !strings.HasPrefix(rawURL, "http://") && !strings.HasPrefix(rawURL, "https://") {
		rawURL = "https://" + rawURL
	}
	parsedURL, err := url.Parse(rawURL)
	if err != nil {
		return nil, err
	}
	if parsedURL.Host == "" {
		return nil, errors.New("cluster proxy url host is empty")
	}
	baseURL := fmt.Sprintf(
		"%s://%s/proxy/namespace/%s/cluster/%s",
		parsedURL.Scheme,
		parsedURL.Host,
		constant.ServerlessNamespace,
		defaultServerlessSvc,
	)

	var tlsConfig *tls.Config
	if strings.HasPrefix(baseURL, "https://") {
		tlsConfig, err = aws.ProxyTlsConfigV2(ctx, cfg.SvcAddrConfig.PCAArn)
		if err != nil {
			return nil, err
		}
	}

	return &newArchTiCDCClient{
		baseURL: baseURL,
		httpClient: &http.Client{
			Timeout: newArchTiCDCRequestTimeout,
			Transport: &http.Transport{
				TLSClientConfig:   tlsConfig,
				DisableKeepAlives: true,
			},
		},
		authCfg: tidbcloud.ProxyConfig{
			Auth0Domain:       cfg.Auth0Config.ProxyAuth0Domain,
			Auth0ClientID:     cfg.Auth0Config.ProxyAuth0ClientID,
			Auth0ClientSecret: cfg.Auth0Config.ProxyAuth0ClientSecret,
			Auth0Audience:     cfg.Auth0Config.ProxyAuth0Audience,
			DevMode:           cfg.ServerConfig.Stage != config.StageProd,
		},
	}, nil
}

func (c *newArchTiCDCClient) CreateChangefeed(
	ctx context.Context,
	crName string,
	keyspace string,
	req *clusterproxy.ChangefeedConfig,
) (*clusterproxy.ChangeFeedInfo, error) {
	resp := &clusterproxy.ChangeFeedInfo{}
	return resp, c.request(ctx, "POST", c.buildChangefeedURL(crName, "", keyspace), req, resp)
}

func (c *newArchTiCDCClient) UpdateChangefeed(
	ctx context.Context,
	crName string,
	keyspace string,
	changefeedID string,
	req *clusterproxy.ChangefeedConfig,
) (*clusterproxy.ChangeFeedInfo, error) {
	resp := &clusterproxy.ChangeFeedInfo{}
	return resp, c.request(ctx, "PUT", c.buildChangefeedURL(crName, "/"+changefeedID, keyspace), req, resp)
}

func (c *newArchTiCDCClient) GetChangefeed(
	ctx context.Context,
	crName string,
	keyspace string,
	changefeedID string,
) (*clusterproxy.ChangeFeedInfo, error) {
	resp := &clusterproxy.ChangeFeedInfo{}
	return resp, c.request(ctx, "GET", c.buildChangefeedURL(crName, "/"+changefeedID, keyspace), nil, resp)
}

func (c *newArchTiCDCClient) ListChangefeeds(
	ctx context.Context,
	crName string,
	keyspace string,
) (*clusterproxy.ListResponse[clusterproxy.ChangefeedCommonInfo], error) {
	resp := &clusterproxy.ListResponse[clusterproxy.ChangefeedCommonInfo]{}
	return resp, c.request(ctx, "GET", c.buildChangefeedURL(crName, "", keyspace)+"&state=all", nil, resp)
}

func (c *newArchTiCDCClient) PauseChangefeed(
	ctx context.Context,
	crName string,
	keyspace string,
	changefeedID string,
) error {
	return c.request(ctx, "POST", c.buildChangefeedURL(crName, "/"+changefeedID+"/pause", keyspace), nil, nil)
}

func (c *newArchTiCDCClient) ResumeChangefeed(
	ctx context.Context,
	crName string,
	keyspace string,
	changefeedID string,
	req *clusterproxy.ResumeChangefeedConfig,
) error {
	if req == nil {
		req = &clusterproxy.ResumeChangefeedConfig{}
	}
	return c.request(ctx, "POST", c.buildChangefeedURL(crName, "/"+changefeedID+"/resume", keyspace), req, nil)
}

func (c *newArchTiCDCClient) DeleteChangefeed(
	ctx context.Context,
	crName string,
	keyspace string,
	changefeedID string,
) error {
	return c.request(ctx, "DELETE", c.buildChangefeedURL(crName, "/"+changefeedID, keyspace), nil, nil)
}

func (c *newArchTiCDCClient) buildChangefeedURL(crName, suffix, keyspace string) string {
	return fmt.Sprintf("%s/ticdc/%s/api/v2/changefeeds%s?keyspace=%s",
		c.baseURL,
		crName,
		suffix,
		url.QueryEscape(keyspace),
	)
}

func (c *newArchTiCDCClient) request(
	ctx context.Context,
	method string,
	url string,
	reqBody any,
	respBody any,
) error {
	var bodyReader io.Reader
	if reqBody != nil {
		payload, err := json.Marshal(reqBody)
		if err != nil {
			return err
		}
		bodyReader = bytes.NewBuffer(payload)
	}
	req, err := http.NewRequestWithContext(ctx, method, url, bodyReader)
	if err != nil {
		return err
	}
	req.Header.Set("Accept", "application/json")
	if reqBody != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	if !c.authCfg.DevMode {
		token, err := auth0.GetClientToken(
			c.authCfg.Auth0Domain,
			c.authCfg.Auth0ClientID,
			c.authCfg.Auth0ClientSecret,
			c.authCfg.Auth0Audience,
		)
		if err != nil {
			return err
		}
		req.Header.Set("Authorization", fmt.Sprintf("bearer %s", token))
	}

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return err
	}
	defer func() { _ = resp.Body.Close() }()

	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return err
	}
	if resp.StatusCode >= http.StatusBadRequest {
		return &tidbcloud.ResponseError{
			StatusCode: resp.StatusCode,
			Err:        fmt.Errorf("Request fail, %d-%s", resp.StatusCode, string(body)),
		}
	}
	if respBody == nil || len(body) == 0 {
		return nil
	}
	return json.Unmarshal(body, respBody)
}