package cloudconfig

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

	ccproto "github.com/tidbcloud/config-center/proto/config_center"
	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	ccapi "github.com/tidbcloud/devops/api/cloudconfig"
	ccent "github.com/tidbcloud/devops/entity/cloudconfig"
	"github.com/tidbcloud/devops/pkg/infra/log"
	ccsvcapi "github.com/tidbcloud/devops/pkg/service/cloudconfig/api"
	"github.com/tidbcloud/devops/pkg/service/cloudconfig/cvt"
	serverlessrpc "github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
)

const (
	unifiedChangefeedListPageSize = 1000
	changefeedListRPCTimeout      = 5 * time.Second
)

// ListInternalChangefeeds returns full, non-sensitive business changefeed
// details from the Dataflow service selected by the source cluster type.
// Internal lists are best-effort enrichment sources for kernel IDs and TiCDC
// cluster metadata; they never decide whether a changefeed is included.
func (s *svc) ListInternalChangefeeds(
	ctx context.Context,
	param ccsvcapi.ListInternalChangefeedsParam,
) (*ccapi.ListInternalChangefeedsResp, error) {
	rpcCtx, cancel := context.WithTimeout(ctx, changefeedListRPCTimeout)
	defer cancel()

	switch param.ClusterType {
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_DEDICATED_TIDB)]:
		return s.listDedicatedInternalChangefeeds(rpcCtx, param.TiDBClusterID)
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_LOGICAL_CLUSTER)],
		ccproto.ClusterType_name[int32(ccproto.ClusterType_SHARED_POOL)]:
		return s.listNextGenInternalChangefeeds(rpcCtx, param.TiDBClusterID)
	case string(ccent.ClusterTypeChangefeedEssential):
		return s.listEssentialInternalChangefeeds(rpcCtx, param.TiDBClusterID)
	default:
		return nil, fmt.Errorf("unsupported cluster_type: %s", param.ClusterType)
	}
}

func (s *svc) listEssentialInternalChangefeeds(
	ctx context.Context,
	clusterID uint64,
) (*ccapi.ListInternalChangefeedsResp, error) {
	changefeeds, err := s.serverlessCli.ListClusterChangefeeds(
		ctx,
		strconv.FormatUint(clusterID, 10),
		unifiedChangefeedListPageSize,
		0,
	)
	if err != nil {
		return nil, fmt.Errorf("list essential changefeeds: %w", err)
	}

	resp := &ccapi.ListInternalChangefeedsResp{Changefeeds: make([]*ccapi.InternalChangefeed, 0)}
	if changefeeds == nil {
		return resp, nil
	}
	for _, changefeed := range changefeeds.GetItems() {
		if changefeed != nil {
			resp.Changefeeds = append(resp.Changefeeds, buildEssentialChangefeed(changefeed))
		}
	}
	sortChangefeedsByCreatedAt(resp)
	return resp, nil
}

func (s *svc) listDedicatedInternalChangefeeds(
	ctx context.Context,
	clusterID uint64,
) (*ccapi.ListInternalChangefeedsResp, error) {
	type internalResult struct {
		resp *dataflow.ListInternalChangefeedsResponse
		err  error
	}
	internalCh := make(chan internalResult, 1)
	go func() {
		resp, err := s.listInternalChangefeeds(ctx, &dataflow.ListInternalChangefeedsRequest{
			TidbClusterId: clusterID,
		})
		internalCh <- internalResult{resp: resp, err: err}
	}()

	business, err := s.listChangefeeds(ctx, &dataflow.ConnectorListReq{
		ClusterId:     clusterID,
		Page:          1,
		PerPage:       unifiedChangefeedListPageSize,
		ConnectorType: dataflow.ConnectorType_TOTAL,
	})
	if err != nil {
		return nil, fmt.Errorf("list dedicated business changefeeds: %w", err)
	}

	resp := &ccapi.ListInternalChangefeedsResp{Changefeeds: make([]*ccapi.InternalChangefeed, 0)}
	if business == nil || len(business.GetItems()) == 0 {
		return resp, nil
	}

	internal := <-internalCh
	if internal.err != nil {
		log.FromContext(ctx).Error(internal.err, "list dedicated internal changefeeds for enrichment",
			"cluster_id", clusterID)
	}
	internalByDisplayID := indexInternalChangefeedsByDisplayID(internal.resp)
	for _, connector := range business.GetItems() {
		if connector == nil {
			continue
		}
		changefeed, err := cvt.From.Dataflow.FullChangefeed(connector)
		if err != nil {
			return nil, fmt.Errorf("build dedicated changefeed %d: %w", connector.GetConnectorId(), err)
		}
		cvt.From.Dataflow.EnrichChangefeed(changefeed, internalByDisplayID[connector.GetConnectorId()])
		resp.Changefeeds = append(resp.Changefeeds, changefeed)
	}

	sortChangefeedsByCreatedAt(resp)
	return resp, nil
}

