package clusterproxy

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/patrickmn/go-cache"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy/api"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy/client"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	"github.com/tidbcloud/ticdc-service/pkg/regionalservice"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
)

const (
	// DefaultClusterProxyTimeout is the default timeout for cluster proxy
	DefaultClusterProxyTimeout = 60
	// CDCHeavyOperationTimeout for cdc create and update Request, cdc need to list all the tables, and it can spend much time
	// so we need a longer timeout
	CDCHeavyOperationTimeout = 60 * 10
)

// ClusterProxy is the interface for the cluster proxy
// it provides the PD client, SQL client and changefeed client
type ClusterProxy interface {
	Client() client.Client
	api.PDClientGetter
	api.SQLClientGetter
	api.ChangefeedClientGetter
}

// Model is the interface for the cluster proxy model
type Model interface {
	// GetClusterProxy returns the cluster proxy for the given cluster ID
	GetClusterProxy(ctx context.Context, clusterId string, timeout int) (ClusterProxy, error)
}

var _ Model = &clusterProxyModelImpl{}

// The clusterProxyModelImpl struct provides the implementation of the Model interface.
type clusterProxyModelImpl struct {
	cache           *cache.Cache
	clusterPwdCache *cache.Cache
	cfg             *config.Config
	opts            []client.Option
	regionalCli     regionalservice.Client
	clientCreator   client.Creator
}

func NewClusterProxyModelImpl(
	cfg *config.Config,
	regionalCli regionalservice.Client,
	clientCreator client.Creator,
	opts ...client.Option,
) Model {
	return &clusterProxyModelImpl{
		cfg:             cfg,
		cache:           cache.New(1*time.Minute, 30*time.Second),
		clusterPwdCache: cache.New(1*time.Minute, 30*time.Second),
		opts:            opts,
		regionalCli:     regionalCli,
		clientCreator:   clientCreator,
	}
}

func (p *clusterProxyModelImpl) GetClusterProxy(
	ctx context.Context, clusterIdStr string, timeout int,
) (ClusterProxy, error) {
	logger := tilog.WithContextL(ctx)
	cacheKey := clusterIdStr
	if proxy, ok := p.cache.Get(cacheKey); ok {
		return proxy.(ClusterProxy), nil
	}

	cluster, err := p.regionalCli.GetCluster(ctx, clusterIdStr)
	if err != nil {
		logger.Error("Failed to get cluster", zap.Error(err))
		if apierrors.IsNotFound(err) {
			return nil, status.Errorf(codes.NotFound, "Cluster %s not found", clusterIdStr)
		}
		return nil, status.Errorf(codes.Internal, "Failed to get cluster %s: %s", clusterIdStr, err)
	}
	if cluster.ProxyBaseUrl == "" {
		logger.Error("can't get proxy URL")
		return nil, status.Errorf(codes.Internal, "Failed to get proxy URL for cluster %s, please try again later", clusterIdStr)
	}

	clusterId, err := strconv.ParseUint(clusterIdStr, 10, 64)
	if err != nil { // clusterproxy library requires uint64 clusterId
		logger.Error("cluster id is not uint64", zap.Error(err))
		return nil, status.Errorf(codes.InvalidArgument, "invalid cluster id: %s", clusterIdStr)
	}
	opts := append(p.opts, client.WithTimeout(timeout))
	proxyClient, err := p.clientCreator.CreateClient(ctx, clusterId,
		cluster.ProxyBaseUrl, opts...)
	if err != nil {
		return nil, fmt.Errorf("failed to get proxy, %s", err.Error())
	}
	proxy := NewClusterProxy(proxyClient)
	p.cache.SetDefault(cacheKey, proxy)
	return proxy, nil
}

var _ ClusterProxy = &clusterProxyImpl{}

type clusterProxyImpl struct {
	client client.Client
}

func NewClusterProxy(client client.Client) ClusterProxy {
	return &clusterProxyImpl{
		client: client,
	}
}

func (p *clusterProxyImpl) Client() client.Client {
	return p.client
}

func (p *clusterProxyImpl) ChangefeedClient() api.ChangefeedAPIV2 {
	return api.NewChangefeedAPIV2(p.client)
}

func (p *clusterProxyImpl) PDClient() api.PDClient {
	return api.NewPdClient(p.client)
}

func (p *clusterProxyImpl) SQLClient() api.SQLClient {
	return api.NewSQLClient(p.client)
}