package dm_svc

import (
	"context"
	"strings"
	"time"

	"github.com/hashicorp/go-multierror"
	"github.com/samber/lo"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	"go.uber.org/zap"
	"k8s.io/utils/pointer"
)

var (
	removeZoneAPIRetryCode = []string{
		"EndpointOperationDenied",           // not in active state
		"EndpointZoneNotFound",              // already deleted
		"EndpointConnectionOperationDenied", // The endpoint is being connected.
	}
	deleteVpcEndpointRetryCode = []string{
		"EndpointOperationDenied",     // not in active state
		"EndpointDependenceViolation", // has dependent zone(s). remove zones in the next iter.
	}
)

// DMEndpointWorker periodically polls mgmt service for VPC endpoint status and updates our repo.
// It discovers endpoints from DM tasks' sources that use private link connection.
type DMEndpointWorker struct {
	dmTaskRepo    repo.DMTaskRepo
	mgmtSvcFacade facade.MgmtSvcFacade
}

func NewDMEndpointWorker(dmTaskRepo repo.DMTaskRepo, mgmtSvcFacade facade.MgmtSvcFacade) *DMEndpointWorker {
	return &DMEndpointWorker{dmTaskRepo: dmTaskRepo, mgmtSvcFacade: mgmtSvcFacade}
}

func (w *DMEndpointWorker) Run(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)
	start := time.Now()
	logger.Info("start to run dm endpoint worker")

	// List VPC endpoints in states that require polling of external status
	// Include pending, pendingAcceptance, and deleting.
	endpoints, bizErr := w.dmTaskRepo.ListVpcEndpoints(ctx, &entity.VpcEndpointFilter{
		States: []constant.DMVpcEndpointState{
			constant.DMVpcEndpointStatePending,
			constant.DMVpcEndpointStatePendingAcceptance,
			constant.DMVpcEndpointStateDeleting,
		},
	})
	if bizErr != nil {
		logger.Error("failed to list vpc endpoints for polling", zap.Error(bizErr))
		return bizErr
	}

	var errs error
	for _, e := range endpoints {
		if err := w.handleEndpoint(ctx, e); err != nil {
			errs = multierror.Append(errs, err)
		}
	}

	// Cleanup: find VPC endpoints matching conditions and mark them deleting
	if err := w.cleanupDanglingEndpoints(ctx); err != nil {
		errs = multierror.Append(errs, err)
	}

	logger.Info(
		"finished dm endpoint worker",
		zap.Int("endpointCount", len(endpoints)),
		zap.Float64("costSeconds", time.Since(start).Seconds()),
	)
	return errs
}

func (w *DMEndpointWorker) handleEndpoint(ctx context.Context, endpoint *entity.DMVpcEndpoint) bizerr.BizError {
	switch endpoint.State {
	case constant.DMVpcEndpointStateDeleting:
		return w.OnEndpointDeleting(ctx, endpoint)
	case constant.DMVpcEndpointStatePending, constant.DMVpcEndpointStatePendingAcceptance:
		return w.OnEndpointPolling(ctx, endpoint)
	default:
		logger := utils.GetWorkerLoggerFromContext(ctx).
			With(zap.String("endpoint_id", endpoint.ID), zap.String("cluster_id", endpoint.ClusterID))
		logger.Warn("unexpected vpc endpoint state in worker; skipping", zap.String("state", string(endpoint.State)))
		return nil
	}
}

