package console

import (
	"context"
	"strconv"

	"github.com/tidbcloud/dataflow-service-ng/app/internal/console_auditlog"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
)

func changefeedTypeString(t dataflow.ChangefeedType) string {
	return dataflow.ChangefeedType_name[int32(t)]
}

func networkTypeString(info *dataflow.NetworkInfo) string {
	if info == nil {
		return ""
	}
	return dataflow.NetworkType_name[int32(info.GetNetworkType())]
}

type changefeedAuditContext struct {
	name           string
	changefeedType string
	endpoint       string
	downstreamID   string
}

func kafkaChangefeedAuditEndpoint(kafka *dataflow.KafkaInfo) string {
	if kafka != nil && kafka.GetBroker() != nil {
		return kafka.GetBroker().GetBrokerEndpoints()
	}
	return ""
}

func mysqlChangefeedAuditEndpoint(mysql *dataflow.MySQLInfo) string {
	if mysql != nil && mysql.GetConnection() != nil {
		return mysql.GetConnection().GetEndpoint()
	}
	return ""
}

func s3ChangefeedAuditEndpoint(s3Info *dataflow.S3Info) string {
	if s3Info != nil && s3Info.GetEndpoint() != nil {
		return s3Info.GetEndpoint().GetS3Uri()
	}
	return ""
}

func gcsChangefeedAuditEndpoint(gcsInfo *dataflow.GcsInfo) string {
	if gcsInfo != nil && gcsInfo.GetEndpoint() != nil {
		return gcsInfo.GetEndpoint().GetGcsUri()
	}
	return ""
}

func pulsarChangefeedAuditEndpoint(pulsar *dataflow.PulsarInfo) string {
	if pulsar != nil {
		return pulsar.GetEndpoint()
	}
	return ""
}

func serverlessChangefeedAuditEndpointDownstream(
	serverless *dataflow.ServerlessTidbInfo,
) (endpoint, downstreamID string) {
	if serverless == nil {
		return "", ""
	}
	if serverless.GetClusterId() != 0 {
		downstreamID = strconv.FormatUint(serverless.GetClusterId(), 10)
	}
	if serverless.GetMysqlInfo() != nil && serverless.GetMysqlInfo().GetConnection() != nil {
		endpoint = serverless.GetMysqlInfo().GetConnection().GetEndpoint()
	}
	return endpoint, downstreamID
}

func changefeedAuditEndpointDownstream(
	cfType dataflow.ChangefeedType,
	kafka *dataflow.KafkaInfo,
	mysql *dataflow.MySQLInfo,
	s3Info *dataflow.S3Info,
	gcsInfo *dataflow.GcsInfo,
	pulsar *dataflow.PulsarInfo,
	serverless *dataflow.ServerlessTidbInfo,
) (endpoint, downstreamID string) {
	switch cfType {
	case dataflow.ChangefeedType_KAFKA:
		endpoint = kafkaChangefeedAuditEndpoint(kafka)
	case dataflow.ChangefeedType_MYSQL:
		endpoint = mysqlChangefeedAuditEndpoint(mysql)
	case dataflow.ChangefeedType_S3:
		endpoint = s3ChangefeedAuditEndpoint(s3Info)
	case dataflow.ChangefeedType_GCS:
		endpoint = gcsChangefeedAuditEndpoint(gcsInfo)
	case dataflow.ChangefeedType_PULSAR:
		endpoint = pulsarChangefeedAuditEndpoint(pulsar)
	case dataflow.ChangefeedType_SERVERLESS_TIDB:
		endpoint, downstreamID = serverlessChangefeedAuditEndpointDownstream(serverless)
	}
	return endpoint, downstreamID
}

func changefeedBaseDetails(
	changefeedID uint64,
	name, changefeedType, networkType, endpoint, downstreamID string,
	sinkEndpointID *uint64,
	rErr error,
) auditlog.Details {
	details := auditlog.Details{
		auditlog.KeyError:          rErr,
		auditlog.KeyChangefeedType: changefeedType,
	}
	if changefeedID != 0 {
		details[auditlog.KeyChangefeedId] = strconv.FormatUint(changefeedID, 10)
	}
	if name != "" {
		details[auditlog.KeyJobName] = name
	}
	if networkType != "" {
		details[auditlog.KeyNetworkType] = networkType
	}
	if endpoint != "" {
		details[auditlog.KeyEndpoint] = endpoint
	}
	if downstreamID != "" {
		details[auditlog.KeyDownstreamId] = downstreamID
	}
	if sinkEndpointID != nil {
		details[auditlog.KeyPrivateEndpointForExternalServiceId] = strconv.FormatUint(*sinkEndpointID, 10)
	}
	return details
}

func (c *bizChangefeedControllerImpl) recordAuditLogForCreateChangefeed(
	ctx context.Context,
	cluster *mgmt.Cluster,
	req *dataflow.ChangefeedCreateReq,
	changefeedID uint64,
	rErr error,
) {
	var sinkEndpointID *uint64
	if req.GetNetworkInfo() != nil && req.GetNetworkInfo().SinkEndpointId != nil {
		sinkEndpointID = req.GetNetworkInfo().SinkEndpointId
	}
	endpoint, downstreamID := changefeedAuditEndpointDownstream(
		req.GetChangefeedType(),
		req.GetKafkaInfo(),
		req.GetMysqlInfo(),
		req.GetS3Info(),
		req.GetGcsInfo(),
		req.GetPulsarInfo(),
		req.GetServerlessTidbInfo(),
	)
	details := changefeedBaseDetails(
		changefeedID,
		req.GetName(),
		changefeedTypeString(req.GetChangefeedType()),
		networkTypeString(req.GetNetworkInfo()),
		endpoint,
		downstreamID,
		sinkEndpointID,
		rErr,
	)
	console_auditlog.SubmitAsync(ctx, auditlog.EventCreateChangefeed, cluster, details, rErr == nil)
}

