package impl

import (
	"context"
	"strings"

	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/replay"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api"
	replay_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/cluster-server/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

// replayOperatorImpl implement the operations to interact with the actual Replay resource
type replayOperatorImpl struct {
	MgmtSvcGrpcClient       grpcli.MgmtSvcGrpcClient
	ReplayRegionalAPIClient regionalapi.ReplayClient
	encryptor               encryptor_wrapper.Encryptor
}

func NewReplayOperator(
	mgmtSvcGrpcClient grpcli.MgmtSvcGrpcClient,
	replayCli regionalapi.ReplayClient,
	encryptor encryptor_wrapper.Encryptor,
) replay.ReplayOperator {
	return &replayOperatorImpl{
		MgmtSvcGrpcClient:       mgmtSvcGrpcClient,
		ReplayRegionalAPIClient: replayCli,
		encryptor:               encryptor,
	}
}

// Deploy deploy the Replay resource
//
//nolint:dupl
func (o *replayOperatorImpl) Deploy(
	ctx context.Context, req *domain.DeployReplayReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("deploying replay resource")

	replayProto, regionInfo, bizErr := o.prepareReplayRequest(ctx, req.Replay, entity.ReplayDesiredStateDeploy)
	if bizErr != nil {
		return bizErr.Chain("convert to replay proto")
	}

	// call the regional API to create Replay
	_, bizErr = o.ReplayRegionalAPIClient.CreateReplay(ctx,
		regionInfo,
		&replay_v1.CreateReplayRequest{Replay: replayProto})
	if bizErr != nil {
		return bizErr.Chain("create replay")
	}

	return nil
}

// Update update the Replay resource
//
//nolint:dupl
func (o *replayOperatorImpl) Modify(
	ctx context.Context, req *domain.ModifyReplayReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("modifying replay resource")

	replayProto, regionInfo, bizErr := o.prepareReplayRequest(ctx, req.Replay, req.State)
	if bizErr != nil {
		return bizErr.Chain("convert to replay proto")
	}

	// call the regional API to update Replay
	_, bizErr = o.ReplayRegionalAPIClient.UpdateReplay(ctx,
		regionInfo,
		&replay_v1.UpdateReplayRequest{Replay: replayProto})
	if bizErr != nil {
		return bizErr.Chain("modify replay")
	}

	return nil
}

// Start start the Replay task
func (o *replayOperatorImpl) Start(
	ctx context.Context, req *domain.StartReplayReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("starting replay task")

	regionInfo, bizErr := o.buildReplayRouteRegion(ctx, req.Replay)
	if bizErr != nil {
		return bizErr.Chain("build replay route region")
	}

	// call the regional API to start Replay
	_, bizErr = o.ReplayRegionalAPIClient.StartReplay(ctx,
		regionInfo,
		&replay_v1.StartReplayRequest{
			ReplayId: domain.BuildRegionalReplayID(req.Replay.ID),
		})
	if bizErr != nil {
		return bizErr.Chain("start replay")
	}

	return nil
}

// Stop stop the Replay task
func (o *replayOperatorImpl) Stop(
	ctx context.Context, req *domain.StopReplayReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("stopping replay task")

	regionInfo, bizErr := o.buildReplayRouteRegion(ctx, req.Replay)
	if bizErr != nil {
		return bizErr.Chain("build replay route region")
	}

	// call the regional API to stop Replay
	_, bizErr = o.ReplayRegionalAPIClient.StopReplay(ctx,
		regionInfo,
		&replay_v1.StopReplayRequest{
			ReplayId: domain.BuildRegionalReplayID(req.Replay.ID),
		})
	if bizErr != nil {
		return bizErr.Chain("stop replay")
	}

	return nil
}

// Destroy destroy the Replay resource
func (o *replayOperatorImpl) Destroy(
	ctx context.Context, req *domain.DestroyReplayReq,
) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("destroying replay resource")

	regionInfo, bizErr := o.buildReplayRouteRegion(ctx, req.Replay)
	if bizErr != nil {
		return bizErr.Chain("build replay route region")
	}

	// call the regional API to delete Replay
	_, bizErr = o.ReplayRegionalAPIClient.DeleteReplay(ctx,
		regionInfo,
		&replay_v1.DeleteReplayRequest{
			ReplayId: domain.BuildRegionalReplayID(req.Replay.ID),
		})
	if bizErr != nil {
		return bizErr.Chain("delete replay")
	}

	return nil
}

