package ticdc

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	serverlessrpc "github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"go.uber.org/zap"
	"google.golang.org/grpc/status"
)

const (
	changefeedListPageSize = 1000
	listTimeout            = 5 * time.Second
	detailTimeout          = 10 * time.Second
)

type Service struct {
	clients *Clients
}

func NewService(clients *Clients) *Service {
	return &Service{clients: clients}
}

func ResolveArchitecture(deployType, deployTypeV2 string) (Architecture, error) {
	legacy := strings.ToLower(strings.TrimSpace(deployType))
	v2 := strings.ToLower(strings.TrimSpace(deployTypeV2))
	// Keep this mapping aligned with DevOps TiCDC metadata ownership:
	// Dedicated -> Dataflow; Premium/BYOC/shared pool -> Dataflow-NG;
	// Essential -> Serverless ChangefeedOpsService. Check the v2 plan first,
	// because Essential clusters retain legacy deployType=shared.
	if legacy == "essential" || v2 == "essential" {
		return ArchitectureEssential, nil
	}
	if legacy == "dedicated" || v2 == "dedicated" {
		return ArchitectureDedicated, nil
	}
	// The canonical classes above cover the known routes. Preserve the
	// all-cluster-types API compatibility of the original skill by treating an
	// unrecognized non-Dedicated, non-Essential value as the Dataflow-NG family.
	return ArchitectureNextGen, nil
}

func (s *Service) List(
	ctx context.Context,
	orgID, projectID, clusterID uint64,
	architecture Architecture,
) (*ListResponse, error) {
	if s == nil || s.clients == nil {
		return nil, ErrUnavailable
	}
	ctx, cancel := context.WithTimeout(ctx, listTimeout)
	defer cancel()
	var err error
	if architecture != ArchitectureEssential {
		ctx, err = withDataflowUser(ctx, orgID, projectID)
		if err != nil {
			return nil, fmt.Errorf("%w: %v", ErrUpstream, err)
		}
	}

	var response *ListResponse
	switch architecture {
	case ArchitectureDedicated:
		response, err = s.listDedicated(ctx, clusterID)
	case ArchitectureNextGen:
		response, err = s.listNextGen(ctx, clusterID)
	case ArchitectureEssential:
		response, err = s.listEssential(ctx, clusterID)
	default:
		return nil, fmt.Errorf("%w: unsupported TiCDC architecture", ErrInvalidArgument)
	}
	if err != nil {
		return nil, err
	}
	sort.SliceStable(response.Changefeeds, func(i, j int) bool {
		return response.Changefeeds[i].CreatedAt > response.Changefeeds[j].CreatedAt
	})
	return response, nil
}

func (s *Service) Detail(
	ctx context.Context,
	orgID, projectID, clusterID uint64,
	changefeedID string,
	architecture Architecture,
) (*DetailResponse, error) {
	if s == nil || s.clients == nil {
		return nil, ErrUnavailable
	}
	ctx, cancel := context.WithTimeout(ctx, detailTimeout)
	defer cancel()
	var err error
	if architecture != ArchitectureEssential {
		ctx, err = withDataflowUser(ctx, orgID, projectID)
		if err != nil {
			return nil, fmt.Errorf("%w: %v", ErrUpstream, err)
		}
	}

	var changefeed *Changefeed
	switch architecture {
	case ArchitectureDedicated:
		changefeedIDUint, parseErr := parseKernelChangefeedID(changefeedID)
		if parseErr != nil {
			return nil, parseErr
		}
		changefeed, err = s.getDedicated(ctx, clusterID, changefeedIDUint)
	case ArchitectureNextGen:
		changefeedIDUint, parseErr := parseKernelChangefeedID(changefeedID)
		if parseErr != nil {
			return nil, parseErr
		}
		changefeed, err = s.getNextGen(ctx, clusterID, changefeedIDUint)
	case ArchitectureEssential:
		changefeed, err = s.getEssential(ctx, clusterID, changefeedID)
	default:
		return nil, fmt.Errorf("%w: unsupported TiCDC architecture", ErrInvalidArgument)
	}
	if err != nil {
		return nil, err
	}
	return &DetailResponse{Changefeed: changefeed}, nil
}

