package applier

import (
	"context"
	"fmt"
	"math"

	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	"github.com/tidbcloud/devops/pkg/infra/clients/serverlesscli"
	"github.com/tidbcloud/devops/pkg/infra/log"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"
	serverlessrpc "github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"google.golang.org/protobuf/encoding/protojson"
)

type ServerlessChangefeedApplierRequest struct {
	ChangefeedID string
	Requests     []*serverlessrpc.ApplyConfigChangesRequest
}

func (r *ServerlessChangefeedApplierRequest) Modify(
	configType serverlessrpc.ConfigFileType,
	set func(req *serverlessrpc.ApplyConfigChangesRequest) error,
) error {
	var req *serverlessrpc.ApplyConfigChangesRequest
	for _, candidate := range r.Requests {
		if candidate.ConfigurationFile == configType {
			req = candidate
			break
		}
	}
	if req == nil {
		req = &serverlessrpc.ApplyConfigChangesRequest{
			ClusterType:       serverlessrpc.ConfigClusterType_CHANGEFEED_CLUSTER,
			ClusterId:         r.ChangefeedID,
			ChangefeedId:      r.ChangefeedID,
			ConfigurationFile: configType,
		}
		r.Requests = append(r.Requests, req)
	}
	return set(req)
}

type ServerlessChangefeedApplier struct {
	serverless *serverlesscli.Client
}

func NewServerlessChangefeedApplier(serverless *serverlesscli.Client) *ServerlessChangefeedApplier {
	return &ServerlessChangefeedApplier{serverless: serverless}
}

func (a *ServerlessChangefeedApplier) Name() string {
	return "ServerlessChangefeedApplier"
}

func (a *ServerlessChangefeedApplier) AcceptVersionChange(*ccent.ConfigurationVersionChange, string) bool {
	return false
}

func (a *ServerlessChangefeedApplier) AcceptComponentChange(*ccent.ConfigurationComponentChange) bool {
	return false
}

func (a *ServerlessChangefeedApplier) AcceptFileChange(fileChg *ccent.ConfigurationFileChange) bool {
	return cvt.To.Serverless.ConfigType(fileChg.ConfigType) != serverlessrpc.ConfigFileType_CONFIG_FILE_TYPE_UNSPECIFIED
}

func (a *ServerlessChangefeedApplier) Prepare(_ context.Context, actx *ApplyContext) error {
	chg := actx.Input.Change
	if chg.TargetCluster == nil {
		return fmt.Errorf("target cluster is required")
	}
	request := &ServerlessChangefeedApplierRequest{
		ChangefeedID: chg.TargetCluster.ClusterID,
		Requests:     make([]*serverlessrpc.ApplyConfigChangesRequest, 0),
	}
	actx.ServerlessChangefeedApplierRequest = request

	for _, fileChg := range actx.Input.Files {
		if err := request.Modify(cvt.To.Serverless.ConfigType(fileChg.ConfigType), func(req *serverlessrpc.ApplyConfigChangesRequest) error {
			if fileChg.ModifiedTemplate != nil {
				if fileChg.ModifiedTemplate.To > math.MaxInt64 {
					return fmt.Errorf("template id %d exceeds maximum supported value", fileChg.ModifiedTemplate.To)
				}
				templateID := int64(fileChg.ModifiedTemplate.To)
				req.TemplateId = &templateID
			}
			req.CustomParams = make(map[string]*serverlessrpc.ParamItem)
			for _, item := range fileChg.AllParameters {
				if item.Source != ccent.ParameterCustom {
					continue
				}
				param, err := cvt.To.Serverless.NamedParameter(item.NamedParameter)
				if err != nil {
					return fmt.Errorf("convert cluster config item failed: %w", err)
				}
				req.CustomParams[item.Name] = param
			}
			return nil
		}); err != nil {
			return err
		}
	}
	return nil
}

func (a *ServerlessChangefeedApplier) Apply(ctx context.Context, actx *ApplyContext) error {
	logger := log.FromContext(ctx).WithName(a.Name())
	for _, req := range actx.ServerlessChangefeedApplierRequest.Requests {
		if !actx.ApplyMode.Appliable(api.ServerlessApply) {
			logger.Info("skip apply via serverless service cause apply method missed in apply mode")
			continue
		}
		logger.Info("applying changefeed config change to serverless service", "req", protojson.Format(req))
		if _, err := a.serverless.ApplyChangefeedConfigChanges(ctx, req); err != nil {
			return fmt.Errorf("apply changefeed config via serverless service failed: %w", err)
		}
	}
	return nil
}

func (a *ServerlessChangefeedApplier) CheckResult(context.Context, *CheckResultContext) (bool, string, error) {
	return true, "", nil
}