// GetStatus get the Replay resource status
func (o *replayOperatorImpl) GetStatus(
	ctx context.Context, req *domain.GetReplayStatusReq,
) (*domain.GetReplayStatusResp, bizerr.BizError) {
	logger := log.GetLogger(ctx)
	logger.Debug("getting replay status")

	regionInfo, bizErr := o.buildReplayRouteRegion(ctx, req.Replay)
	if bizErr != nil {
		return nil, bizErr.Chain("build replay route region")
	}

	// call the regional API to get the Replay status
	resp, bizErr := o.ReplayRegionalAPIClient.GetReplay(ctx,
		regionInfo,
		&replay_v1.GetReplayRequest{
			ReplayId: domain.BuildRegionalReplayID(req.Replay.ID),
		})
	if bizErr != nil {
		return nil, bizErr.Chain("get replay status")
	}

	// use the domain layer reverse conversion method
	replayStatus := domain.FromReplayStatusProto(resp.GetStatus())
	var phase entity.ReplayPhase
	var message string

	if replayStatus != nil {
		phase = replayStatus.Phase
		message = replayStatus.Message
	}

	return &domain.GetReplayStatusResp{
		Phase:   phase,
		Status:  replayStatus,
		Message: message,
		State:   domain.FromReplayStateProto(resp.Replay.State),
	}, nil
}
func (o *replayOperatorImpl) prepareReplayRequest(
	ctx context.Context,
	replayEnt *entity.Replay,
	desiredState entity.ReplayDesiredState,
) (*replay_v1.Replay, *regionalapi.Region, bizerr.BizError) {
	if replayEnt == nil {
		return nil, nil, bizerr.New(bizerr.ErrCodeParams, "replay is required")
	}

	var sourceCluster *mgmt.Cluster
	var bizErr bizerr.BizError
	if replayEnt.SourceClusterID.Uint64() > 0 {
		sourceCluster, bizErr = o.loadReplayClusterRequired(ctx, replayEnt.SourceClusterID)
		if bizErr != nil {
			return nil, nil, bizErr
		}
	}
	targetCluster, bizErr := o.loadReplayClusterRequired(ctx, replayEnt.TargetClusterID)
	if bizErr != nil {
		return nil, nil, bizErr
	}

	preparedReplay := cloneReplayForExecution(replayEnt)
	if bizErr := decryptReplayAuditLogSecretFields(o.encryptor, preparedReplay, auditLogSecretScopeExecution); bizErr != nil {
		return nil, nil, bizErr.Chain("decrypt replay secret fields")
	}

	hasExplicitTasks := len(preparedReplay.ReplayTasks) > 0
	sourceProvider, bizErr := newReplayClusterProvider(o, sourceCluster, preparedReplay)
	if bizErr != nil {
		return nil, nil, bizErr.Chain("resolve source replay provider")
	}
	var sourceAuditLogAddr *entity.AuditLogAddr
	if sourceProvider != nil {
		if hasExplicitTasks {
			if bizErr := sourceProvider.PrepareTaskSourceAuditLogs(preparedReplay.ReplayTasks); bizErr != nil {
				return nil, nil, bizErr.Chain("prepare task source audit logs")
			}
		} else {
			sourceAuditLogAddr, bizErr = sourceProvider.PrepareSourceAuditLog(ctx)
			if bizErr != nil {
				return nil, nil, bizErr.Chain("prepare source audit log")
			}
		}
	}

	targetProvider, bizErr := newReplayClusterProvider(o, targetCluster, preparedReplay)
	if bizErr != nil {
		return nil, nil, bizErr.Chain("resolve target replay provider")
	}
	var targetEndpoint string
	if targetProvider != nil && !hasExplicitTasks {
		targetEndpoint, bizErr = targetProvider.PrepareTargetEndpoint()
		if bizErr != nil {
			return nil, nil, bizErr.Chain("cluster target %d prepare target endpoint", targetCluster.GetClusterId())
		}
	}
	// Manual task mode owns source audit-log inputs at task level, so
	// sourceAuditLogAddr intentionally stays nil and ToReplayProto omits
	// Defaults.SourceAuditLog. Auto mode forwards the prepared replay-level
	// source audit-log material as Defaults.SourceAuditLog.
	preparedReplay.SourceAuditCredential = sourceAuditLogAddr

	routeCluster, bizErr := resolveReplayDeploymentRoute(preparedReplay, sourceCluster, targetCluster)
	if bizErr != nil {
		return nil, nil, bizErr
	}
	replayProto, bizErr := domain.ToReplayProto(
		preparedReplay,
		sourceCluster,
		targetCluster,
		desiredState,
		o.encryptor,
		normalizeReplayProvider(routeCluster.GetCloudProvider()),
		routeCluster.GetRegion(),
		targetEndpoint,
	)
	if bizErr != nil {
		return nil, nil, bizErr
	}

	return replayProto, buildRegionInfo(preparedReplay, routeCluster), nil
}