func (s *svc) listNextGenInternalChangefeeds(
	ctx context.Context,
	clusterID uint64,
) (*ccapi.ListInternalChangefeedsResp, error) {
	type internalResult struct {
		resp *dataflowng.ListInternalChangefeedsResponse
		err  error
	}
	internalCh := make(chan internalResult, 1)
	go func() {
		resp, err := s.listInternalNgChangefeeds(ctx, &dataflowng.ListInternalChangefeedsRequest{
			TidbClusterId: clusterID,
		})
		internalCh <- internalResult{resp: resp, err: err}
	}()

	business, err := s.listNgChangefeeds(ctx, &dataflowng.ChangefeedListReq{
		ClusterId:      clusterID,
		Page:           1,
		PerPage:        unifiedChangefeedListPageSize,
		ChangefeedType: dataflowng.ChangefeedType_TOTAL,
	})
	if err != nil {
		return nil, fmt.Errorf("list next-gen business changefeeds: %w", err)
	}

	resp := &ccapi.ListInternalChangefeedsResp{Changefeeds: make([]*ccapi.InternalChangefeed, 0)}
	if business == nil || len(business.GetItems()) == 0 {
		return resp, nil
	}

	internal := <-internalCh
	if internal.err != nil {
		log.FromContext(ctx).Error(internal.err, "list next-gen internal changefeeds for enrichment",
			"cluster_id", clusterID)
	}
	internalByDisplayID := indexInternalNgChangefeedsByDisplayID(internal.resp)
	for _, item := range business.GetItems() {
		if item == nil {
			continue
		}
		changefeed, err := cvt.From.DataflowNg.FullChangefeed(item)
		if err != nil {
			return nil, fmt.Errorf("build next-gen changefeed %d: %w", item.GetChangefeedId(), err)
		}
		cvt.From.DataflowNg.EnrichChangefeed(changefeed, internalByDisplayID[item.GetChangefeedId()])
		resp.Changefeeds = append(resp.Changefeeds, changefeed)
	}

	sortChangefeedsByCreatedAt(resp)
	return resp, nil
}

func sortChangefeedsByCreatedAt(resp *ccapi.ListInternalChangefeedsResp) {
	sort.SliceStable(resp.Changefeeds, func(i, j int) bool {
		return resp.Changefeeds[i].CreatedAt > resp.Changefeeds[j].CreatedAt
	})
}

func buildEssentialChangefeed(changefeed *serverlessrpc.ChangefeedInfo) *ccapi.InternalChangefeed {
	result := &ccapi.InternalChangefeed{
		Id:               changefeed.GetChangefeedId(),
		DisplayId:        changefeed.GetChangefeedId(),
		ClusterType:      string(ccent.ClusterTypeChangefeedEssential),
		Arch:             changefeed.GetArch(),
		DisplayName:      changefeed.GetDisplayName(),
		DownstreamType:   changefeed.GetSinkType(),
		Status:           changefeed.GetState(),
		CreatedAt:        parseEssentialChangefeedTime(changefeed.GetCreatedAt()),
		TiCDCClusterName: changefeed.GetTicdcClusterName(),
		CheckpointTSO:    strconv.FormatUint(changefeed.GetCheckpointTso(), 10),
		CheckpointTS:     parseEssentialChangefeedTime(changefeed.GetCheckpointTime()),
		RCU:              strconv.FormatUint(uint64(changefeed.GetRcu()), 10),
	}
	if changefeed.GetStartTso() != 0 {
		result.StartPosition = &ccapi.ChangefeedStartPosition{
			Mode: "tso",
			TSO:  strconv.FormatUint(changefeed.GetStartTso(), 10),
		}
	}
	return result
}

func parseEssentialChangefeedTime(value string) int64 {
	if value == "" {
		return 0
	}
	for _, layout := range []string{time.RFC3339Nano, "2006-01-02 15:04:05"} {
		if parsed, err := time.Parse(layout, value); err == nil {
			return parsed.Unix()
		}
	}
	return 0
}

