package console

import (
	"context"
	"time"

	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/models/private_link"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/pkgv2/bizerr"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type SinkEndpointController interface {
	dataflow.SinkEndpointServiceServer
}

type sinkEndpointControllerImpl struct {
	dataflow.UnimplementedSinkEndpointServiceServer
	sinkEndpointModel private_link.SinkEndpointModel
}

func NewSinkEndpointController(sinkEndpointModel private_link.SinkEndpointModel) SinkEndpointController {
	return &sinkEndpointControllerImpl{
		sinkEndpointModel: sinkEndpointModel,
	}
}

func (m *sinkEndpointControllerImpl) CreateSinkEndpoint(
	ctx context.Context, req *dataflow.CreateSinkEndpointRequest,
) (*dataflow.SinkEndpoint, error) {
	modelReq := createSinkEndpointRequestDataflow2Model(req)
	modelResp, err := m.sinkEndpointModel.CreateSinkEndpoint(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	// Newly created endpoint has no usage info yet
	return sinkEndpointWithUsageInfo2Dataflow(modelResp.SinkEndpoint, nil), nil
}

func (m *sinkEndpointControllerImpl) ListSinkEndpoints(
	ctx context.Context, req *dataflow.ListSinkEndpointsRequest,
) (*dataflow.ListSinkEndpointsResponse, error) {
	modelReq := &domain.ListSinkEndpointsReq{
		TenantID:  req.OrgId,
		ProjectID: req.ProjectId,
		Provider:  entity.CloudProviderType(req.Provider),
		Region:    req.Region,
		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, sinkEndpointWithUsage2Dataflow(&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,
	}
	modelResp, err := m.sinkEndpointModel.QuerySinkEndpoint(ctx, modelReq)
	if err != nil {
		return nil, err
	}

	sinkEndpointEnt := modelResp.SinkEndpoint
	if sinkEndpointEnt.TenantID.Uint64() != req.OrgId || sinkEndpointEnt.ProjectID.Uint64() != req.ProjectId {
		return nil, bizerr.New(
			bizerr.ErrCodePermission, "the sink endpoint is not belong to the tenant and project")
	}
	return sinkEndpointWithUsageInfo2Dataflow(sinkEndpointEnt, modelResp.UsageInfo), nil
}

func (m *sinkEndpointControllerImpl) DeleteSinkEndpoint(
	ctx context.Context, req *dataflow.DeleteSinkEndpointRequest,
) (*emptypb.Empty, error) {
	queryReq := &domain.QuerySinkEndpointReq{
		ID: req.SinkEndpointId,
	}
	queryResp, err := m.sinkEndpointModel.QuerySinkEndpoint(ctx, queryReq)
	if err != nil {
		return nil, err
	}
	sinkEndpointEnt := queryResp.SinkEndpoint
	if sinkEndpointEnt.TenantID.Uint64() != req.OrgId || sinkEndpointEnt.ProjectID.Uint64() != req.ProjectId {
		return nil, bizerr.New(
			bizerr.ErrCodePermission, "the sink endpoint is not belong to the tenant and project")
	}

	deleteReq := &domain.DeleteSinkEndpointReq{
		ID: req.SinkEndpointId,
	}
	err = m.sinkEndpointModel.DeleteSinkEndpoint(ctx, deleteReq)
	if err != nil {
		return nil, err
	}
	return &emptypb.Empty{}, nil
}

func (m *sinkEndpointControllerImpl) ListSinkAvailabilityZones(
	ctx context.Context, req *dataflow.ListSinkAvailabilityZonesRequest,
) (*dataflow.ListSinkAvailabilityZonesResponse, error) {
	modelReq := &domain.GetAvailabilityZonesReq{
		TenantID:  req.OrgId,
		ProjectID: req.ProjectId,
		Provider:  entity.CloudProviderType(req.Provider),
		Region:    req.Region,
	}
	modelResp, err := m.sinkEndpointModel.GetAvailabilityZones(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	return &dataflow.ListSinkAvailabilityZonesResponse{
		AvailabilityZones: modelResp.Zones,
	}, nil
}

func (m *sinkEndpointControllerImpl) GenerateSinkAdvertisedListener(
	ctx context.Context, req *dataflow.GenerateSinkAdvertisedListenerRequest,
) (*dataflow.GenerateSinkAdvertisedListenerResponse, error) {
	modelReq := &domain.GenerateSinkAdvertisedListenerReq{
		TenantID:     req.OrgId,
		ProjectID:    req.ProjectId,
		Provider:     entity.CloudProviderType(req.Provider),
		Region:       req.Region,
		Zones:        req.Zones,
		ServiceName:  req.EndpointServiceName,
		DownstreamID: req.DownstreamId,
	}
	modelResp, err := m.sinkEndpointModel.GenerateSinkAdvertisedListener(ctx, modelReq)
	if err != nil {
		return nil, err
	}

	return &dataflow.GenerateSinkAdvertisedListenerResponse{
		AdvertisedListener: convertSinkAdvertisedListener(
			dataflow.SinkAdvertisedListener_GENERATED, modelResp.RootDomainName, modelResp.DNSRecords),
	}, nil
}

func (m *sinkEndpointControllerImpl) ValidateSinkAdvertisedListenerDomain(
	ctx context.Context, req *dataflow.ValidateSinkAdvertisedListenerDomainRequest,
) (*dataflow.ValidateSinkAdvertisedListenerDomainResponse, error) {
	modelReq := &domain.ValidateSinkAdvertisedListenerDomainReq{
		TenantID:       req.OrgId,
		ProjectID:      req.ProjectId,
		Provider:       entity.CloudProviderType(req.Provider),
		Region:         req.Region,
		RootDomainName: req.RootDomainName,
		Zones:          req.Zones,
	}
	modelResp, err := m.sinkEndpointModel.ValidateSinkAdvertisedListenerDomain(ctx, modelReq)
	if err != nil {
		return nil, err
	}

	return &dataflow.ValidateSinkAdvertisedListenerDomainResponse{
		AdvertisedListener: convertSinkAdvertisedListener(
			dataflow.SinkAdvertisedListener_CUSTOM, req.RootDomainName, modelResp.DNSRecords),
	}, nil
}

func (m *sinkEndpointControllerImpl) GetSinkEndpointAuthInfo(
	ctx context.Context, req *dataflow.GetSinkEndpointAuthInfoRequest,
) (*dataflow.GetSinkEndpointAuthInfoResponse, error) {
	modelReq := &domain.GetSinkEndpointAuthInfoReq{
		TenantID:  req.OrgId,
		ProjectID: req.ProjectId,
		Provider:  entity.CloudProviderType(req.Provider),
		Region:    req.Region,
	}
	modelResp, err := m.sinkEndpointModel.GetSinkEndpointAuthInfo(ctx, modelReq)
	if err != nil {
		return nil, err
	}
	return &dataflow.GetSinkEndpointAuthInfoResponse{
		Network:        modelResp.Network,
		CloudAccountId: modelResp.CloudAccountID,
		AwsPrincipalId: modelResp.AWSPrincipalID,
	}, nil
}

func createSinkEndpointRequestDataflow2Model(
	req *dataflow.CreateSinkEndpointRequest,
) *domain.CreateSinkEndpointReq {
	modelReq := &domain.CreateSinkEndpointReq{
		TenantID:      req.OrgId,
		ProjectID:     req.ProjectId,
		Provider:      entity.CloudProviderType(req.SinkEndpoint.Provider),
		Region:        req.SinkEndpoint.Region,
		Zones:         req.SinkEndpoint.EndpointZones,
		Name:          req.SinkEndpoint.EndpointName,
		ServiceName:   req.SinkEndpoint.EndpointServiceName,
		ServiceRegion: req.SinkEndpoint.ServiceRegion,
		Properties:    req.SinkEndpoint.Properties,
	}

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

func sinkEndpointWithUsage2Dataflow(endpointWithUsage *domain.SinkEndpointWithUsage) *dataflow.SinkEndpoint {
	return sinkEndpointWithUsageInfo2Dataflow(endpointWithUsage.Endpoint, endpointWithUsage.UsageInfo)
}

func sinkEndpointWithUsageInfo2Dataflow(
	endpointEnt *entity.BizPrivateLinkEndpoint,
	usageInfo *domain.SinkEndpointUsageInfo,
) *dataflow.SinkEndpoint {
	endpoint := &dataflow.SinkEndpoint{
		Id:                  endpointEnt.ID.Uint64(),
		EndpointName:        endpointEnt.Name,
		EndpointServiceName: endpointEnt.ServiceName,
		Provider:            string(endpointEnt.CloudProvider),
		Region:              endpointEnt.Region,
		EndpointZones:       endpointEnt.Zones,
		ServiceRegion:       endpointEnt.ServiceRegion,
		Status:              sinkEndpointStatusEntity2Dataflow(endpointEnt.Status),
		EndpointId:          endpointEnt.CloudEndpointID,
		CloudAccountId:      endpointEnt.CloudAccountID,
		EndpointAddress:     endpointEnt.EndpointAddress,
		CreatedAt:           timestamppb.New(endpointEnt.CreatedAt),
		Properties:          endpointEnt.Properties,
	}
	if endpointEnt.RootDomainName != "" {
		advertisedListener := &dataflow.SinkAdvertisedListener{
			DomainType:     dataflow.SinkAdvertisedListener_GENERATED,
			RootDomainName: endpointEnt.RootDomainName,
			Records:        make([]*dataflow.SinkAdvertisedListener_Record, 0, len(endpointEnt.DNSRecords)),
		}
		if endpointEnt.IsCustomDomain {
			advertisedListener.DomainType = dataflow.SinkAdvertisedListener_CUSTOM
		}
		for _, record := range endpointEnt.DNSRecords {
			advertisedListener.Records = append(advertisedListener.Records, &dataflow.SinkAdvertisedListener_Record{
				Zone:          record.AvailabilityZone,
				SubdomainName: record.RecordName,
			})
		}
		endpoint.AdvertisedListener = advertisedListener
	}

	// Set usage information
	if usageInfo != nil {
		if len(usageInfo.ChangefeedIds) > 0 {
			endpoint.UsedBy = &dataflow.SinkEndpointUsage{
				ChangefeedIds: usageInfo.ChangefeedIds,
			}
		}
		if usageInfo.LastUsedAt != nil {
			endpoint.LastUsedAt = timestamppb.New(*usageInfo.LastUsedAt)
		}
	}

	// Set will delete time: if UnusedAt is set, add retention days; otherwise use DeletedTime if > 0
	if endpointEnt.UnusedAt != nil {
		willDeleteAt := endpointEnt.UnusedAt.AddDate(0, 0, private_link.UnusedEndpointRetentionDays)
		endpoint.WillDeleteAt = timestamppb.New(willDeleteAt)
	} else if endpointEnt.DeletedTime > 0 {
		willDeleteAt := time.Unix(endpointEnt.DeletedTime, 0)
		endpoint.WillDeleteAt = timestamppb.New(willDeleteAt)
	}

	return endpoint
}

func sinkEndpointStatusEntity2Dataflow(status entity.BizPLEndpointStatus) 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.BizPLEndpointStatus {
	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.BizPLEndpointDNSRecord,
) *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
}