func (s *Service) listEssential(ctx context.Context, clusterID uint64) (*ListResponse, error) {
	if s.clients.essential == nil {
		return nil, ErrUnavailable
	}
	startedAt := time.Now()
	logServerlessUpstreamStarted(ctx, "essential_list_changefeeds", clusterID, "")
	response, err := s.clients.essential.ListClusterChangefeeds(ctx, &serverlessrpc.ListClusterChangefeedsRequest{
		ClusterId: strconv.FormatUint(clusterID, 10),
		Limit:     changefeedListPageSize,
		Offset:    0,
	})
	logServerlessUpstreamFinished(ctx, "essential_list_changefeeds", clusterID, "", startedAt, err, essentialChangefeedCount(response))
	if err != nil {
		return nil, err
	}
	result := &ListResponse{Changefeeds: make([]*Changefeed, 0)}
	if response == nil {
		return result, nil
	}
	clusterIDString := strconv.FormatUint(clusterID, 10)
	for _, item := range response.GetItems() {
		if item == nil {
			continue
		}
		if item.GetClusterId() != clusterIDString || strings.TrimSpace(item.GetChangefeedId()) == "" {
			return nil, fmt.Errorf("%w: Essential list returned an invalid source cluster", ErrUpstream)
		}
		result.Changefeeds = append(result.Changefeeds, essentialChangefeed(item))
	}
	return result, nil
}

func (s *Service) getEssential(ctx context.Context, clusterID uint64, changefeedID string) (*Changefeed, error) {
	if s.clients.essential == nil {
		return nil, ErrUnavailable
	}
	changefeedID = strings.TrimSpace(changefeedID)
	if changefeedID == "" {
		return nil, ErrInvalidArgument
	}
	startedAt := time.Now()
	logServerlessUpstreamStarted(ctx, "essential_get_changefeed", clusterID, changefeedID)
	changefeed, err := s.clients.essential.GetChangefeed(ctx, &serverlessrpc.GetChangefeedRequest{
		ClusterId:    strconv.FormatUint(clusterID, 10),
		ChangefeedId: changefeedID,
	})
	logServerlessUpstreamFinished(ctx, "essential_get_changefeed", clusterID, changefeedID, startedAt, err, boolCount(changefeed != nil))
	if err != nil {
		return nil, err
	}
	if changefeed == nil || changefeed.GetChangefeedId() != changefeedID {
		return nil, ErrNotFound
	}
	if changefeed.GetClusterId() != strconv.FormatUint(clusterID, 10) {
		return nil, ErrNotFound
	}
	return essentialChangefeed(changefeed), nil
}

func parseKernelChangefeedID(value string) (uint64, error) {
	changefeedID, err := strconv.ParseUint(strings.TrimSpace(value), 10, 64)
	if err != nil || changefeedID == 0 {
		return 0, ErrInvalidArgument
	}
	return changefeedID, nil
}

func (s *Service) listDedicated(ctx context.Context, clusterID uint64) (*ListResponse, error) {
	if s.clients.dedicatedBusiness == nil || s.clients.dedicatedInternal == nil {
		return nil, ErrUnavailable
	}
	type internalResult struct {
		response *dataflow.ListInternalChangefeedsResponse
		err      error
	}
	internalCh := make(chan internalResult, 1)
	go func() {
		startedAt := time.Now()
		logUpstreamStarted(ctx, ArchitectureDedicated, "internal_list_changefeeds", clusterID, 0, true)
		response, err := s.clients.dedicatedInternal.ListInternalChangefeeds(ctx,
			&dataflow.ListInternalChangefeedsRequest{TidbClusterId: clusterID})
		logUpstreamFinished(ctx, ArchitectureDedicated, "internal_list_changefeeds", clusterID, 0, startedAt, err, internalChangefeedCount(response))
		internalCh <- internalResult{response: response, err: err}
	}()

	startedAt := time.Now()
	logUpstreamStarted(ctx, ArchitectureDedicated, "business_list_connectors", clusterID, 0, false)
	business, err := s.clients.dedicatedBusiness.ListConnectors(ctx, &dataflow.ConnectorListReq{
		ClusterId: clusterID, Page: 1, PerPage: changefeedListPageSize, ConnectorType: dataflow.ConnectorType_TOTAL,
	})
	logUpstreamFinished(ctx, ArchitectureDedicated, "business_list_connectors", clusterID, 0, startedAt, err, connectorCount(business))
	if err != nil {
		return nil, err
	}
	result := &ListResponse{Changefeeds: make([]*Changefeed, 0)}
	if business == nil || len(business.GetItems()) == 0 {
		return result, nil
	}

	internal := <-internalCh
	if internal.err != nil {
		log.WithContextL(ctx).Warn("TiCDC internal list enrichment failed",
			zap.String("architecture", string(ArchitectureDedicated)), zap.String("grpc_code", grpcStatusCode(internal.err)))
	}
	byDisplayID := indexDedicatedInternal(internal.response)
	for _, item := range business.GetItems() {
		if item == nil {
			continue
		}
		if item.GetRequest() == nil || item.GetRequest().GetClusterId() != clusterID {
			return nil, fmt.Errorf("%w: Dedicated list returned an invalid source cluster", ErrUpstream)
		}
		changefeed, convertErr := fullDedicatedChangefeed(item)
		if convertErr != nil {
			return nil, fmt.Errorf("%w: convert Dedicated changefeed: %v", ErrUpstream, convertErr)
		}
		enrichDedicated(changefeed, byDisplayID[item.GetConnectorId()])
		result.Changefeeds = append(result.Changefeeds, changefeed)
	}
	return result, nil
}