func (c *bizChangefeedControllerImpl) recordAuditLogForEditChangefeed(
	ctx context.Context,
	cluster *mgmt.Cluster,
	req *dataflow.ChangefeedEditReq,
	oldName string,
	rErr error,
) {
	endpoint, downstreamID := changefeedAuditEndpointDownstream(
		req.GetChangefeedType(),
		req.GetKafkaInfo(),
		req.GetMysqlInfo(),
		req.GetS3Info(),
		req.GetGcsInfo(),
		req.GetPulsarInfo(),
		req.GetServerlessTidbInfo(),
	)
	details := changefeedBaseDetails(
		req.GetChangefeedId(),
		req.GetName(),
		changefeedTypeString(req.GetChangefeedType()),
		"",
		endpoint,
		downstreamID,
		nil,
		rErr,
	)
	if oldName != "" || req.GetName() != "" {
		details[auditlog.KeyOldValue] = oldName
		details[auditlog.KeyNewValue] = req.GetName()
	}
	console_auditlog.SubmitAsync(ctx, auditlog.EventEditChangefeed, cluster, details, rErr == nil)
}

func (c *bizChangefeedControllerImpl) recordAuditLogForDeleteChangefeed(
	ctx context.Context,
	cluster *mgmt.Cluster,
	req *dataflow.ChangefeedID,
	auditCtx changefeedAuditContext,
	rErr error,
) {
	details := changefeedBaseDetails(
		req.GetChangefeedId(),
		auditCtx.name,
		auditCtx.changefeedType,
		"",
		auditCtx.endpoint,
		auditCtx.downstreamID,
		nil,
		rErr,
	)
	console_auditlog.SubmitAsync(ctx, auditlog.EventDeleteChangefeed, cluster, details, rErr == nil)
}

func (c *bizChangefeedControllerImpl) recordAuditLogForPauseChangefeed(
	ctx context.Context,
	cluster *mgmt.Cluster,
	req *dataflow.ChangefeedID,
	auditCtx changefeedAuditContext,
	rErr error,
) {
	details := changefeedBaseDetails(
		req.GetChangefeedId(),
		auditCtx.name,
		auditCtx.changefeedType,
		"",
		auditCtx.endpoint,
		auditCtx.downstreamID,
		nil,
		rErr,
	)
	console_auditlog.SubmitAsync(ctx, auditlog.EventStopChangefeed, cluster, details, rErr == nil)
}

func (c *bizChangefeedControllerImpl) recordAuditLogForResumeChangefeed(
	ctx context.Context,
	cluster *mgmt.Cluster,
	req *dataflow.ChangefeedID,
	auditCtx changefeedAuditContext,
	rErr error,
) {
	details := changefeedBaseDetails(
		req.GetChangefeedId(),
		auditCtx.name,
		auditCtx.changefeedType,
		"",
		auditCtx.endpoint,
		auditCtx.downstreamID,
		nil,
		rErr,
	)
	console_auditlog.SubmitAsync(ctx, auditlog.EventStartChangefeed, cluster, details, rErr == nil)
}

func (c *bizChangefeedControllerImpl) recordAuditLogForScaleChangefeed(
	ctx context.Context,
	cluster *mgmt.Cluster,
	req *dataflow.ChangefeedScaleReq,
	auditCtx changefeedAuditContext,
	rErr error,
) {
	details := changefeedBaseDetails(
		req.GetChangefeedId(),
		auditCtx.name,
		auditCtx.changefeedType,
		"",
		auditCtx.endpoint,
		auditCtx.downstreamID,
		nil,
		rErr,
	)
	console_auditlog.SubmitAsync(ctx, auditlog.EventScaleChangefeed, cluster, details, rErr == nil)
}

func (c *bizChangefeedControllerImpl) loadChangefeedAuditContext(
	ctx context.Context, changefeedID uint64,
) changefeedAuditContext {
	getResp, bizErr := c.changefeedModel.GetBizChangefeed(ctx, &domain.GetBizChangefeedReq{
		BizChangefeedID: changefeedID,
	})
	if bizErr != nil || getResp == nil || getResp.BizChangefeed == nil {
		return changefeedAuditContext{}
	}
	cf := getResp.BizChangefeed
	cfType := domain.TiCDCChangefeedDownStreamType2ConnectorType(cf.DownstreamType)
	auditCtx := changefeedAuditContext{
		name:           cf.Name,
		changefeedType: changefeedTypeString(cfType),
	}

	adaptor, ok := c.downstreamFactory.Get(cfType)
	if !ok {
		return auditCtx
	}
	createReq, assembleErr := adaptor.AssembleGetResponse(cf)
	if assembleErr != nil || createReq == nil {
		return auditCtx
	}
	auditCtx.endpoint, auditCtx.downstreamID = changefeedAuditEndpointDownstream(
		createReq.GetChangefeedType(),
		createReq.GetKafkaInfo(),
		createReq.GetMysqlInfo(),
		createReq.GetS3Info(),
		createReq.GetGcsInfo(),
		createReq.GetPulsarInfo(),
		createReq.GetServerlessTidbInfo(),
	)
	return auditCtx
}