// OnEndpointDeleting triggers provider-side deletion then refreshes status from mgmt service.
func (w *DMEndpointWorker) OnEndpointDeleting(ctx context.Context, endpoint *entity.DMVpcEndpoint) bizerr.BizError {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(zap.String("endpoint_id", endpoint.ID), zap.String("cluster_id", endpoint.ClusterID))

	// 1) Describe current endpoint; if not active, just return and wait for next loop
	desc, bizErr := w.mgmtSvcFacade.DescribeVpcEndpoint(ctx, &globalserverlessv1.DescribeVpcEndpointRequest{
		ClusterId:  endpoint.ClusterID,
		EndpointId: endpoint.EndpointId,
	})
	if bizErr != nil {
		if handled, bizError := w.handleEndpointNotFoundError(ctx, endpoint, bizErr, logger); handled ||
			bizError != nil {
			return bizError
		}
		logger.Error("describe vpc endpoint in deleting flow failed", zap.Error(bizErr))
		return bizErr
	}

	alicloudState := desc.GetEndpointDetails().GetAlicloud()
	if alicloudState == nil || !strings.EqualFold(alicloudState.EndpointState, "active") {
		// Not ready to delete yet; wait for next loop
		logger.Info("endpoint not ready for deleting yet; skip this loop", zap.Any("state", alicloudState))
		return nil
	}

	// 2) Get AZ info via ListVpcEndpointZones; if empty, delete endpoint directly
	zonesResp, zerr := w.mgmtSvcFacade.ListVpcEndpointZones(ctx, &globalserverlessv1.ListVpcEndpointZonesRequest{
		ClusterId:  endpoint.ClusterID,
		EndpointId: endpoint.EndpointId,
	})
	if zerr != nil {
		if handled, bizError := w.handleEndpointNotFoundError(ctx, endpoint, zerr, logger); handled || bizError != nil {
			return bizError
		}
		logger.Error("list vpc endpoint zones failed", zap.Error(zerr))
		return zerr
	}
	azs := zonesResp.GetZones()
	if len(azs) == 0 {
		if derr := w.mgmtSvcFacade.DeleteVpcEndpoint(ctx, &globalserverlessv1.DeleteVpcEndpointRequest{
			ClusterId:  endpoint.ClusterID,
			EndpointId: endpoint.EndpointId,
		}); derr != nil {
			if lo.SomeBy(deleteVpcEndpointRetryCode, func(code string) bool {
				return strings.Contains(derr.ErrMsg(), code)
			}) {
				logger.Warn("retry delete endpoint in next loop", zap.Error(derr))
				return nil
			}
			if handled, bizError := w.handleEndpointNotFoundError(ctx, endpoint, derr, logger); handled ||
				bizError != nil {
				return bizError
			}
			logger.Error("delete vpc endpoint via mgmt svc failed", zap.Error(derr))
			return derr
		}
		logger.Info("deleted vpc endpoint as no AZs remain")
		return nil
	}

	// 3) AZs not empty: delete the first AZ from endpoint
	// Delete the first zone by passing its ZoneId; the server uses the provided ZoneId to determine which zone to remove.
	az := azs[0]
	if derr := w.mgmtSvcFacade.RemoveZoneFromVpcEndpoint(ctx, &globalserverlessv1.RemoveZoneFromVpcEndpointRequest{
		ClusterId:  endpoint.ClusterID,
		EndpointId: endpoint.EndpointId,
		ZoneId:     az.ZoneId,
	}); derr != nil {
		if lo.SomeBy(removeZoneAPIRetryCode, func(code string) bool {
			return strings.Contains(derr.ErrMsg(), code)
		}) {
			logger.Warn("retry remove zone in next loop", zap.Error(derr))
			return nil
		}
		if handled, bizError := w.handleEndpointNotFoundError(ctx, endpoint, derr, logger); handled || bizError != nil {
			return bizError
		}
		logger.Error("remove zone from vpc endpoint via mgmt svc failed", zap.Error(derr))
		return derr
	}
	logger.Info("initiated vpc endpoint AZ deletion", zap.String("zone_id", az.ZoneId))
	return nil
}

func (w *DMEndpointWorker) handleEndpointNotFoundError(
	ctx context.Context,
	endpoint *entity.DMVpcEndpoint,
	bizErr bizerr.BizError,
	logger *zap.Logger,
) (bool, bizerr.BizError) {
	if strings.Contains(bizErr.ErrMsg(), "EndpointNotFound") {
		bizErr := w.dmTaskRepo.UpdateVpcEndpoint(
			ctx,
			endpoint.ID,
			map[string]any{"state": constant.DMVpcEndpointStateDeleted},
		)
		if bizErr != nil {
			logger.Error("update vpc endpoint state to deleted failed", zap.Error(bizErr))
			return false, bizErr
		}
		logger.Info("update vpc endpoint state to deleted succeeded")
		return true, nil
	}
	return false, nil
}