func (s *Service) listNextGen(ctx context.Context, clusterID uint64) (*ListResponse, error) {
	if s.clients.nextGenBusiness == nil || s.clients.nextGenInternal == nil {
		return nil, ErrUnavailable
	}
	type internalResult struct {
		response *dataflowng.ListInternalChangefeedsResponse
		err      error
	}
	internalCh := make(chan internalResult, 1)
	go func() {
		startedAt := time.Now()
		logUpstreamStarted(ctx, ArchitectureNextGen, "internal_list_changefeeds", clusterID, 0, true)
		response, err := s.clients.nextGenInternal.ListInternalChangefeeds(ctx,
			&dataflowng.ListInternalChangefeedsRequest{TidbClusterId: clusterID})
		logUpstreamFinished(ctx, ArchitectureNextGen, "internal_list_changefeeds", clusterID, 0, startedAt, err, internalChangefeedNGCount(response))
		internalCh <- internalResult{response: response, err: err}
	}()

	startedAt := time.Now()
	logUpstreamStarted(ctx, ArchitectureNextGen, "business_list_changefeeds", clusterID, 0, false)
	business, err := s.clients.nextGenBusiness.ListChangefeeds(ctx, &dataflowng.ChangefeedListReq{
		ClusterId: clusterID, Page: 1, PerPage: changefeedListPageSize, ChangefeedType: dataflowng.ChangefeedType_TOTAL,
	})
	logUpstreamFinished(ctx, ArchitectureNextGen, "business_list_changefeeds", clusterID, 0, startedAt, err, changefeedNGCount(business))
	if err != nil {
		return nil, err
	}
	result := &ListResponse{Changefeeds: make([]*Changefeed, 0)}
	if business == nil || len(business.GetItems()) == 0 {
		return result, nil
	}

	internal := <-internalCh
	if internal.err != nil {
		log.WithContextL(ctx).Warn("TiCDC internal list enrichment failed",
			zap.String("architecture", string(ArchitectureNextGen)), zap.String("grpc_code", grpcStatusCode(internal.err)))
	}
	byDisplayID := indexNextGenInternal(internal.response)
	for _, item := range business.GetItems() {
		if item == nil {
			continue
		}
		if item.GetRequest() == nil || item.GetRequest().GetClusterId() != clusterID {
			return nil, fmt.Errorf("%w: next-gen list returned an invalid source cluster", ErrUpstream)
		}
		changefeed, convertErr := fullNextGenChangefeed(item)
		if convertErr != nil {
			return nil, fmt.Errorf("%w: convert next-gen changefeed: %v", ErrUpstream, convertErr)
		}
		enrichNextGen(changefeed, byDisplayID[item.GetChangefeedId()])
		result.Changefeeds = append(result.Changefeeds, changefeed)
	}
	return result, nil
}

