package ops_svc

import (
	"context"
	"strconv"

	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

// cdcRpcService wires the gRPC/HTTP ChangefeedOpsService handler to
// the domain CDCService. It carries no state of its own — every
// request is a thin pass-through into cdcSvc.
type cdcRpcService struct {
	cdcSvc service.CDCService
	rpc.UnimplementedChangefeedOpsServiceServer
}

// NewCdcRpcService builds the ChangefeedOpsService handler. Wired in
// via the DI container next to the other ops handlers.
func NewCdcRpcService(cdcSvc service.CDCService) rpc.ChangefeedOpsServiceServer {
	return &cdcRpcService{cdcSvc: cdcSvc}
}

func (s *cdcRpcService) ListTiCDCClusters(
	ctx context.Context, req *rpc.ListTiCDCClustersRequest,
) (*rpc.ListTiCDCClustersResponse, error) {
	clusters, err := s.cdcSvc.ListNewArchTiCDCClusters(ctx, req.GetProvider(), req.GetRegion(), req.GetState())
	if err != nil {
		return nil, err
	}
	items := make([]*rpc.TiCDCClusterInfo, 0, len(clusters))
	for _, c := range clusters {
		items = append(items, toTiCDCClusterInfo(c))
	}
	return &rpc.ListTiCDCClustersResponse{
		Items: items,
		Total: uint32(len(items)),
	}, nil
}

func (s *cdcRpcService) GetTiCDCCluster(
	ctx context.Context, req *rpc.GetTiCDCClusterRequest,
) (*rpc.TiCDCClusterInfo, error) {
	cluster, err := s.cdcSvc.GetNewArchTiCDCCluster(
		ctx, req.GetProvider(), req.GetRegion(), req.GetKubernetes(), "global", "global",
	)
	if err != nil {
		return nil, err
	}
	return toTiCDCClusterInfo(cluster), nil
}

func (s *cdcRpcService) UpgradeTiCDCCluster(
	ctx context.Context, req *rpc.UpgradeTiCDCClusterRequest,
) (*rpc.TiCDCClusterInfo, error) {
	cluster, err := s.cdcSvc.UpgradeNewArchTiCDCCluster(
		ctx,
		req.GetProvider(),
		req.GetRegion(),
		req.GetKubernetes(),
		"global",
		"global",
		req.GetTargetVersion(),
	)
	if err != nil {
		// Surface upgrade-succeeded-but-reconcile-failed cases as the
		// row-with-error: the DB write went through and the dataplane
		// will catch up via the background sync worker, so callers
		// can still observe the new intent.
		if cluster != nil {
			return toTiCDCClusterInfo(cluster), err
		}
		return nil, err
	}
	return toTiCDCClusterInfo(cluster), nil
}

func (s *cdcRpcService) MigrateChangefeed(
	ctx context.Context, req *rpc.MigrateChangefeedRequest,
) (*rpc.MigrateChangefeedResponse, error) {
	result, err := s.cdcSvc.MigrateChangefeed(
		ctx, req.GetClusterId(), req.GetChangefeedId(), req.GetRcu(), req.GetDisplayName(),
	)
	if err != nil {
		return nil, err
	}
	return toMigrateChangefeedResponse(result), nil
}

func (s *cdcRpcService) PrecheckMigrateChangefeed(
	ctx context.Context, req *rpc.PrecheckMigrateChangefeedRequest,
) (*rpc.PrecheckMigrateChangefeedResponse, error) {
	result, err := s.cdcSvc.PrecheckMigrateChangefeed(
		ctx, req.GetClusterId(), req.GetChangefeedId(), req.GetRcu(), req.GetDisplayName(),
	)
	if err != nil {
		return nil, err
	}
	checks := make([]*rpc.MigrationCheckResult, 0, len(result.Checks))
	for _, check := range result.Checks {
		checks = append(checks, &rpc.MigrationCheckResult{
			Name: check.Name, Passed: check.Passed, Message: check.Message,
		})
	}
	return &rpc.PrecheckMigrateChangefeedResponse{
		Passed:               result.Passed,
		Checks:               checks,
		SourceArch:           result.SourceArch,
		SourceState:          result.SourceState,
		CurrentCheckpointTso: result.CurrentCheckpointTSO,
		TargetRcu:            result.TargetRCU,
		TargetProvider:       result.TargetProvider,
		TargetRegion:         result.TargetRegion,
		TargetKubernetes:     result.TargetKubernetes,
	}, nil
}

func (s *cdcRpcService) CompareChangefeeds(
	ctx context.Context, req *rpc.CompareChangefeedsRequest,
) (*rpc.CompareChangefeedsResponse, error) {
	result, err := s.cdcSvc.CompareChangefeeds(
		ctx, req.GetClusterId(), req.GetSourceChangefeedId(), req.GetTargetChangefeedId(),
	)
	if err != nil {
		return nil, err
	}
	checks := make([]*rpc.MigrationCheckResult, 0, len(result.Checks))
	for _, check := range result.Checks {
		checks = append(checks, &rpc.MigrationCheckResult{
			Name: check.Name, Passed: check.Passed, Message: check.Message,
		})
	}
	return &rpc.CompareChangefeedsResponse{
		Passed:              result.Passed,
		Checks:              checks,
		SourceCheckpointTso: result.SourceCheckpointTSO,
		TargetStartTso:      result.TargetStartTSO,
		TargetCheckpointTso: result.TargetCheckpointTSO,
		SourceState:         result.SourceState,
		TargetState:         result.TargetState,
	}, nil
}

// defaultListChangefeedsLimit guards the ops list endpoint: the repo
// layer translates limit 0 into "LIMIT 0" (no rows), so an unset limit
// must be replaced before it reaches the repo.
const defaultListChangefeedsLimit = 100

func (s *cdcRpcService) ListClusterChangefeeds(
	ctx context.Context, req *rpc.ListClusterChangefeedsRequest,
) (*rpc.ListClusterChangefeedsResponse, error) {
	limit := req.GetLimit()
	if limit <= 0 {
		limit = defaultListChangefeedsLimit
	}
	changefeeds, total, err := s.cdcSvc.ListChangefeeds(ctx, &vo.ListChangefeedsOptions{
		ClusterID: req.GetClusterId(),
		SinkType:  req.GetSinkType(),
		Limit:     limit,
		Offset:    req.GetOffset(),
	})
	if err != nil {
		return nil, err
	}
	items := make([]*rpc.ChangefeedInfo, 0, len(changefeeds))
	for _, cf := range changefeeds {
		items = append(items, toChangefeedInfo(cf))
	}
	if err := s.fillTiCDCClusterNames(ctx, req.GetClusterId(), items...); err != nil {
		return nil, err
	}
	return &rpc.ListClusterChangefeedsResponse{
		Items: items,
		Total: total,
	}, nil
}

func (s *cdcRpcService) GetChangefeed(
	ctx context.Context, req *rpc.GetChangefeedRequest,
) (*rpc.ChangefeedInfo, error) {
	clusterID, err := s.resolveChangefeedClusterID(ctx, req.GetClusterId(), req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	changefeed, bizErr := s.cdcSvc.GetChangefeed(ctx, clusterID, req.GetChangefeedId())
	if bizErr != nil {
		if bizErr.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, status.Error(codes.NotFound, bizErr.Error())
		}
		return nil, bizErr
	}
	if err := s.cdcSvc.EnsureChangefeedConfiguration(ctx, clusterID, req.GetChangefeedId()); err != nil {
		// Existing changefeed metadata remains useful when Config Center is
		// unavailable or its Essential templates have not been deployed yet.
		// ApplyConfigChanges still performs a mandatory ensure before changing
		// runtime configuration.
		log.WithContextL(ctx).Warn(
			"failed to ensure changefeed configuration while reading changefeed",
			zap.String("cluster_id", clusterID),
			zap.String("changefeed_id", req.GetChangefeedId()),
			zap.Error(err),
		)
	}
	info := toChangefeedInfo(changefeed)
	if err := s.fillTiCDCClusterNames(ctx, clusterID, info); err != nil {
		return nil, err
	}
	return info, nil
}

func (s *cdcRpcService) ApplyConfigChanges(
	ctx context.Context, req *rpc.ApplyConfigChangesRequest,
) (*rpc.ApplyConfigChangesResponse, error) {
	if req.GetClusterType() != rpc.ConfigClusterType_CHANGEFEED_CLUSTER {
		return nil, status.Errorf(
			codes.InvalidArgument,
			"unsupported cluster type: %s",
			req.GetClusterType().String(),
		)
	}
	var configurationFilename string
	switch req.GetConfigurationFile() {
	case rpc.ConfigFileType_CHANGEFEED_FILE:
		configurationFilename = "changefeed"
	case rpc.ConfigFileType_SINK_URI:
		configurationFilename = "changefeed-uri"
	default:
		return nil, status.Errorf(
			codes.InvalidArgument,
			"unsupported configuration file: %s",
			req.GetConfigurationFile().String(),
		)
	}
	params := make(map[string]*config_center.ParamItem, len(req.GetCustomParams()))
	for key, item := range req.GetCustomParams() {
		params[key] = &config_center.ParamItem{
			Value:   item.GetValue(),
			Comment: item.GetComment(),
		}
	}
	clusterID, err := s.resolveChangefeedClusterID(ctx, req.GetClusterId(), req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	if err := s.cdcSvc.ApplyChangefeedConfiguration(
		ctx,
		clusterID,
		req.GetChangefeedId(),
		configurationFilename,
		params,
		req.TemplateId,
	); err != nil {
		return nil, err
	}
	return &rpc.ApplyConfigChangesResponse{}, nil
}

func (s *cdcRpcService) PauseChangefeed(
	ctx context.Context, req *rpc.PauseChangefeedRequest,
) (*rpc.ChangefeedInfo, error) {
	clusterID, err := s.resolveChangefeedClusterID(ctx, req.GetClusterId(), req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	existing, err := s.getAndCheckExpectedArch(ctx, clusterID, req.GetChangefeedId(), req.GetExpectedArch())
	if err != nil {
		return nil, err
	}
	if existing.State == constant.ChangefeedStatePaused {
		return toChangefeedInfo(existing), nil
	}
	changefeed, err := s.cdcSvc.StopChangefeed(ctx, clusterID, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	info := toChangefeedInfo(changefeed)
	if err := s.fillTiCDCClusterNames(ctx, clusterID, info); err != nil {
		return nil, err
	}
	return info, nil
}

func (s *cdcRpcService) ResumeChangefeed(
	ctx context.Context, req *rpc.ResumeChangefeedRequest,
) (*rpc.ChangefeedInfo, error) {
	clusterID, err := s.resolveChangefeedClusterID(ctx, req.GetClusterId(), req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	existing, err := s.getAndCheckExpectedArch(ctx, clusterID, req.GetChangefeedId(), req.GetExpectedArch())
	if err != nil {
		return nil, err
	}
	if existing.State == constant.ChangfeedStateRunning {
		return toChangefeedInfo(existing), nil
	}
	changefeed, err := s.cdcSvc.StartChangefeed(ctx, clusterID, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	info := toChangefeedInfo(changefeed)
	if err := s.fillTiCDCClusterNames(ctx, clusterID, info); err != nil {
		return nil, err
	}
	return info, nil
}

func (s *cdcRpcService) DeleteChangefeed(
	ctx context.Context, req *rpc.DeleteChangefeedRequest,
) (*rpc.ChangefeedInfo, error) {
	if _, err := s.getAndCheckExpectedArch(
		ctx, req.GetClusterId(), req.GetChangefeedId(), req.GetExpectedArch(),
	); err != nil {
		return nil, err
	}
	changefeed, err := s.cdcSvc.DeleteChangefeed(ctx, req.GetClusterId(), req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	return toChangefeedInfo(changefeed), nil
}

func (s *cdcRpcService) getAndCheckExpectedArch(
	ctx context.Context, clusterID, changefeedID, expectedArch string,
) (*vo.Changefeed, error) {
	changefeed, err := s.cdcSvc.GetChangefeed(ctx, clusterID, changefeedID)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return nil, status.Error(codes.NotFound, err.Error())
		}
		return nil, err
	}
	if expectedArch != "" && string(changefeed.Arch) != expectedArch {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"changefeed %s arch is %s, expected %s",
			changefeedID,
			changefeed.Arch,
			expectedArch,
		)
	}
	return changefeed, nil
}

func (s *cdcRpcService) RestartChangefeed(
	ctx context.Context, req *rpc.RestartChangefeedRequest,
) (*rpc.ChangefeedInfo, error) {
	clusterID, err := s.resolveChangefeedClusterID(ctx, req.GetClusterId(), req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	if _, err := s.cdcSvc.StopChangefeed(ctx, clusterID, req.GetChangefeedId()); err != nil {
		return nil, err
	}
	changefeed, err := s.cdcSvc.StartChangefeed(ctx, clusterID, req.GetChangefeedId())
	if err != nil {
		return nil, err
	}
	info := toChangefeedInfo(changefeed)
	if err := s.fillTiCDCClusterNames(ctx, clusterID, info); err != nil {
		return nil, err
	}
	return info, nil
}

func (s *cdcRpcService) resolveChangefeedClusterID(
	ctx context.Context, clusterID, changefeedID string,
) (string, error) {
	if clusterID != "" && clusterID != changefeedID {
		return clusterID, nil
	}
	changefeed, err := s.cdcSvc.GetChangefeedByID(ctx, changefeedID)
	if err != nil {
		if err.ErrCode() == bizerr.ErrCodeRecordNotFound {
			return "", status.Error(codes.NotFound, err.Error())
		}
		return "", err
	}
	return changefeed.ClusterID, nil
}

func (s *cdcRpcService) ScaleChangefeed(
	ctx context.Context, req *rpc.ScaleChangefeedRequest,
) (*rpc.ChangefeedInfo, error) {
	changefeed, err := s.cdcSvc.ScaleChangefeed(ctx, req.GetClusterId(), req.GetChangefeedId(), req.GetRcu())
	if err != nil {
		return nil, err
	}
	info := toChangefeedInfo(changefeed)
	if err := s.fillTiCDCClusterNames(ctx, req.GetClusterId(), info); err != nil {
		return nil, err
	}
	return info, nil
}

func (s *cdcRpcService) ListClusterCDCJobs(
	ctx context.Context, req *rpc.ListClusterCDCJobsRequest,
) (*rpc.ListClusterCDCJobsResponse, error) {
	statuses, err := s.cdcSvc.ListClusterCDCJobs(ctx, req.GetClusterId())
	if err != nil {
		return nil, err
	}
	items := make([]*rpc.ClusterCDCJobInfo, 0, len(statuses))
	for _, st := range statuses {
		items = append(items, toClusterCDCJobInfo(st))
	}
	return &rpc.ListClusterCDCJobsResponse{
		Items: items,
		Total: uint32(len(items)),
	}, nil
}

func (s *cdcRpcService) DeleteClusterCDCJob(
	ctx context.Context, req *rpc.DeleteClusterCDCJobRequest,
) (*rpc.DeleteClusterCDCJobResponse, error) {
	if err := s.cdcSvc.DeleteClusterCDCJob(ctx, req.GetClusterId(), req.GetStatusId()); err != nil {
		return nil, err
	}
	return &rpc.DeleteClusterCDCJobResponse{}, nil
}

func toTiCDCClusterInfo(c *entity.TiCDCCluster) *rpc.TiCDCClusterInfo {
	if c == nil {
		return &rpc.TiCDCClusterInfo{}
	}
	info := &rpc.TiCDCClusterInfo{
		Provider:              c.CloudProvider,
		Region:                c.Region,
		Kubernetes:            c.Kubernetes,
		TicdcClusterName:      c.TiCDCClusterName,
		Version:               c.Version,
		ExpectedSpecification: c.ExpectedSpec,
		ActualSpecification:   c.ActualSpec,
		State:                 string(c.State),
		CreatedAt:             c.CreatedAt.UTC().Format("2006-01-02T15:04:05Z"),
	}
	if c.UpdatedAt != nil {
		info.UpdatedAt = c.UpdatedAt.UTC().Format("2006-01-02T15:04:05Z")
	}
	if c.DeletedAt != nil {
		info.DeletedAt = c.DeletedAt.UTC().Format("2006-01-02T15:04:05Z")
	}
	return info
}

// isLegacyCDCClusterLock reports whether the status is the old-arch cluster-level
// CDC lock: a non-expired CDC-typed status whose own mutex still lists CDC. That
// self-mutex is what blocks newarch per-changefeed registration. Per-changefeed jobs
// are CDC-typed too but carry an empty status mutex, so they are excluded.
func isLegacyCDCClusterLock(s *tidbmgmtservicev1.ClusterCustomStatus) bool {
	if s.GetStatusType() != tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_CDC || s.GetIsExpired() {
		return false
	}
	for _, t := range s.GetMutex().GetStatusType() {
		if t == tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_CDC {
			return true
		}
	}
	return false
}

func toClusterCDCJobInfo(s *tidbmgmtservicev1.ClusterCustomStatus) *rpc.ClusterCDCJobInfo {
	if s == nil {
		return &rpc.ClusterCDCJobInfo{}
	}
	mutexStatusTypes := make([]string, 0, len(s.GetMutex().GetStatusType()))
	for _, t := range s.GetMutex().GetStatusType() {
		mutexStatusTypes = append(mutexStatusTypes, t.String())
	}
	mutexClusterActions := make([]string, 0, len(s.GetMutex().GetClusterAction()))
	for _, a := range s.GetMutex().GetClusterAction() {
		mutexClusterActions = append(mutexClusterActions, a.String())
	}
	return &rpc.ClusterCDCJobInfo{
		StatusId:            strconv.FormatUint(s.GetStatusId(), 10),
		StatusName:          s.GetStatusName(),
		StatusType:          s.GetStatusType().String(),
		MutexStatusTypes:    mutexStatusTypes,
		MutexClusterActions: mutexClusterActions,
		IsLegacyClusterLock: isLegacyCDCClusterLock(s),
		IsExpired:           s.GetIsExpired(),
	}
}

// fillTiCDCClusterNames sets ticdc_cluster_name on the newarch entries
// of infos with one batched domain call; legacy replication_worker
// entries are skipped and keep the field empty.
func (s *cdcRpcService) fillTiCDCClusterNames(
	ctx context.Context, clusterID string, infos ...*rpc.ChangefeedInfo,
) error {
	ids := make([]string, 0, len(infos))
	for _, info := range infos {
		if info.GetArch() == string(constant.ChangefeedArchTiCDCNewArch) {
			ids = append(ids, info.GetChangefeedId())
		}
	}
	if len(ids) == 0 {
		return nil
	}
	names, err := s.cdcSvc.GetChangefeedTiCDCClusterNames(ctx, clusterID, ids)
	if err != nil {
		return err
	}
	for _, info := range infos {
		info.TicdcClusterName = names[info.GetChangefeedId()]
	}
	return nil
}

func toChangefeedInfo(changefeed *vo.Changefeed) *rpc.ChangefeedInfo {
	if changefeed == nil {
		return &rpc.ChangefeedInfo{}
	}
	info := &rpc.ChangefeedInfo{
		ClusterId:    changefeed.ClusterID,
		ChangefeedId: changefeed.ChangefeedID,
		State:        string(changefeed.State),
		Arch:         string(changefeed.Arch),
		Rcu:          changefeed.RCU,
		DisplayName:  changefeed.Name,
		Error:        changefeed.Error,
		Provider:     changefeed.CloudProvider,
		Region:       changefeed.Region,
		Kubernetes:   changefeed.Kubernetes,
	}
	if changefeed.Sink != nil {
		info.SinkType = string(changefeed.Sink.Type)
	}
	if changefeed.Checkpoint != nil {
		info.CheckpointTso = changefeed.Checkpoint.TSO
		if changefeed.Checkpoint.Time != nil {
			info.CheckpointTime = changefeed.Checkpoint.Time.UTC().Format("2006-01-02T15:04:05Z")
		}
	}
	if changefeed.CreateTime != nil {
		info.CreatedAt = changefeed.CreateTime.UTC().Format("2006-01-02T15:04:05Z")
	}
	if changefeed.UpdateTime != nil {
		info.UpdatedAt = changefeed.UpdateTime.UTC().Format("2006-01-02T15:04:05Z")
	}
	if changefeed.StartPosition != nil {
		info.StartTso = changefeed.StartPosition.TSO
	}
	return info
}

func toMigrateChangefeedResponse(result *vo.ChangefeedMigrationResult) *rpc.MigrateChangefeedResponse {
	if result == nil || result.Target == nil {
		return &rpc.MigrateChangefeedResponse{}
	}
	changefeed := result.Target
	return &rpc.MigrateChangefeedResponse{
		ClusterId:           changefeed.ClusterID,
		ChangefeedId:        changefeed.ChangefeedID,
		State:               string(changefeed.State),
		Arch:                string(changefeed.Arch),
		Rcu:                 changefeed.RCU,
		DisplayName:         changefeed.Name,
		SourceChangefeedId:  result.SourceChangefeedID,
		TargetChangefeedId:  changefeed.ChangefeedID,
		SourceCheckpointTso: result.SourceCheckpointTSO,
		TargetStartTso:      result.SourceCheckpointTSO,
	}
}