package console

import (
	"context"

	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service-ng/internal/bizutils"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type SinkEndpointController interface {
	dataflow.SinkEndpointServiceServer
}

type sinkEndpointControllerImpl struct {
	dataflow.UnimplementedSinkEndpointServiceServer
	*BaseController
	sinkEndpointModel                    sink_endpoint.SinkEndpointModel
	endpointServiceAvailabilityZoneModel sink_endpoint.EndpointServiceAvailabilityZoneModel
	mgmtSvcCli                           grpcli.MgmtSvcGrpcClient
}

func NewSinkEndpointController(
	sinkEndpointModel sink_endpoint.SinkEndpointModel,
	endpointServiceAvailabilityZoneModel sink_endpoint.EndpointServiceAvailabilityZoneModel,
	mgmtSvcCli grpcli.MgmtSvcGrpcClient,
) SinkEndpointController {
	return &sinkEndpointControllerImpl{
		BaseController:                       NewBaseController(nil),
		sinkEndpointModel:                    sinkEndpointModel,
		endpointServiceAvailabilityZoneModel: endpointServiceAvailabilityZoneModel,
		mgmtSvcCli:                           mgmtSvcCli,
	}
}

func (m *sinkEndpointControllerImpl) CreateSinkEndpoint(
	ctx context.Context, req *dataflow.CreateSinkEndpointRequest,
) (resp *dataflow.SinkEndpoint, rErr error) {
	if req == nil || req.SinkEndpoint == nil {
		return nil, bizerr.New(bizerr.ErrCodeParams, "sink endpoint is required")
	}
	var cluster *mgmt.Cluster
	defer func() {
		m.recordAuditLogForCreateSinkEndpoint(ctx, cluster, resp, rErr)
	}()

	cluster, bizErr := m.mgmtSvcCli.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		rErr = bizErr
		return nil, bizErr
	}

	modelReq := createSinkEndpointRequestDataflow2Model(req)
	user, userErr := bizutils.GetUserFromContext(ctx)
	if userErr != nil {
		rErr = bizerr.New(bizerr.ErrCodePermission, "user not in context")
		return nil, rErr
	}
	modelReq.UserID = user.ID
	modelReq.UserEmail = user.Email
	modelResp, err := m.sinkEndpointModel.CreateSinkEndpoint(ctx, modelReq)
	if err != nil {
		rErr = err
		return nil, err
	}
	resp = sinkEndpointEntity2Dataflow(modelResp.SinkEndpoint)
	return resp, nil
}