func (s *Service) getDedicated(ctx context.Context, clusterID, changefeedID uint64) (*Changefeed, error) {
	if s.clients.dedicatedBusiness == nil || s.clients.dedicatedInternal == nil {
		return nil, ErrUnavailable
	}
	startedAt := time.Now()
	logUpstreamStarted(ctx, ArchitectureDedicated, "internal_get_changefeed", clusterID, changefeedID, true)
	internal, err := s.clients.dedicatedInternal.GetInternalChangefeed(ctx,
		&dataflow.GetInternalChangefeedRequest{ChangefeedId: changefeedID})
	logUpstreamFinished(ctx, ArchitectureDedicated, "internal_get_changefeed", clusterID, changefeedID, startedAt, err, boolCount(internal != nil))
	if err != nil {
		return nil, err
	}
	if internal == nil || internal.GetId() != changefeedID || internal.GetDisplayId() == 0 {
		return nil, ErrNotFound
	}
	startedAt = time.Now()
	logUpstreamStarted(ctx, ArchitectureDedicated, "business_get_connector", clusterID, internal.GetDisplayId(), false)
	detail, err := s.clients.dedicatedBusiness.GetConnector(ctx,
		&dataflow.ConnectorID{ConnectorId: internal.GetDisplayId()})
	logUpstreamFinished(ctx, ArchitectureDedicated, "business_get_connector", clusterID, internal.GetDisplayId(), startedAt, err, boolCount(detail != nil))
	if err != nil {
		return nil, err
	}
	if detail == nil || detail.GetConnectorId() != internal.GetDisplayId() || detail.GetRequest() == nil {
		return nil, fmt.Errorf("%w: incomplete Dedicated changefeed identity", ErrUpstream)
	}
	if detail.GetRequest().GetClusterId() != clusterID {
		return nil, ErrNotFound
	}
	result, err := fullDedicatedChangefeed(detail)
	if err != nil {
		return nil, fmt.Errorf("%w: convert Dedicated changefeed: %v", ErrUpstream, err)
	}
	enrichDedicated(result, internal)
	return result, nil
}

func (s *Service) getNextGen(ctx context.Context, clusterID, changefeedID uint64) (*Changefeed, error) {
	if s.clients.nextGenBusiness == nil || s.clients.nextGenInternal == nil {
		return nil, ErrUnavailable
	}
	startedAt := time.Now()
	logUpstreamStarted(ctx, ArchitectureNextGen, "internal_get_changefeed", clusterID, changefeedID, true)
	internal, err := s.clients.nextGenInternal.GetInternalChangefeed(ctx,
		&dataflowng.GetInternalChangefeedRequest{ChangefeedId: changefeedID})
	logUpstreamFinished(ctx, ArchitectureNextGen, "internal_get_changefeed", clusterID, changefeedID, startedAt, err, boolCount(internal != nil))
	if err != nil {
		return nil, err
	}
	if internal == nil || internal.GetId() != changefeedID || internal.GetDisplayId() == 0 {
		return nil, ErrNotFound
	}
	startedAt = time.Now()
	logUpstreamStarted(ctx, ArchitectureNextGen, "business_get_changefeed", clusterID, internal.GetDisplayId(), false)
	detail, err := s.clients.nextGenBusiness.GetChangefeed(ctx,
		&dataflowng.ChangefeedID{ChangefeedId: internal.GetDisplayId()})
	logUpstreamFinished(ctx, ArchitectureNextGen, "business_get_changefeed", clusterID, internal.GetDisplayId(), startedAt, err, boolCount(detail != nil))
	if err != nil {
		return nil, err
	}
	if detail == nil || detail.GetChangefeedId() != internal.GetDisplayId() || detail.GetRequest() == nil {
		return nil, fmt.Errorf("%w: incomplete next-gen changefeed identity", ErrUpstream)
	}
	if detail.GetRequest().GetClusterId() != clusterID {
		return nil, ErrNotFound
	}
	result, err := fullNextGenChangefeed(detail)
	if err != nil {
		return nil, fmt.Errorf("%w: convert next-gen changefeed: %v", ErrUpstream, err)
	}
	enrichNextGen(result, internal)
	return result, nil
}

func indexDedicatedInternal(response *dataflow.ListInternalChangefeedsResponse) map[uint64]*dataflow.InternalChangefeed {
	result := make(map[uint64]*dataflow.InternalChangefeed)
	if response == nil {
		return result
	}
	for _, item := range response.GetChangefeeds() {
		if item != nil && item.GetDisplayId() != 0 {
			result[item.GetDisplayId()] = item
		}
	}
	return result
}

func indexNextGenInternal(response *dataflowng.ListInternalChangefeedsResponse) map[uint64]*dataflowng.InternalChangefeed {
	result := make(map[uint64]*dataflowng.InternalChangefeed)
	if response == nil {
		return result
	}
	for _, item := range response.GetChangefeeds() {
		if item != nil && item.GetDisplayId() != 0 {
			result[item.GetDisplayId()] = item
		}
	}
	return result
}