// buildRegionInfo uses the resolved route cluster to address the infra-provider
// region that owns the replay CR.
// Examples:
//   - target-owned replay uses the target cluster provider and region.
//   - source-owned replay uses the source cluster provider and region.
func buildRegionInfo(replay *entity.Replay, routeCluster *mgmt.Cluster) *regionalapi.Region {
	if replay == nil {
		return nil
	}
	cloudProvider := replay.CloudProvider
	region := replay.Region
	if routeCluster != nil {
		cloudProvider = routeCluster.GetCloudProvider()
		region = routeCluster.GetRegion()
	}
	regionInfo := &regionalapi.Region{
		TenantID:      replay.TenantID.Uint64(),
		CloudProvider: cloudProvider,
		Name:          region,
	}
	if routeCluster != nil && routeCluster.GetDeployType() == mgmt.DeployType_DEPLOY_TYPE_NEXTGEN {
		regionInfo.DeployType = routeCluster.GetDeployType()
		regionInfo.ServicePlan = routeCluster.GetNextGenCluster().GetServicePlan()
	}
	return regionInfo
}

// resolveReplayDeploymentRoute chooses whether the source or target cluster owns
// the replay CR. Replay placement must match exactly one side by tenant,
// project, provider, and region.
func resolveReplayDeploymentRoute(
	replayEnt *entity.Replay,
	sourceCluster *mgmt.Cluster,
	targetCluster *mgmt.Cluster,
) (*mgmt.Cluster, bizerr.BizError) {
	if replayClusterMatchesPlacement(replayEnt, targetCluster) {
		return targetCluster, nil
	}
	if replayClusterMatchesPlacement(replayEnt, sourceCluster) {
		return sourceCluster, nil
	}
	return nil, bizerr.New(
		bizerr.ErrCodeParams,
		"replay deployment placement does not match source or target cluster",
	)
}

// replayClusterMatchesPlacement compares replay placement fields against a
// management cluster after normalizing provider aliases.
func replayClusterMatchesPlacement(replayEnt *entity.Replay, cluster *mgmt.Cluster) bool {
	if replayEnt == nil || cluster == nil {
		return false
	}
	if replayEnt.TenantID.Uint64() != cluster.GetOrganizationId() ||
		replayEnt.ProjectID.Uint64() != cluster.GetProjectId() ||
		replayEnt.Region != cluster.GetRegion() {
		return false
	}
	return normalizeReplayProvider(replayEnt.CloudProvider) == normalizeReplayProvider(cluster.GetCloudProvider())
}

func normalizeReplayProvider(provider string) string {
	infraProvider := domain.CloudProviderToInfraProvider(provider)
	if infraProvider != "" {
		return infraProvider
	}
	return strings.ToLower(strings.TrimSpace(provider))
}

func (o *replayOperatorImpl) buildReplayRouteRegion(
	ctx context.Context,
	replayEnt *entity.Replay,
) (*regionalapi.Region, bizerr.BizError) {
	if replayEnt == nil {
		return nil, nil
	}

	routeCluster, bizErr := o.loadReplayClusterForRoute(ctx, replayEnt)
	if bizErr != nil {
		return nil, bizErr
	}
	if routeCluster == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "replay route cluster is required")
	}

	return buildRegionInfo(replayEnt, routeCluster), nil
}

// loadReplayClusterForRoute fetches source and target as route candidates.
// A missing candidate is tolerated so lifecycle calls can still reach the side
// that owns the replay CR.
func (o *replayOperatorImpl) loadReplayClusterForRoute(
	ctx context.Context,
	replayEnt *entity.Replay,
) (*mgmt.Cluster, bizerr.BizError) {
	if replayEnt == nil {
		return nil, nil
	}

	var targetCluster *mgmt.Cluster
	if replayEnt.TargetClusterID.Uint64() > 0 {
		var bizErr bizerr.BizError
		targetCluster, bizErr = o.MgmtSvcGrpcClient.GetAnyClusterByID(ctx, replayEnt.TargetClusterID.Uint64())
		if bizErr != nil {
			if bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
				return nil, bizErr.Chain("get replay route cluster %d info", replayEnt.TargetClusterID.Uint64())
			}
			targetCluster = nil
		}
	}

	var sourceCluster *mgmt.Cluster
	if replayEnt.SourceClusterID.Uint64() > 0 {
		var bizErr bizerr.BizError
		sourceCluster, bizErr = o.MgmtSvcGrpcClient.GetAnyClusterByID(ctx, replayEnt.SourceClusterID.Uint64())
		if bizErr != nil {
			if bizErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
				return nil, bizErr.Chain("get replay route cluster %d info", replayEnt.SourceClusterID.Uint64())
			}
			sourceCluster = nil
		}
	}

	routeCluster, routeErr := resolveReplayDeploymentRoute(replayEnt, sourceCluster, targetCluster)
	if routeErr != nil {
		return nil, routeErr
	}
	return routeCluster, nil
}

func (o *replayOperatorImpl) loadReplayClusterRequired(
	ctx context.Context,
	clusterID entity.ID,
) (*mgmt.Cluster, bizerr.BizError) {
	if clusterID.Uint64() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	cluster, bizErr := o.MgmtSvcGrpcClient.GetClusterByID(ctx, clusterID.Uint64())
	if bizErr != nil {
		return nil, bizErr.Chain("get replay cluster %d info", clusterID.Uint64())
	}
	return cluster, nil
}