package applier

import (
	"context"
	"fmt"
	"strconv"

	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	"github.com/tidbcloud/devops/pkg/infra/clients/configcentercli"
	"github.com/tidbcloud/devops/pkg/infra/clients/dataflowcli"
	"github.com/tidbcloud/devops/pkg/infra/log"
	"github.com/tidbcloud/devops/pkg/infra/utils/pointer"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"google.golang.org/protobuf/encoding/protojson"
)

type DataflowApplierRequest struct {
	ClusterType dataflow.ConfigClusterType
	ClusterId   uint64
	Requests    []*dataflow.ApplyConfigChangesReq
}

func (r *DataflowApplierRequest) Modify(configType dataflow.ConfigFileType, set func(req *dataflow.ApplyConfigChangesReq) error) error {
	var req *dataflow.ApplyConfigChangesReq

	for _, tmp := range r.Requests {
		if tmp.ConfigurationFile == configType {
			req = tmp
			break
		}
	}

	if req == nil {
		req = &dataflow.ApplyConfigChangesReq{
			ClusterType:       r.ClusterType,
			ClusterId:         r.ClusterId,
			ConfigurationFile: configType,
		}
		r.Requests = append(r.Requests, req)
	}

	return set(req)
}

type DataflowApplier struct {
	dataflow     *dataflowcli.DataflowCli
	configcenter *configcentercli.Client
}

func NewDataflowApplier(dataflow *dataflowcli.DataflowCli, configcenter *configcentercli.Client) *DataflowApplier {
	return &DataflowApplier{
		dataflow:     dataflow,
		configcenter: configcenter,
	}
}

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

func (a *DataflowApplier) AcceptVersionChange(verChange *ccent.ConfigurationVersionChange, changeType string) bool {
	return (changeType == string(api.ChangeTypeTiCDC)) || (changeType == string(api.ChangeTypeTidb))
}

func (a *DataflowApplier) AcceptComponentChange(compChg *ccent.ConfigurationComponentChange) bool {
	return a.acceptComponent(compChg.Component)
}

func (a *DataflowApplier) AcceptFileChange(fileChg *ccent.ConfigurationFileChange) bool {
	return a.acceptFile(fileChg.ConfigType)
}

func (a *DataflowApplier) acceptFile(configType string) bool {
	return cvt.To.Dataflow.ConfigType(configType) != dataflow.ConfigFileType_CONFIG_FILE_TYPE_UNSPECIFIED
}

func (a *DataflowApplier) acceptComponent(comp string) bool {
	return cvt.To.Dataflow.Component(comp) != dataflow.ConfigFileType_CONFIG_FILE_TYPE_UNSPECIFIED
}

func (a *DataflowApplier) Prepare(ctx context.Context, actx *ApplyContext) error {
	chg := actx.Input.Change

	clusterIDInt, err := strconv.ParseUint(chg.TargetCluster.ClusterID, 10, 64)
	if err != nil {
		return fmt.Errorf("parse cluster id to int failed: %w", err)
	}

	actx.DataflowApplierRequest = &DataflowApplierRequest{
		ClusterType: cvt.To.Dataflow.ClusterType(chg.TargetCluster.ClusterType),
		ClusterId:   clusterIDInt,
		Requests:    make([]*dataflow.ApplyConfigChangesReq, 0),
	}
	request := actx.DataflowApplierRequest

	// update requests from file changes
	for _, fileChg := range actx.Input.Files {
		err := request.Modify(cvt.To.Dataflow.ConfigType(fileChg.ConfigType), func(req *dataflow.ApplyConfigChangesReq) error {
			if fileChg.ModifiedTemplate != nil {
				req.TemplateId = &fileChg.ModifiedTemplate.To
			}

			req.CustomParams = make(map[string]*dataflow.ParamItem)
			for _, item := range fileChg.AllParameters {
				if item.Source != ccent.ParameterCustom { // skip non-custom parameters
					continue
				}
				param, err := cvt.To.Dataflow.NamedParameter(item.NamedParameter)
				if err != nil {
					return fmt.Errorf("convert cluster config item failed: %w", err)
				}

				req.CustomParams[item.Name] = param
			}

			return nil
		})

		if err != nil {
			return err
		}
	}

	// update request from component change
	for _, compChg := range actx.Input.Components {
		err := request.Modify(cvt.To.Dataflow.Component(compChg.Component), func(req *dataflow.ApplyConfigChangesReq) error {
			if actx.Input.Version != nil {
				req.Version = &actx.Input.Version.To
			}

			if compChg.ModifiedOverwriteImageTag != nil {
				req.CustomImageTag = pointer.Pointer(compChg.ModifiedOverwriteImageTag.To)
			}

			return nil
		})

		if err != nil {
			return err
		}
	}
	return nil
}

func (a *DataflowApplier) Apply(ctx context.Context, actx *ApplyContext) error {
	logger := log.FromContext(ctx).WithName(a.Name())

	for _, req := range actx.DataflowApplierRequest.Requests {
		if actx.ApplyMode.Appliable(api.DataflowApply) {
			logger.Info("applying cluster config change to dataflow service", "req", protojson.Format(req))
			_, err := a.dataflow.Ops.ApplyConfigChanges(ctx, req)
			if err != nil {
				return fmt.Errorf("apply cluster config via dataflow failed: %w", err)
			}
		} else {
			logger.Info("skip apply via dataflow cause apply method missed in apply mode")
		}

	}

	return nil
}

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