func ParseProjectID(value string) (uint64, error) {
	value = strings.TrimSpace(value)
	if value == "" {
		return 0, nil
	}
	projectID, err := strconv.ParseUint(value, 10, 64)
	if err != nil || projectID == 0 {
		return 0, fmt.Errorf("%w: invalid canonical project ID", ErrUpstream)
	}
	return projectID, nil
}

func grpcStatusCode(err error) string {
	if err == nil {
		return "ok"
	}
	return status.Code(err).String()
}

func logUpstreamStarted(ctx context.Context, architecture Architecture, operation string, clusterID, changefeedID uint64, internal bool) {
	log.WithContextL(ctx).Info("TiCDC upstream gRPC request started",
		zap.String("architecture", string(architecture)),
		zap.String("upstream_operation", operation),
		zap.Bool("upstream_internal", internal),
		zap.Uint64("cluster_id", clusterID),
		zap.Uint64("changefeed_id", changefeedID),
		zap.Int("page", 1),
		zap.Int("per_page", changefeedListPageSize),
	)
}

func logUpstreamFinished(ctx context.Context, architecture Architecture, operation string, clusterID, changefeedID uint64, startedAt time.Time, err error, resultCount int) {
	fields := []zap.Field{
		zap.String("architecture", string(architecture)),
		zap.String("upstream_operation", operation),
		zap.Uint64("cluster_id", clusterID),
		zap.Uint64("changefeed_id", changefeedID),
		zap.Duration("duration", time.Since(startedAt)),
		zap.String("grpc_code", grpcStatusCode(err)),
		zap.Int("result_count", resultCount),
	}
	if err != nil {
		log.WithContextL(ctx).Warn("TiCDC upstream gRPC request failed", append(fields,
			zap.String("error_class", strings.ToLower(status.Code(err).String())),
		)...)
		return
	}
	log.WithContextL(ctx).Info("TiCDC upstream gRPC response received", fields...)
}

func logServerlessUpstreamStarted(ctx context.Context, operation string, clusterID uint64, changefeedID string) {
	log.WithContextL(ctx).Info("TiCDC Serverless upstream gRPC request started",
		zap.String("architecture", string(ArchitectureEssential)),
		zap.String("upstream_operation", operation),
		zap.Uint64("cluster_id", clusterID),
		zap.String("changefeed_id", changefeedID),
		zap.Int("limit", changefeedListPageSize),
		zap.Int("offset", 0),
		zap.String("authentication_method", "not_applicable"),
	)
}

func logServerlessUpstreamFinished(ctx context.Context, operation string, clusterID uint64, changefeedID string, startedAt time.Time, err error, resultCount int) {
	fields := []zap.Field{
		zap.String("architecture", string(ArchitectureEssential)),
		zap.String("upstream_operation", operation),
		zap.Uint64("cluster_id", clusterID),
		zap.String("changefeed_id", changefeedID),
		zap.Duration("duration", time.Since(startedAt)),
		zap.String("grpc_code", grpcStatusCode(err)),
		zap.Int("result_count", resultCount),
	}
	if err != nil {
		log.WithContextL(ctx).Warn("TiCDC Serverless upstream gRPC request failed", append(fields,
			zap.String("error_class", strings.ToLower(status.Code(err).String())),
		)...)
		return
	}
	log.WithContextL(ctx).Info("TiCDC Serverless upstream gRPC response received", fields...)
}

func connectorCount(response *dataflow.Connectors) int {
	if response == nil {
		return 0
	}
	return len(response.GetItems())
}

func internalChangefeedCount(response *dataflow.ListInternalChangefeedsResponse) int {
	if response == nil {
		return 0
	}
	return len(response.GetChangefeeds())
}

func changefeedNGCount(response *dataflowng.Changefeeds) int {
	if response == nil {
		return 0
	}
	return len(response.GetItems())
}

func internalChangefeedNGCount(response *dataflowng.ListInternalChangefeedsResponse) int {
	if response == nil {
		return 0
	}
	return len(response.GetChangefeeds())
}

func essentialChangefeedCount(response *serverlessrpc.ListClusterChangefeedsResponse) int {
	if response == nil {
		return 0
	}
	return len(response.GetItems())
}

func boolCount(value bool) int {
	if value {
		return 1
	}
	return 0
}