package cdc_svc

import (
	"context"
	"strings"

	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/service/console_auditlog"
	"github.com/tidbcloud/tiinf/sdk/auditlog"
)

func recordCreateChangefeedAuditLog(
	ctx context.Context,
	req *console.CreateChangefeedReq,
	cf *vo.Changefeed,
	err error,
) {
	details := buildChangefeedDetails(req.GetDisplayName(), cf)
	console_auditlog.SubmitWithStringIDs(
		ctx,
		auditlog.EventCreateChangefeed,
		req.GetOrgId(),
		req.GetProjectId(),
		req.GetClusterId(),
		details,
		err,
	)
}

func recordEditChangefeedAuditLog(
	ctx context.Context,
	req *console.EditChangefeedReq,
	cf *vo.Changefeed,
	err error,
) {
	details := buildChangefeedDetails(req.GetDisplayName(), cf)
	if req.GetChangefeedId() != "" {
		details[auditlog.KeyChangefeedId] = req.GetChangefeedId()
	}
	console_auditlog.SubmitWithStringIDs(
		ctx,
		auditlog.EventEditChangefeed,
		req.GetOrgId(),
		req.GetProjectId(),
		req.GetClusterId(),
		details,
		err,
	)
}

func recordChangefeedIDAuditLog(
	ctx context.Context,
	event auditlog.EventType,
	req *console.ChangefeedID,
	cf *vo.Changefeed,
	err error,
) {
	details := buildChangefeedDetails("", cf)
	if req.GetChangefeedId() != "" {
		details[auditlog.KeyChangefeedId] = req.GetChangefeedId()
	}
	console_auditlog.SubmitWithStringIDs(
		ctx,
		event,
		req.GetOrgId(),
		req.GetProjectId(),
		req.GetClusterId(),
		details,
		err,
	)
}

func recordScaleChangefeedAuditLog(
	ctx context.Context,
	req *console.ScaleChangefeedReq,
	cf *vo.Changefeed,
	err error,
) {
	details := buildChangefeedDetails("", cf)
	if req.GetChangefeedId() != "" {
		details[auditlog.KeyChangefeedId] = req.GetChangefeedId()
	}
	if req.GetRcu() != 0 {
		details[auditlog.KeyMaxRcu] = req.GetRcu()
	}
	console_auditlog.SubmitWithStringIDs(
		ctx,
		auditlog.EventScaleChangefeed,
		req.GetOrgId(),
		req.GetProjectId(),
		req.GetClusterId(),
		details,
		err,
	)
}

func buildChangefeedDetails(displayName string, cf *vo.Changefeed) auditlog.Details {
	details := auditlog.Details{}
	if displayName != "" {
		details[auditlog.KeyDisplayName] = displayName
	}
	if cf == nil {
		return details
	}
	if cf.ChangefeedID != "" {
		details[auditlog.KeyChangefeedId] = cf.ChangefeedID
	}
	if cf.Name != "" {
		details[auditlog.KeyDisplayName] = cf.Name
	}
	if cf.Sink != nil {
		details[auditlog.KeyChangefeedType] = string(cf.Sink.Type)
		fillSinkNetworkDetails(details, cf.Sink)
	}
	if cf.Filter != nil && len(cf.Filter.FilterRule) > 0 {
		details[auditlog.KeyTableFilter] = strings.Join(cf.Filter.FilterRule, ",")
	}
	return details
}

func fillSinkNetworkDetails(details auditlog.Details, sink *vo.SinkInfo) {
	if sink == nil {
		return
	}
	switch sink.Type {
	case constant.ChangefeedTypeKafka:
		if sink.Kafka == nil {
			return
		}
		details[auditlog.KeyNetworkType] = string(sink.Kafka.Network.NetworkType)
		if sink.Kafka.Network.PrivateLinkConnectionId != "" {
			details[auditlog.KeyDownstreamId] = sink.Kafka.Network.PrivateLinkConnectionId
		}
	case constant.ChangefeedTypeMySQL:
		if sink.MySQL == nil {
			return
		}
		details[auditlog.KeyNetworkType] = string(sink.MySQL.Network.NetworkType)
		if sink.MySQL.Network.PrivateLinkConnectionId != "" {
			details[auditlog.KeyDownstreamId] = sink.MySQL.Network.PrivateLinkConnectionId
		}
	}
}