// OnEndpointPolling performs describe and updates persisted state/fields.
func (w *DMEndpointWorker) OnEndpointPolling(ctx context.Context, endpoint *entity.DMVpcEndpoint) bizerr.BizError {
	logger := utils.GetWorkerLoggerFromContext(ctx).
		With(zap.String("endpoint_id", endpoint.ID), zap.String("cluster_id", endpoint.ClusterID))

	resp, bizErr := w.mgmtSvcFacade.DescribeVpcEndpoint(ctx, &globalserverlessv1.DescribeVpcEndpointRequest{
		ClusterId:  endpoint.ClusterID,
		EndpointId: endpoint.EndpointId,
	})
	if bizErr != nil {
		if handled, bizError := w.handleEndpointNotFoundError(ctx, endpoint, bizErr, logger); handled ||
			bizError != nil {
			return bizError
		}
		logger.Error("describe vpc endpoint failed", zap.Error(bizErr))
		return bizErr
	}

	updates := map[string]any{
		"service_name":      resp.EndpointDetails.GetServiceName(),
		"state":             vo.ConvertEndpointState(resp.EndpointDetails.GetState()),
		"vpc_id":            resp.EndpointDetails.GetVpcId(),
		"creation_time_str": resp.EndpointDetails.GetCreationTimestamp(),
		"domain":            resp.EndpointDetails.GetDomain(),
	}
	if uerr := w.dmTaskRepo.UpdateVpcEndpoint(ctx, endpoint.ID, updates); uerr != nil {
		logger.Error("update vpc endpoint failed", zap.Error(uerr))
		return uerr
	}
	return nil
}

// cleanupDanglingEndpoints finds endpoints that either
// 1) have empty DMTaskID and were created more than 1 hour ago, or
// 2) have a DMTaskID, but the corresponding DMTask stage is deleted,
// and sets their state to deleting, then invokes mgmt service to delete them.
func (w *DMEndpointWorker) cleanupDanglingEndpoints(ctx context.Context) error {
	logger := utils.GetWorkerLoggerFromContext(ctx)

	cutoff := time.Now().Add(-1 * time.Hour)
	nonTerminalStates := []constant.DMVpcEndpointState{
		constant.DMVpcEndpointStatePending,
		constant.DMVpcEndpointStatePendingAcceptance,
		constant.DMVpcEndpointStateAvailable,
	}
	const batchLimit = 500

	var errs error

	// Query 1: orphan endpoints older than cutoff
	orphanFilter := &entity.VpcEndpointFilter{
		States:        nonTerminalStates,
		CreatedBefore: &cutoff,
		// use empty task id to filter orphaned endpoints
		EmptyTaskID: true,
		Limit:       pointer.Int(batchLimit),
		OrderBy:     "created_at asc",
	}

	orphanEndpoints, bizErr := w.dmTaskRepo.ListVpcEndpoints(ctx, orphanFilter)
	if bizErr != nil {
		logger.Error("failed to list orphan vpc endpoints for cleanup", zap.Error(bizErr))
		return bizErr
	}
	for _, ep := range orphanEndpoints {
		l := logger.With(
			zap.String("endpoint", ep.ID),
			zap.String("cluster_id", ep.ClusterID),
			zap.String("endpoint_id", ep.EndpointId),
		)
		if err := w.dmTaskRepo.UpdateVpcEndpoint(ctx, ep.ID, map[string]any{
			"state":        constant.DMVpcEndpointStateDeleting,
			"deleted_time": time.Now(),
		}); err != nil {
			l.Error("failed to mark orphan vpc endpoint as deleting", zap.Error(err))
			errs = multierror.Append(errs, err)
		}
		l.Info("marked vpc endpoint as deleting (orphan)")
	}

	// Query 2: endpoints whose task is deleted
	deletedTaskFilter := &entity.VpcEndpointFilter{
		States:      nonTerminalStates,
		TaskStageIn: []constant.DMTaskStage{constant.DMTaskStageDeleted},
		Limit:       pointer.Int(batchLimit),
		OrderBy:     "created_at asc",
	}
	danglingByTask, bizErr := w.dmTaskRepo.ListVpcEndpoints(ctx, deletedTaskFilter)
	if bizErr != nil {
		logger.Error("failed to list vpc endpoints linked to deleted tasks for cleanup", zap.Error(bizErr))
		return bizErr
	}

	for _, ep := range danglingByTask {
		l := logger.With(
			zap.String("endpoint", ep.ID),
			zap.String("cluster_id", ep.ClusterID),
			zap.String("endpoint_id", ep.EndpointId),
		)
		if err := w.dmTaskRepo.UpdateVpcEndpoint(ctx, ep.ID, map[string]any{
			"state":        constant.DMVpcEndpointStateDeleting,
			"deleted_time": time.Now(),
		}); err != nil {
			l.Error("failed to mark vpc endpoint as deleting (task deleted)", zap.Error(err))
			errs = multierror.Append(errs, err)
		}
		l.Info("marked vpc endpoint as deleting (task deleted)")
	}

	return errs
}