func (m *sinkEndpointControllerImpl) ListSinkEndpoints(
	ctx context.Context, req *dataflow.ListSinkEndpointsRequest,
) (*dataflow.ListSinkEndpointsResponse, error) {
	modelReq := &domain.ListSinkEndpointsReq{
		ClusterID: req.ClusterId,
		Status:    sinkEndpointStatusDataflow2Entity(req.Status),
	}
	modelResp, err := m.sinkEndpointModel.ListSinkEndpoint(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	sinkEndpoints := make([]*dataflow.SinkEndpoint, 0, len(modelResp.SinkEndpoints))
	for i := range modelResp.SinkEndpoints {
		sinkEndpoints = append(sinkEndpoints, sinkEndpointEntity2Dataflow(&modelResp.SinkEndpoints[i]))
	}
	return &dataflow.ListSinkEndpointsResponse{
		SinkEndpoints: sinkEndpoints,
	}, nil
}

func (m *sinkEndpointControllerImpl) GetSinkEndpoint(
	ctx context.Context, req *dataflow.GetSinkEndpointRequest,
) (*dataflow.SinkEndpoint, error) {
	modelReq := &domain.QuerySinkEndpointReq{
		ID:        req.SinkEndpointId,
		ClusterID: req.ClusterId,
	}
	modelResp, err := m.sinkEndpointModel.QuerySinkEndpoint(ctx, modelReq)
	if err != nil {
		return nil, err
	}

	sinkEndpointEnt := modelResp.SinkEndpoint
	return sinkEndpointEntity2Dataflow(sinkEndpointEnt), nil
}

func (m *sinkEndpointControllerImpl) DeleteSinkEndpoint(
	ctx context.Context, req *dataflow.DeleteSinkEndpointRequest,
) (_ *emptypb.Empty, rErr error) {
	var (
		cluster  *mgmt.Cluster
		endpoint *entity.BizSinkEndpoint
	)
	defer func() {
		m.recordAuditLogForDeleteSinkEndpoint(ctx, cluster, endpoint, rErr)
	}()

	cluster, bizErr := m.mgmtSvcCli.GetClusterByID(ctx, req.GetClusterId())
	if bizErr != nil {
		rErr = bizErr
		return nil, bizErr
	}
	user, userErr := bizutils.GetUserFromContext(ctx)
	if userErr != nil {
		rErr = bizerr.New(bizerr.ErrCodePermission, "user not in context")
		return nil, rErr
	}

	queryResp, err := m.sinkEndpointModel.QuerySinkEndpoint(ctx, &domain.QuerySinkEndpointReq{
		ID:        req.SinkEndpointId,
		ClusterID: req.ClusterId,
	})
	if err != nil {
		rErr = err
		return nil, err
	}
	endpoint = queryResp.SinkEndpoint

	deleteReq := &domain.DeleteSinkEndpointReq{
		ID:        req.SinkEndpointId,
		ClusterID: req.ClusterId,
		ProjectID: cluster.GetProjectId(),
		UserID:    user.ID,
		UserEmail: user.Email,
	}
	if err := m.sinkEndpointModel.DeleteSinkEndpoint(ctx, deleteReq); err != nil {
		rErr = err
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (m *sinkEndpointControllerImpl) ListSinkAvailabilityZones(
	ctx context.Context, req *dataflow.ListSinkAvailabilityZonesRequest,
) (*dataflow.ListSinkAvailabilityZonesResponse, error) {
	modelReq := &domain.GetAvailabilityZonesReq{
		ClusterID: req.ClusterId,
	}
	modelResp, err := m.sinkEndpointModel.GetAvailabilityZones(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	return &dataflow.ListSinkAvailabilityZonesResponse{
		AvailabilityZones: modelResp.Zones,
	}, nil
}

func (m *sinkEndpointControllerImpl) DescribeEndpointServiceAvailabilityZones(
	ctx context.Context, req *dataflow.DescribeEndpointServiceAvailabilityZonesRequest,
) (*dataflow.DescribeEndpointServiceAvailabilityZonesResponse, error) {
	modelReq := &domain.DescribeEndpointServiceAvailabilityZonesReq{
		ClusterID:     req.ClusterId,
		ServiceName:   req.EndpointServiceName,
		ServiceRegion: req.ServiceRegion,
	}
	modelResp, err := m.endpointServiceAvailabilityZoneModel.DescribeEndpointServiceAvailabilityZones(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	zones := make([]*dataflow.EndpointServiceAvailabilityZone, 0, len(modelResp.Zones))
	for _, zone := range modelResp.Zones {
		zones = append(zones, &dataflow.EndpointServiceAvailabilityZone{
			AvailabilityZone:   zone.AvailabilityZone,
			AvailabilityZoneId: zone.AvailabilityZoneID,
		})
	}
	return &dataflow.DescribeEndpointServiceAvailabilityZonesResponse{
		Zones: zones,
	}, nil
}

func (m *sinkEndpointControllerImpl) GenerateSinkAdvertisedListener(
	ctx context.Context, req *dataflow.GenerateSinkAdvertisedListenerRequest,
) (*dataflow.GenerateSinkAdvertisedListenerResponse, error) {
	modelReq := &domain.GenerateSinkAdvertisedListenerReq{
		ClusterID:    req.ClusterId,
		Zones:        req.Zones,
		ServiceName:  req.EndpointServiceName,
		DownstreamID: req.DownstreamId,
	}
	modelResp, err := m.sinkEndpointModel.GenerateSinkAdvertisedListener(ctx, modelReq)
	if err != nil {
		return nil, err
	}

	advertisedListener := &dataflow.SinkAdvertisedListener{
		DomainType:     dataflow.SinkAdvertisedListener_GENERATED,
		RootDomainName: modelResp.RootDomainName,
		Records:        make([]*dataflow.SinkAdvertisedListener_Record, 0, len(modelResp.DNSRecords)),
	}
	for _, record := range modelResp.DNSRecords {
		advertisedListener.Records = append(advertisedListener.Records, &dataflow.SinkAdvertisedListener_Record{
			Zone:          record.AvailabilityZone,
			SubdomainName: record.RecordName,
		})
	}
	return &dataflow.GenerateSinkAdvertisedListenerResponse{
		AdvertisedListener: advertisedListener,
	}, nil
}

func (m *sinkEndpointControllerImpl) ValidateSinkAdvertisedListenerDomain(
	ctx context.Context, req *dataflow.ValidateSinkAdvertisedListenerDomainRequest,
) (*dataflow.ValidateSinkAdvertisedListenerDomainResponse, error) {
	modelReq := &domain.ValidateSinkAdvertisedListenerDomainReq{
		ClusterID:      req.ClusterId,
		ServiceName:    req.EndpointServiceName,
		RootDomainName: req.RootDomainName,
		Zones:          req.Zones,
	}
	resp, err := m.sinkEndpointModel.ValidateSinkAdvertisedListenerDomain(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	return &dataflow.ValidateSinkAdvertisedListenerDomainResponse{
		AdvertisedListener: convertSinkAdvertisedListener(dataflow.SinkAdvertisedListener_GENERATED,
			req.RootDomainName, resp.DNSRecords),
	}, nil
}

func (m *sinkEndpointControllerImpl) GetSinkEndpointAuthInfo(
	ctx context.Context, req *dataflow.GetSinkEndpointAuthInfoRequest,
) (*dataflow.GetSinkEndpointAuthInfoResponse, error) {
	modelReq := &domain.GetSinkEndpointAuthInfoReq{
		ClusterID: req.ClusterId,
	}
	modelResp, err := m.sinkEndpointModel.GetSinkEndpointAuthInfo(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	return &dataflow.GetSinkEndpointAuthInfoResponse{
		CloudAccountId: modelResp.CloudAccountID,
		AwsPrincipalId: modelResp.AWSPrincipalID,
	}, nil
}

func createSinkEndpointRequestDataflow2Model(
	req *dataflow.CreateSinkEndpointRequest,
) *domain.CreateSinkEndpointReq {
	modelReq := &domain.CreateSinkEndpointReq{
		ClusterID:     req.ClusterId,
		Zones:         req.SinkEndpoint.EndpointZones,
		Name:          req.SinkEndpoint.EndpointName,
		ServiceName:   req.SinkEndpoint.EndpointServiceName,
		ServiceRegion: req.SinkEndpoint.ServiceRegion,
		EndpointType:  sinkEndpointTypeDataflow2Entity(req.SinkEndpoint.EndpointType),
		MSKArn:        req.SinkEndpoint.MskArn,
		AuthType:      mskAuthTypeDataflow2Entity(req.SinkEndpoint.AuthType),
	}

	if advertisedListener := req.SinkEndpoint.AdvertisedListener; advertisedListener != nil {
		modelReq.RootDomainName = advertisedListener.RootDomainName
		modelReq.DNSRecords = make([]entity.BizSinkEndpointDNSRecord, 0, len(advertisedListener.Records))
		for _, record := range advertisedListener.Records {
			modelReq.DNSRecords = append(modelReq.DNSRecords, entity.BizSinkEndpointDNSRecord{
				AvailabilityZone: record.Zone,
				RecordName:       record.SubdomainName,
			})
		}
		if advertisedListener.DomainType == dataflow.SinkAdvertisedListener_CUSTOM {
			modelReq.IsCustomDomain = true
		}
	}
	return modelReq
}

func sinkEndpointEntity2Dataflow(endpointEnt *entity.BizSinkEndpoint) *dataflow.SinkEndpoint {
	endpoint := &dataflow.SinkEndpoint{
		Id:                  endpointEnt.ID.Uint64(),
		EndpointName:        endpointEnt.Name,
		EndpointServiceName: endpointEnt.ServiceName,
		Region:              endpointEnt.Region,
		EndpointZones:       endpointEnt.Zones,
		ServiceRegion:       endpointEnt.ServiceRegion,
		Status:              sinkEndpointStatusEntity2Dataflow(endpointEnt.Status),
		EndpointId:          endpointEnt.CloudEndpointID,
		CreatedAt:           timestamppb.New(endpointEnt.CreatedAt),
		CloudAccountId:      endpointEnt.CloudAccountID,
		EndpointAddress:     endpointEnt.EndpointAddress,
		EndpointType:        sinkEndpointTypeEntity2Dataflow(endpointEnt.EndpointType),
		MskArn:              endpointEnt.MSKArn,
		AuthType:            mskAuthTypeEntity2Dataflow(endpointEnt.AuthType),
		BootstrapBroker:     endpointEnt.BootstrapBroker,
	}
	if endpointEnt.RootDomainName != "" {
		domainType := dataflow.SinkAdvertisedListener_GENERATED
		if endpointEnt.IsCustomDomain {
			domainType = dataflow.SinkAdvertisedListener_CUSTOM
		}
		endpoint.AdvertisedListener = convertSinkAdvertisedListener(domainType,
			endpointEnt.RootDomainName, endpointEnt.DNSRecords)
	}
	return endpoint
}

func sinkEndpointTypeDataflow2Entity(t dataflow.SinkEndpointType) entity.BizSinkEndpointType {
	switch t {
	case dataflow.SinkEndpointType_AMAZON_MSK:
		return entity.BizSinkEndpointTypeAmazonMSK
	case dataflow.SinkEndpointType_VPC_ENDPOINT_SERVICE:
		return entity.BizSinkEndpointTypeVPCEndpointService
	default:
		// Empty / unspecified defaults to VPC endpoint service for backward compatibility.
		return entity.BizSinkEndpointTypeVPCEndpointService
	}
}

func sinkEndpointTypeEntity2Dataflow(t entity.BizSinkEndpointType) dataflow.SinkEndpointType {
	switch t {
	case entity.BizSinkEndpointTypeAmazonMSK:
		return dataflow.SinkEndpointType_AMAZON_MSK
	default:
		return dataflow.SinkEndpointType_VPC_ENDPOINT_SERVICE
	}
}

func mskAuthTypeDataflow2Entity(t dataflow.MskAuthType) entity.BizSinkEndpointAuthType {
	switch t {
	case dataflow.MskAuthType_SASL_SCRAM:
		return entity.BizSinkEndpointAuthSASLScram
	case dataflow.MskAuthType_SASL_IAM:
		return entity.BizSinkEndpointAuthSASLIAM
	default:
		return ""
	}
}

func mskAuthTypeEntity2Dataflow(t entity.BizSinkEndpointAuthType) dataflow.MskAuthType {
	switch t {
	case entity.BizSinkEndpointAuthSASLScram:
		return dataflow.MskAuthType_SASL_SCRAM
	case entity.BizSinkEndpointAuthSASLIAM:
		return dataflow.MskAuthType_SASL_IAM
	default:
		return dataflow.MskAuthType_MSK_AUTH_TYPE_UNSPECIFIED
	}
}

func sinkEndpointStatusEntity2Dataflow(status entity.BizSinkEndpointStatus) dataflow.SinkEndpointStatus {
	switch status {
	case entity.BizPLEndpointStatusCreating:
		return dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_CREATING
	case entity.BizPLEndpointStatusPendingAcceptance:
		return dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_PENDING_ACCEPTANCE
	case entity.BizPLEndpointStatusAvailable:
		return dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_AVAILABLE
	case entity.BizPLEndpointStatusFailed:
		return dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_FAILED
	default:
		return dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_UNSPECIFIED
	}
}

func sinkEndpointStatusDataflow2Entity(status dataflow.SinkEndpointStatus) entity.BizSinkEndpointStatus {
	switch status {
	case dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_CREATING:
		return entity.BizPLEndpointStatusCreating
	case dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_PENDING_ACCEPTANCE:
		return entity.BizPLEndpointStatusPendingAcceptance
	case dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_AVAILABLE:
		return entity.BizPLEndpointStatusAvailable
	case dataflow.SinkEndpointStatus_SINK_ENDPOINT_STATUS_FAILED:
		return entity.BizPLEndpointStatusFailed
	default:
		// UNSPECIFIED means no status filter — all records are returned.
		// Proto validation (defined_only = true) ensures no other value reaches here.
		return ""
	}
}

func convertSinkAdvertisedListener(
	domainType dataflow.SinkAdvertisedListener_DomainType,
	rootDomainName string,
	records []entity.BizSinkEndpointDNSRecord,
) *dataflow.SinkAdvertisedListener {
	advertisedListener := &dataflow.SinkAdvertisedListener{
		DomainType:     domainType,
		RootDomainName: rootDomainName,
		Records:        make([]*dataflow.SinkAdvertisedListener_Record, 0, len(records)),
	}
	for _, record := range records {
		advertisedListener.Records = append(advertisedListener.Records, &dataflow.SinkAdvertisedListener_Record{
			Zone:          record.AvailabilityZone,
			SubdomainName: record.RecordName,
		})
	}
	return advertisedListener
}