// GetInternalChangefeed resolves a kernel changefeed ID to its business detail
// and returns the source TiDB cluster ID required for scope authorization.
func (s *svc) GetInternalChangefeed(
	ctx context.Context,
	param ccsvcapi.GetInternalChangefeedParam,
) (*ccsvcapi.GetInternalChangefeedResult, error) {
	switch param.ClusterType {
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED)]:
		return s.getDedicatedChangefeedDetail(ctx, param.ChangefeedID)
	case ccproto.ClusterType_name[int32(ccproto.ClusterType_CHANGEFEED_NG)]:
		return s.getNextGenChangefeedDetail(ctx, param.ChangefeedID)
	case string(ccent.ClusterTypeChangefeedEssential):
		return s.getEssentialChangefeedDetail(ctx, param.ChangefeedIDString)
	default:
		return nil, fmt.Errorf("unsupported cluster_type: %s", param.ClusterType)
	}
}

func (s *svc) getEssentialChangefeedDetail(
	ctx context.Context,
	changefeedID string,
) (*ccsvcapi.GetInternalChangefeedResult, error) {
	if changefeedID == "" {
		return nil, fmt.Errorf("essential changefeed ID is required")
	}
	changefeed, err := s.serverlessCli.GetChangefeed(ctx, changefeedID)
	if err != nil {
		return nil, err
	}
	if changefeed == nil || changefeed.GetChangefeedId() != changefeedID || changefeed.GetClusterId() == "" {
		return nil, fmt.Errorf("essential changefeed %s has incomplete source metadata", changefeedID)
	}
	clusterID, err := strconv.ParseUint(changefeed.GetClusterId(), 10, 64)
	if err != nil {
		return nil, fmt.Errorf("parse essential source cluster ID %q: %w", changefeed.GetClusterId(), err)
	}
	return &ccsvcapi.GetInternalChangefeedResult{
		Changefeed:    buildEssentialChangefeed(changefeed),
		TiDBClusterID: clusterID,
	}, nil
}

func (s *svc) getDedicatedChangefeedDetail(
	ctx context.Context,
	changefeedID uint64,
) (*ccsvcapi.GetInternalChangefeedResult, error) {
	internal, err := s.getInternalChangefeed(ctx, &dataflow.GetInternalChangefeedRequest{
		ChangefeedId: changefeedID,
	})
	if err != nil {
		return nil, err
	}
	if internal == nil || internal.GetId() != changefeedID || internal.GetDisplayId() == 0 {
		return nil, fmt.Errorf("internal changefeed %d has incomplete identity metadata", changefeedID)
	}

	detail, err := s.getChangefeed(ctx, &dataflow.ConnectorID{
		ConnectorId: internal.GetDisplayId(),
	})
	if err != nil {
		return nil, err
	}
	if detail == nil || detail.GetConnectorId() != internal.GetDisplayId() || detail.GetRequest() == nil ||
		detail.GetRequest().GetClusterId() == 0 {
		return nil, fmt.Errorf("changefeed %d has incomplete source metadata", internal.GetDisplayId())
	}

	changefeed, err := cvt.From.Dataflow.FullChangefeed(detail)
	if err != nil {
		return nil, fmt.Errorf("build changefeed %d: %w", internal.GetDisplayId(), err)
	}
	cvt.From.Dataflow.EnrichChangefeed(changefeed, internal)
	return &ccsvcapi.GetInternalChangefeedResult{
		Changefeed:    changefeed,
		TiDBClusterID: detail.GetRequest().GetClusterId(),
	}, nil
}

func (s *svc) getNextGenChangefeedDetail(
	ctx context.Context,
	changefeedID uint64,
) (*ccsvcapi.GetInternalChangefeedResult, error) {
	internal, err := s.getInternalNgChangefeed(ctx, &dataflowng.GetInternalChangefeedRequest{
		ChangefeedId: changefeedID,
	})
	if err != nil {
		return nil, err
	}
	if internal == nil || internal.GetId() != changefeedID || internal.GetDisplayId() == 0 {
		return nil, fmt.Errorf("internal next-gen changefeed %d has incomplete identity metadata", changefeedID)
	}

	detail, err := s.getNgChangefeed(ctx, &dataflowng.ChangefeedID{
		ChangefeedId: internal.GetDisplayId(),
	})
	if err != nil {
		return nil, err
	}
	if detail == nil || detail.GetChangefeedId() != internal.GetDisplayId() || detail.GetRequest() == nil ||
		detail.GetRequest().GetClusterId() == 0 {
		return nil, fmt.Errorf("changefeed %d has incomplete source metadata", internal.GetDisplayId())
	}

	changefeed, err := cvt.From.DataflowNg.FullChangefeed(detail)
	if err != nil {
		return nil, fmt.Errorf("build next-gen changefeed %d: %w", internal.GetDisplayId(), err)
	}
	cvt.From.DataflowNg.EnrichChangefeed(changefeed, internal)
	return &ccsvcapi.GetInternalChangefeedResult{
		Changefeed:    changefeed,
		TiDBClusterID: detail.GetRequest().GetClusterId(),
	}, nil
}

func (s *svc) listChangefeeds(ctx context.Context, param *dataflow.ConnectorListReq) (*dataflow.Connectors, error) {
	return s.dataflow.Connector.ListConnectors(ctx, param)
}

func (s *svc) getChangefeed(ctx context.Context, param *dataflow.ConnectorID) (*dataflow.Connector, error) {
	return s.dataflow.Connector.GetConnector(ctx, param)
}

func (s *svc) listInternalChangefeeds(ctx context.Context, param *dataflow.ListInternalChangefeedsRequest) (*dataflow.ListInternalChangefeedsResponse, error) {
	return s.dataflow.Ops.ListInternalChangefeeds(ctx, param)
}

func (s *svc) getInternalChangefeed(ctx context.Context, param *dataflow.GetInternalChangefeedRequest) (*dataflow.InternalChangefeed, error) {
	return s.dataflow.Ops.GetInternalChangefeed(ctx, param)
}

func (s *svc) RestartChangefeed(ctx context.Context, param *dataflow.RestartChangefeedRequest) error {
	_, err := s.dataflow.Ops.RestartChangefeed(ctx, param)
	return err
}

func (s *svc) GetInternalTiCDCCluster(ctx context.Context, param *dataflow.GetInternalTiCDCClusterRequest) (*dataflow.InternalTiCDCCluster, error) {
	return s.dataflow.Ops.GetInternalTiCDCCluster(ctx, param)
}
func (s *svc) RestartTiCDCCluster(ctx context.Context, param *dataflow.RestartTiCDCClusterRequest) error {
	_, err := s.dataflow.Ops.RestartTiCDCCluster(ctx, param)
	return err
}

func (s *svc) listNgChangefeeds(ctx context.Context, param *dataflowng.ChangefeedListReq) (*dataflowng.Changefeeds, error) {
	return s.dataflow.NgChangefeed.ListChangefeeds(ctx, param)
}

func (s *svc) getNgChangefeed(ctx context.Context, param *dataflowng.ChangefeedID) (*dataflowng.Changefeed, error) {
	return s.dataflow.NgChangefeed.GetChangefeed(ctx, param)
}

func (s *svc) listInternalNgChangefeeds(ctx context.Context, param *dataflowng.ListInternalChangefeedsRequest) (*dataflowng.ListInternalChangefeedsResponse, error) {
	return s.dataflow.NgOps.ListInternalChangefeeds(ctx, param)
}

func (s *svc) getInternalNgChangefeed(ctx context.Context, param *dataflowng.GetInternalChangefeedRequest) (*dataflowng.InternalChangefeed, error) {
	return s.dataflow.NgOps.GetInternalChangefeed(ctx, param)
}

func (s *svc) RestartNgChangefeed(ctx context.Context, param *dataflowng.RestartChangefeedRequest) error {
	_, err := s.dataflow.NgOps.RestartChangefeed(ctx, param)
	return err
}

func (s *svc) GetInternalNgTiCDCCluster(ctx context.Context, param *dataflowng.GetInternalTiCDCClusterRequest) (*dataflowng.InternalTiCDCCluster, error) {
	return s.dataflow.NgOps.GetInternalTiCDCCluster(ctx, param)
}

func (s *svc) RestartNgTiCDCCluster(ctx context.Context, param *dataflowng.RestartTiCDCClusterRequest) error {
	_, err := s.dataflow.NgOps.RestartTiCDCCluster(ctx, param)
	return err
}

func indexInternalChangefeedsByDisplayID(
	resp *dataflow.ListInternalChangefeedsResponse,
) map[uint64]*dataflow.InternalChangefeed {
	result := make(map[uint64]*dataflow.InternalChangefeed)
	if resp == nil {
		return result
	}
	for _, changefeed := range resp.GetChangefeeds() {
		if changefeed == nil || changefeed.GetDisplayId() == 0 {
			continue
		}
		result[changefeed.GetDisplayId()] = changefeed
	}
	return result
}

func indexInternalNgChangefeedsByDisplayID(
	resp *dataflowng.ListInternalChangefeedsResponse,
) map[uint64]*dataflowng.InternalChangefeed {
	result := make(map[uint64]*dataflowng.InternalChangefeed)
	if resp == nil {
		return result
	}
	for _, changefeed := range resp.GetChangefeeds() {
		if changefeed == nil || changefeed.GetDisplayId() == 0 {
			continue
		}
		result[changefeed.GetDisplayId()] = changefeed
	}
	return result
}