package userbehavior

import (
	"context"
	"fmt"
	"math/rand"
	"slices"
	"strconv"
	"strings"
	"time"

	"github.com/go-logr/logr"
	larkim "github.com/larksuite/oapi-sdk-go/v3/service/im/v1"
	userv1 "github.com/tidbcloud/aisupport/packages/sdk-go/contracts/api/core-server/user/v1"
	coreuser "github.com/tidbcloud/aisupport/packages/sdk-go/contracts/entity/core-server/user/v1"
	"github.com/tidbcloud/devops/entity"
	"github.com/tidbcloud/devops/entity/event"
	tntent "github.com/tidbcloud/devops/entity/tenant"
	ubent "github.com/tidbcloud/devops/entity/userbehavior"
	"github.com/tidbcloud/devops/pkg/infra/clients/accountcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/billingcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/coresvccli"
	"github.com/tidbcloud/devops/pkg/infra/clients/dataflowcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/larkcli"
	"github.com/tidbcloud/devops/pkg/infra/clients/llm_cli"
	"github.com/tidbcloud/devops/pkg/infra/clients/sendgridcli"
	"github.com/tidbcloud/devops/pkg/infra/config"
	"github.com/tidbcloud/devops/pkg/infra/eventbus"
	"github.com/tidbcloud/devops/pkg/infra/log"
	gormutil "github.com/tidbcloud/devops/pkg/infra/utils/gorm"
	"github.com/tidbcloud/devops/pkg/infra/utils/pointer"
	"github.com/tidbcloud/devops/pkg/service"
	"github.com/tidbcloud/devops/pkg/service/operation"
	"github.com/tidbcloud/devops/pkg/service/tenant"
	"github.com/tidbcloud/devops/pkg/service/tenant/region"
	"github.com/tidbcloud/devops/pkg/service/userbehavior/externallarkcli"
	"github.com/tidbcloud/devops/pkg/service/userbehavior/gtm"
	"github.com/tidbcloud/devops/pkg/service/userbehavior/repo"
	"github.com/tidbcloud/pkgv2/db"

	tidbcloudv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	mgmtevent "github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"go.uber.org/dig"
	"golang.org/x/text/cases"
	"golang.org/x/text/language"
	"gorm.io/gorm"
)

// nolint:unused
const (
	dbaasTenantPlanEventGroupKey = "userbehavior_plan_change"
	dbaasAIStartupEventGroupKey  = "devops_userbehavior_startup_event"
	dbaasClusterEventGroupKey    = "devops_userbehavior_cluster_event"
	dbaasDisplayEventGroupKey    = "devops_userbehavior_display_event"
)

type ServiceParam struct {
	dig.In

	Conf           *config.RootConfig
	Logger         logr.Logger
	DBProvider     db.TransProvider
	UserService    userv1.UserServiceClient
	TenantService  tenant.TenantServiceAPI
	LarkCli        *larkcli.LarkClient
	AccountSvcCli  accountcli.Client
	BillingSvcCli  billingcli.Client
	CoreSvcCli     *coresvccli.CoreSvcCli
	DataflowSvcCli *dataflowcli.DataflowCli
	EventBus       *eventbus.EventBus
	OpSvc          operation.OperationServiceAPI
	LLMClient      *llm_cli.LLMCli
}

type ServiceResult struct {
	dig.Out

	API     UserBehaviorServiceAPI
	Service service.Service `group:"service"`
}

type svc struct {
	*service.JobService

	conf                 config.UserBehaviorServiceConfig
	ubCh                 chan UBEvent
	tenantEventCaptureCh chan *tenantEventCaptureEvent
	tenantEventCaptureDB *gorm.DB
	larkCliSet           *larkcli.LarkClient
	repo                 repo.UserBehaviorRepo
	logger               logr.Logger
	userSvc              userv1.UserServiceClient
	tenantSvc            tenant.TenantServiceAPI
	accountCli           accountcli.Client
	billingCli           billingcli.Client
	coresvcCli           *coresvccli.CoreSvcCli
	dataflowCli          *dataflowcli.DataflowCli
	eventBus             *eventbus.EventBus
	externalLarkCli      externallarkcli.ExternalLarkCli
	llm                  *llm_cli.LLMCli
	handledTenant        map[uint64]struct{} // account service send repeated active event

	// GTM components
	sendGridCli  sendgridcli.Client
	gtmEnricher  *gtm.Enricher
	gtmQualifier *gtm.SimpleQualifier
	gtmMatcher   *gtm.ReferenceMatcher
	gtmComposer  *gtm.EmailComposer
	nowFn        func() time.Time
	randIntnFn   func(int) int
}

func NewService(p ServiceParam) (ServiceResult, error) {
	base := service.NewJobService("UserBehaviorService", service.WithDummy(p.Conf.Services.UserBehavior.Dummy))
	if p.Conf.Services.UserBehavior.Dummy {
		return ServiceResult{
			API:     &DummyUserBehaviorServiceAPI{},
			Service: base,
		}, nil
	}

	conf := p.Conf.Services.UserBehavior
	ubCh := make(chan UBEvent, 100)
	tenantEventCaptureCh := make(chan *tenantEventCaptureEvent, 1000)
	larCliSet := p.LarkCli
	repo := repo.NewUBRepo(p.DBProvider)
	logger := p.Logger.WithName("UserBehaviorService")
	userSvc := p.UserService
	tenantSvc := p.TenantService
	accountCli := p.AccountSvcCli
	billingcli := p.BillingSvcCli
	coresvccli := p.CoreSvcCli
	dataflowcli := p.DataflowSvcCli
	eventbus := p.EventBus

	if err := validateTenantEventCaptureConfig(conf.TenantEventCapture); err != nil {
		return ServiceResult{}, err
	}

	var tenantEventCaptureDB *gorm.DB
	if conf.TenantEventCapture.Enabled {
		var connectErr error
		targetDBConf := tenantEventCaptureDBConfig(conf.TenantEventCapture)
		tenantEventCaptureDB, connectErr = gormutil.Connect(targetDBConf, "TenantEventCaptureDB")
		if connectErr != nil {
			return ServiceResult{}, fmt.Errorf("connect tenant event capture db failed: %w", connectErr)
		}
		if conf.TenantEventCapture.AutoMigrate {
			if migrateErr := migrateTenantEventCaptureTable(tenantEventCaptureDB); migrateErr != nil {
				return ServiceResult{}, fmt.Errorf("auto migrate tenant event capture table failed: %w", migrateErr)
			}
		}
	}

	s := &svc{
		JobService: base,

		conf:                 conf,
		ubCh:                 ubCh,
		tenantEventCaptureCh: tenantEventCaptureCh,
		tenantEventCaptureDB: tenantEventCaptureDB,
		larkCliSet:           larCliSet,
		repo:                 repo,
		logger:               logger,
		userSvc:              userSvc,
		tenantSvc:            tenantSvc,
		accountCli:           accountCli,
		billingCli:           billingcli,
		coresvcCli:           coresvccli,
		dataflowCli:          dataflowcli,
		eventBus:             eventbus,
		llm:                  p.LLMClient,
		handledTenant:        make(map[uint64]struct{}),
	}

	// update chats
	//err := s.SyncLarkChats(context.Background())
	//if err != nil {
	//return ServiceResult{API: s, Service: s}, err
	//}

	s.externalLarkCli = externallarkcli.NewExternalLarkCli()

	// Initialize GTM components
	sendGridCli, err := sendgridcli.NewClient(p.Conf)
	if err != nil {
		return ServiceResult{}, fmt.Errorf("failed to create SendGrid client: %w", err)
	}
	s.sendGridCli = sendGridCli
	s.gtmEnricher = gtm.NewEnricher(p.LLMClient)
	s.gtmQualifier = gtm.NewSimpleQualifier(logger.WithName("GTMQualifier"))
	s.gtmMatcher = gtm.NewReferenceMatcher(logger.WithName("GTMMatcher"), conf.GTM.UseCaseProfiles)
	s.gtmComposer = gtm.NewEmailComposer(logger.WithName("GTMComposer"), p.LLMClient)

	s.AddJob(
		service.NewConstantCronJob("SyncLarkChats", *s.conf.SyncLarkChatsInterval, s.SyncLarkChats),
		service.NewConstantCronJob("ScanDMJobEvents", *s.conf.SyncLarkChatsInterval, s.ScanDMJobEvent),
		service.NewEventJob("UBEventDispatch", s.ubCh, s.DispatchUBEvents),
		service.NewEventJob("TenantEventCapture", s.tenantEventCaptureCh, s.DispatchTenantEventCapture),
		service.NewConstantCronJob("ProcessQueuedGTMEmails", s.conf.GTM.EmailQueueScanInterval, s.ProcessQueuedGTMEmails),
	)

	s.eventBus.DBaasTenantPlanEvent.Subscribe(dbaasTenantPlanEventGroupKey, s.HandleDBaasTenantPlanEvent)
	s.eventBus.DBaasTenantEvent.Subscribe(dbaasAIStartupEventGroupKey, s.HandleDBaasAITenantEvent)
	s.eventBus.DBaaSClusterEvent.Subscribe(dbaasClusterEventGroupKey, s.HandleDBaaSClusterEvent)
	s.eventBus.DBaaSDisplayEvent.Subscribe(dbaasDisplayEventGroupKey, s.HandleDBaaSDisplayEvent)
	return ServiceResult{API: s, Service: s}, nil
}

// Currently only sync from global
func (s *svc) SyncLarkChats(ctx context.Context) error {
	logger := log.FromContextOrDef(ctx, s.logger)

	chatCli := s.larkCliSet.Chat

	chats, err := chatCli.ListChats(ctx)
	if err != nil {
		return fmt.Errorf("get chats from Lark failed: %w", err)
	}

	for _, chat := range chats {
		// Only sync chats whose name contains 'UB'
		if chat.Name == nil || !strings.Contains(*chat.Name, "UB") {
			continue
		}

		chatEnt := &entity.LarkChat{
			Name:       chat.Name,
			LarkTp:     coreuser.LarkType_LARK_TYPE_GLOBAL,
			IsExternal: false,
			ChatID:     chat.ChatId,
			AvatarURL:  chat.Avatar,
		}
		_, err := s.repo.UpsertChatByChatId(ctx, chatEnt)
		if err != nil {
			logger.V(6).Error(err, "upsert chat failed", "chat_entity", chatEnt)
		}
	}

	return nil

}

// Event Dispatchers
type eventInfo struct {
	ForceSkip        bool
	isPaidTenant     bool
	isInternalTenant bool
	bizRegion        string
	country          string
	tenantID         string
	isSharedCluster  bool
	ExtraInfo        string
}

func (s *svc) DispatchUBEvents(ctx context.Context, ev UBEvent) error {
	logger := log.FromContextOrDef(ctx, s.logger)

	evInfo, err := s.PreprocessUBEvent(ctx, ev)
	if err != nil {
		return err
	}
	// not all upstream events are used
	if evInfo.ForceSkip {
		return nil
	}

	subs, err := s.repo.QuerySubscriptions(ctx, repo.SubscriptionOpts.IsUBEvent(ev.GetUBEventType()))
	if err != nil {
		return err
	}

	for _, sub := range subs {
		if !evInfo.isPaidTenant && sub.OnlyPaidTenant {
			continue
		}

		if evInfo.isInternalTenant && sub.OnlyExternalTenant {
			continue
		}

		if evInfo.isSharedCluster && sub.OnlyDedicatedCluster {
			continue
		}

		// tenant filter
		tenantId := evInfo.tenantID
		if len(sub.Selector.TenantSelectors) != 0 && len(sub.Selector.TenantSelectors[0]) != 0 &&
			!slices.Contains(sub.Selector.TenantSelectors, ubent.ToTenantSelectorTp(tenantId)) {
			continue
		}

		// region/country filter: selectors can contain region codes (lowercase, e.g. "na") or
		// country codes (uppercase, e.g. "US"). Match event's bizRegion or country against them.
		if len(sub.Selector.RegionSelectors) != 0 {
			bizRegion := strings.ToLower(evInfo.bizRegion)
			country := strings.ToLower(evInfo.country)
			regionMatch := slices.Contains(sub.Selector.RegionSelectors, ubent.ToRegionSelectorTp(bizRegion))
			countryMatch := country != "" && slices.Contains(sub.Selector.RegionSelectors, ubent.RegionSelectorTp(country))
			if !regionMatch && !countryMatch {
				continue
			}
		}

		// text-based filter
		if len(sub.TextFilter) > 0 {
			result, err := ev.CustomFilter(ctx, sub.TextFilter)
			if err != nil {
				logger.Info("error when applying text filter", "sub_id", sub.ID, "filter_content", sub.TextFilter, "error", err)
			} else if !result {
				continue
			}
		}

		// Try to build custom card first
		var cardContent string
		customCard, err := ev.BuildCustomCard(ctx)
		if err != nil {
			logger.Error(err, "failed to build custom card, falling back to standard",
				"event_type", ev.GetUBEventType())
		}

		if customCard != nil {
			cardContent, err = customCard.String()
			if err != nil {
				logger.Error(err, "failed to serialize custom card, falling back to standard")
				customCard = nil
			}
		}

		if customCard == nil {
			// Fallback to standard card
			text, err := ev.ProcessTemplate(ctx, sub.Template)
			if err != nil {
				return err
			}

			if len(text) == 0 {
				text = ev.DefaultText(ctx)
			}

			cardUrl := ""
			if sub.DetailLink != nil {
				cardUrl = *sub.DetailLink
			}
			standardCard := s.buildCard(text, cardUrl)
			cardContent, err = standardCard.String()
			if err != nil {
				return err
			}
		}

		for _, chatSimpl := range sub.LarkChats {
			chat, err := s.repo.LoadChat(ctx, chatSimpl.ID)
			if err != nil {
				logger.Error(err, "load chat failed", "chat_id", chatSimpl.ID)
				continue
			}

			if chat.IsExternal {
				if chat.ChatURL != nil {
					go s.externalLarkCli.SendCard(ctx, *chat.ChatURL, cardContent)
				}
				continue
			}

			if chat.ChatID == nil {
				continue
			}

			req := larkim.NewCreateMessageReqBuilder().
				ReceiveIdType("chat_id").
				Body(&larkim.CreateMessageReqBody{
					ReceiveId: pointer.Pointer(*chat.ChatID),
					MsgType:   pointer.Pointer("interactive"),
					Content:   &cardContent,
				}).Build()

			// do dispatching in separate goroutines to avoid blocking
			go s.larkCliSet.Raw.Im.Message.Create(ctx, req)
		}
	}

	return nil
}

func parseLLMTriageFields(triage string) map[string]string {
	fields := map[string]string{}
	for _, line := range strings.Split(triage, "\n") {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		line = strings.TrimPrefix(line, "-")
		line = strings.TrimSpace(line)
		// Normalize full-width colon for key/value split.
		line = strings.ReplaceAll(line, "：", ":")
		parts := strings.SplitN(line, ":", 2)
		if len(parts) != 2 {
			continue
		}
		key := strings.ToLower(strings.TrimSpace(parts[0]))
		val := strings.TrimSpace(parts[1])
		if key != "" && val != "" {
			fields[key] = val
		}
	}
	return fields
}

func extractContactName(email string) *string {
	parts := strings.Split(email, "@")
	if len(parts) != 2 {
		return nil
	}
	name := strings.ReplaceAll(parts[0], ".", " ")
	name = strings.ReplaceAll(name, "_", " ")
	name = cases.Title(language.English).String(name)
	return &name
}

func (s *svc) resolveGTMContactInfo(ctx context.Context, tenantID uint64, signupEmail string) (*string, *string) {
	fallbackContactName := extractContactName(signupEmail)
	if s.accountCli == nil {
		return fallbackContactName, nil
	}

	resp, err := s.accountCli.GetTenantUsers(ctx, strconv.FormatUint(tenantID, 10), accountcli.GetUsersReq{
		Search:    signupEmail,
		Page:      1,
		PageSize:  10,
		FullEmail: true,
	}, "")
	if err != nil {
		s.logger.V(6).Info("failed to resolve GTM contact from account service", "tenant_id", tenantID, "email", signupEmail, "error", err)
		return fallbackContactName, nil
	}
	s.logger.V(6).Info("account service GetTenantUsers response", "tenant_id", tenantID, "email", signupEmail, "response", resp)

	for _, user := range resp.Items {
		if !strings.EqualFold(strings.TrimSpace(user.Email), strings.TrimSpace(signupEmail)) {
			continue
		}

		firstName := strings.TrimSpace(user.FirstName)
		if firstName == "" {
			break
		}

		fullName := strings.TrimSpace(strings.Join([]string{firstName, strings.TrimSpace(user.LastName)}, " "))
		if fullName == "" {
			fullName = firstName
		}

		return pointer.Pointer(fullName), pointer.Pointer(firstName)
	}

	return fallbackContactName, nil
}

func normalizeTriageText(s string) string {
	replacer := strings.NewReplacer(
		"‑", "-", // non-breaking hyphen
		"–", "-", // en dash
		"—", "-", // em dash
		"−", "-", // minus sign
		"\u00a0", " ", // non-breaking space
		"\u202f", " ", // narrow no-breaking space
		"\u2007", " ", // figure space
		"\u200b", "", // zero-width space
		"\u3000", " ", // ideographic space
		"：", ":",
		"\t", " ",
		"\n", " ",
	)
	s = strings.ToLower(strings.TrimSpace(replacer.Replace(s)))
	return strings.Join(strings.Fields(s), " ")
}

func isEducationalTriage(triage string) bool {
	if strings.TrimSpace(triage) == "" {
		return false
	}
	fields := parseLLMTriageFields(triage)
	company := normalizeTriageText(fields["company name"])
	vertical := normalizeTriageText(fields["vertical"])
	full := normalizeTriageText(triage)

	eduKeywords := []string{
		"school district",
		"public schools",
		"school system",
		"board of education",
		"university",
		"college",
		"community college",
		"k-12",
		"high school",
		"middle school",
		"elementary school",
		"primary school",
		"secondary school",
	}

	for _, kw := range eduKeywords {
		kw = normalizeTriageText(kw)
		if (company != "" && strings.Contains(company, kw)) ||
			(vertical != "" && strings.Contains(vertical, kw)) ||
			strings.Contains(full, kw) {
			return true
		}
	}

	return false
}

func (s *svc) PreprocessUBEvent(ctx context.Context, ev UBEvent) (eventInfo, error) {
	ret := eventInfo{}
	//logger := log.FromContextOrDef(ctx, s.logger)
	switch v := ev.(type) {
	case *tenantPlanUBEvent:
		tenant, err := s.tenantSvc.GetTenant(ctx, v.TenantID)
		if err != nil {
			return ret, fmt.Errorf("call tenantSvc.GetTenant() failed: %w", err)
		}
		if tenant.SalesForceRegion == tntent.TenantRegionUnknown {
			bizRegion := string(tenant.CountryRegion)
			v.bizRegion = &bizRegion
		} else {
			bizRegion := string(tenant.SalesForceRegion)
			v.bizRegion = &bizRegion
		}
		ret.country = tenant.Country

		v.tenantAlias = &tenant.Alias

		if tenant.Kind == tntent.TenantKindInternal || tenant.SalesForceKind == tntent.TenantKindInternal {
			ret.isInternalTenant = true
		}

		// Use accountsvc when cannot get info from tenantsvc
		if v.tenantAlias == nil || len(*v.tenantAlias) == 0 {
			org, err := s.accountCli.GetTenant(ctx, v.TenantID)
			if err != nil {
				//logger.V(6).Error(err, "cannot get org from accountsvc", "tenant_id", v.TenantID)
				return ret, fmt.Errorf("call accountSvc.GetTenant() failed: %w, tenant_id: %d", err, v.TenantID)
			} else {
				v.tenantAlias = &org.Alias
				v.tenantName = &org.Name
				if tntent.TenantKind(org.Kind) == tntent.TenantKindInternal {
					ret.isInternalTenant = true
				}
			}
		}

		if v.ChangedPlan == "poc" || v.ChangedPlan == "on_demand" ||
			v.EffectivePlan == "poc" || v.EffectivePlan == "on_demand" ||
			tenant.Plan == tntent.TenantPlanPoc || tenant.Plan == tntent.TenantPlanOnDemand {
			ret.isPaidTenant = true
		}
		ret.bizRegion = *v.bizRegion
		ret.tenantID = strconv.FormatUint(tenant.Id, 10)
		// HACK: for tenant events, `isSharedCluster` should always be false to avoid being filtered out.
		ret.isSharedCluster = false

	case *startupUBEvent:
		s.logger.Info("preprocess startupUBEvent", "event", v, "orgEventType", v.OrgEventType)

		if v.OrgEventType != event.OrgEventTypeActivate {
			s.logger.Info("skip non-activate startupUBEvent", "event", v, "orgEventType", v.OrgEventType, "expected", event.OrgEventTypeActivate)
			ret.ForceSkip = true
			break
		}

		if _, ok := s.handledTenant[v.OrgId]; ok {
			s.logger.Info("skip repeated startupClusterUBEvent for the same tenant", "tenantID", v.OrgId)
			ret.ForceSkip = true
			break
		}
		s.handledTenant[v.OrgId] = struct{}{}

		org, err := s.accountCli.GetTenant(ctx, v.OrgId)
		if err != nil {
			return ret, fmt.Errorf("call accountSvc.GetTenant() failed: %w, tenant_id: %d", err, v.OrgId)
		}
		v.tenantName = &org.Name
		ret.country = org.GetCountry()
		bizRegion := string(region.CountryToRegion(strings.ToUpper(ret.country)))
		if s.tenantSvc != nil {
			tenant, err := s.tenantSvc.GetTenant(ctx, v.OrgId)
			if err != nil {
				s.logger.Info("fallback to account country region for startupUBEvent", "orgId", v.OrgId, "country", ret.country, "error", err)
			} else if tenant != nil && tenant.SalesForceRegion != tntent.TenantRegionUnknown {
				bizRegion = string(tenant.SalesForceRegion)
			}
		}
		v.bizRegion = &bizRegion
		v.emailDomain = getEmailDomain(org.Email)
		if tntent.TenantKind(org.Kind) == tntent.TenantKindInternal {
			s.logger.Info("skip internal tenant startupUBEvent", "orgId", v.OrgId, "kind", org.Kind)
			ret.isInternalTenant = true
			ret.ForceSkip = true
			break
		}
		if isPersonalEmail(v.emailDomain) {
			s.logger.Info("skip startupUBEvent with personal email", "orgId", v.OrgId, "emailDomain", v.emailDomain)
			ret.ForceSkip = true
			break
		}

		ret.bizRegion = *v.bizRegion
		ret.tenantID = strconv.FormatUint(v.OrgId, 10)
		// HACK: avoid being filtered out.
		ret.isSharedCluster = false
		ret.isPaidTenant = true

		// Perform company enrichment using GTM enricher
		searchResult, err := s.gtmEnricher.EnrichByEmailDomain(ctx, v.emailDomain)
		if err != nil {
			s.logger.Error(err, "failed to perform enrichment", "emailDomain", v.emailDomain)
		} else {
			s.logger.Info("enrichment result", "result", searchResult)
			v.extraInfo = searchResult
			v.applyHeadquarterDisplayRegion()
			if isEducationalTriage(searchResult) {
				s.logger.Info("skip startupUBEvent for educational organization", "orgId", v.OrgId, "emailDomain", v.emailDomain)
				ret.ForceSkip = true
				break
			}
		}

		if v.extraInfo == "" {
			break // Skip if no enrichment
		}

		// GTM Pipeline starts here (enrichment is always attempted; GTM email pipeline is gated)
		if !s.conf.GTM.Enabled {
			break // Skip if GTM disabled
		}

		geoMatched, geoMatchedType := matchGeoSelectors(s.conf.GTM.EnabledGeoSelectors, *v.bizRegion, ret.country)
		if !geoMatched {
			s.logger.V(6).Info("skip GTM pipeline (geo selectors mismatch)",
				"tenant_id", v.OrgId,
				"enabledGeoSelectors_len", len(s.conf.GTM.EnabledGeoSelectors),
				"biz_region", *v.bizRegion,
				"country", ret.country,
				"geo_match_type", geoMatchedType)
			break
		}

		// Parse enrichment (tries JSON first, falls back to text)
		enrichment, err := gtm.ParseEnrichment(v.extraInfo)
		if err != nil {
			s.logger.Error(err, "failed to parse enrichment", "tenant_id", v.OrgId)
			break
		}

		contactName, contactFirstName := s.resolveGTMContactInfo(ctx, v.OrgId, org.Email)

		// Qualification check
		qualified, reason := s.gtmQualifier.IsQualified(enrichment)

		// Create lead record (always, even if not qualified)
		lead := &ubent.GTMLead{
			TenantID:            v.OrgId,
			Email:               org.Email,
			EmailDomain:         v.emailDomain,
			ContactName:         contactName,
			ContactFirstName:    contactFirstName,
			CompanyName:         &enrichment.CompanyName,
			Headquarter:         &enrichment.Headquarter,
			Vertical:            &enrichment.Vertical,
			BusinessModel:       &enrichment.BusinessModel,
			AnnualRevenue:       &enrichment.AnnualRevenue,
			FundingSeries:       &enrichment.FundingSeries,
			RawEnrichment:       v.extraInfo,
			IsQualified:         qualified,
			QualificationReason: &reason,
		}

		err = s.repo.CreateGTMLead(ctx, lead)
		if err != nil {
			s.logger.Error(err, "failed to create lead", "tenant_id", v.OrgId)
			break
		}

		v.gtmLead = lead

		// Stop if not qualified or low confidence
		if !qualified || enrichment.ConfidenceLevel == "low" {
			s.logger.V(6).Info("lead not qualified or low confidence",
				"tenant_id", v.OrgId, "reason", reason)
			break
		}

		// Match use case
		matchResult := s.gtmMatcher.Match(lead)
		if !matchResult.Matched {
			s.logger.V(6).Info("no use case match",
				"tenant_id", v.OrgId, "reason", matchResult.Reason)
			break
		}

		if err := s.persistGTMLeadMatch(ctx, lead, matchResult); err != nil {
			s.logger.Error(err, "failed to persist GTM lead match result", "tenant_id", v.OrgId)
			break
		}

		emailDisabled, emailDisabledMatchType := s.shouldSkipGTMEmailForGeo(*v.bizRegion, ret.country)
		if emailDisabled {
			s.logger.Info("skip GTM email because geo selector is disabled",
				"tenant_id", v.OrgId,
				"biz_region", *v.bizRegion,
				"country", ret.country,
				"geo_match_type", emailDisabledMatchType,
			)
			break
		}

		// Generate email
		emailContent, err := s.gtmComposer.ComposeEmail(ctx, lead, matchResult)
		if err != nil {
			s.logger.Error(err, "failed to compose email", "tenant_id", v.OrgId)
			break
		}

		queuedEmail, err := s.queueGTMEmail(ctx, lead, matchResult, emailContent)
		if err != nil {
			s.logger.Error(err, "failed to queue delayed email", "tenant_id", v.OrgId)
			break
		}
		if queuedEmail == nil {
			break
		}

		v.gtmSentEmail = queuedEmail
		v.gtmNotifyUserIDs = s.conf.GTM.NotifyUserIDs

		s.logger.Info("GTM email queued successfully",
			"tenant_id", v.OrgId,
			"scheduled_at", v.gtmSentEmail.ScheduledAt,
			"use_case", matchResult.UseCaseProfile)

	case *startupClusterUBEvent:
		resp, err := s.coresvcCli.Mgmt.Cluster.CountClusters(ctx, &mgmtv1.CountClustersRequest{
			Filter: &mgmtv1.ClusterFilter{
				TenantIds:   []uint64{v.TenantID},
				ClusterType: pointer.Pointer(mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS),
				ServicePlan: tidbcloudv1beta1.ServicePlan_Essential,
				// currently, send event even if customer deleted all clusters then create a new one to debug, may change in the future
				Deleted: pointer.Pointer(false),
			},
		})
		if err != nil {
			return ret, fmt.Errorf("call coresvccli.CountClusters() failed: %w", err)
		}
		if resp.GetCount() != 1 {
			// skip if not the first essential serverless cluster
			// todo: better way to identify first essential cluster
			ret.ForceSkip = true
			break
		}
		clusterResp, err := s.coresvcCli.Mgmt.GetCluster(ctx, v.ClusterID)
		if err != nil {
			return ret, fmt.Errorf("call coresvccli.GetCluster() failed: %w", err)
		}
		if clusterResp.GetDeployType() != mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS ||
			clusterResp.GetServicePlan() != tidbcloudv1beta1.ServicePlan_Essential {
			s.logger.Info("skip non-essential serverless cluster startupClusterUBEvent", "cluster_id", v.ClusterID,
				"deploy_type", clusterResp.GetDeployType(), "service_plan", clusterResp.GetServicePlan())
			ret.ForceSkip = true
			break
		}

		org, err := s.accountCli.GetTenant(ctx, v.TenantID)
		if err != nil {
			return ret, fmt.Errorf("call accountSvc.GetTenant() failed: %w, tenant_id: %d", err, v.TenantID)
		}
		v.emailDomain = getEmailDomain(org.Email)
		v.tenantName = &org.Name
		ret.country = org.GetCountry()
		bizRegion := string(region.CountryToRegion(strings.ToUpper(ret.country)))
		v.bizRegion = &bizRegion

		if tntent.TenantKind(org.Kind) == tntent.TenantKindInternal {
			s.logger.Info("skip internal tenant startupClusterUBEvent", "orgId", v.TenantID, "kind", org.Kind)
			ret.isInternalTenant = true
			ret.ForceSkip = true
			break
		}

		ret.bizRegion = *v.bizRegion
		ret.tenantID = strconv.FormatUint(v.TenantID, 10)
		// HACK: avoid being filtered out.
		ret.isSharedCluster = false
		ret.isPaidTenant = true

	case *startupImportUBEvent:
		s.logger.Info("preprocess startupImportUBEvent", "event", v)
		org, err := s.accountCli.GetTenant(ctx, v.tenantID)
		if err != nil {
			return ret, fmt.Errorf("call accountSvc.GetTenant() failed: %w, tenant_id: %d", err, v.tenantID)
		}
		v.emailDomain = getEmailDomain(org.Email)
		v.tenantName = &org.Name
		ret.country = org.GetCountry()
		bizRegion := string(region.CountryToRegion(strings.ToUpper(ret.country)))
		v.bizRegion = &bizRegion

		if tntent.TenantKind(org.Kind) == tntent.TenantKindInternal {
			s.logger.Info("skip internal tenant startupImportUBEvent", "orgId", v.tenantID, "kind", org.Kind)
			ret.isInternalTenant = true
			ret.ForceSkip = true
			break
		}
		if isPersonalEmail(v.emailDomain) && org.Plan != "on_demand" && org.Plan != "poc" {
			s.logger.Info("skip startupImportUBEvent with free tenant and personal email", "orgId", v.tenantID, "emailDomain", v.emailDomain)
			ret.ForceSkip = true
			break
		}

		if detailStr, ok := v.originEv.Detail.(string); ok && strings.Contains(detailStr, "s3://tidbcloud-sample") {
			s.logger.Info("skip startupImportUBEvent with sample data import", "orgId", v.tenantID, "detail", detailStr)
			ret.ForceSkip = true
			break
		}

		ret.bizRegion = *v.bizRegion
		ret.tenantID = strconv.FormatUint(v.tenantID, 10)
		// HACK: avoid being filtered out.
		ret.isSharedCluster = false
		ret.isPaidTenant = true

	case *clusterUBEvent:
		tenant, err := s.tenantSvc.GetTenant(ctx, v.tenantID)
		if err != nil {
			return ret, fmt.Errorf("call tenantSvc.GetTenant() failed: %w", err)
		}
		if tenant.SalesForceRegion == tntent.TenantRegionUnknown {
			bizRegion := string(tenant.CountryRegion)
			v.bizRegion = &bizRegion
		} else {
			bizRegion := string(tenant.SalesForceRegion)
			v.bizRegion = &bizRegion
		}
		ret.country = tenant.Country
		v.tenantName = &tenant.Name
		v.tenantAlias = &tenant.Alias

		cluster, err := s.coresvcCli.Mgmt.GetCluster(context.Background(), v.clusterID)
		if err != nil {
			return ret, fmt.Errorf("get management cluster failed: %w", err)
		}
		v.clusterName = &cluster.ClusterName

		// eventInfo
		if tenant.Kind == tntent.TenantKindInternal || tenant.SalesForceKind == tntent.TenantKindInternal {
			ret.isInternalTenant = true
		}
		if tenant.Plan == tntent.TenantPlanPoc || tenant.Plan == tntent.TenantPlanOnDemand {
			ret.isPaidTenant = true
		}
		ret.bizRegion = *v.bizRegion
		ret.isSharedCluster = cluster.DeployType == mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS
		ret.tenantID = strconv.FormatUint(tenant.Id, 10)

		// deal with Extra Info, may extract to a separated function in the future.
		if v.originEv.EventType == mgmtevent.EventTypeScaling && v.originEv.ExtraInfo != nil {
			topoBefore := v.originEv.ExtraInfo.TopologyBefore
			topoAfter := v.originEv.ExtraInfo.TopologyAfter
			if topoBefore != nil && topoAfter != nil {
				v.extraInfo = generateBeforeAfterTopologyText(topoBefore, topoAfter)
			}
		} else if v.originEv.EventType == mgmtevent.EventTypeCreated || v.originEv.EventType == mgmtevent.EventTypeDeleted {
			var extraInfo string
			// generate topology info
			if cluster.DeployType == mgmtv1.DeployType_DEPLOY_TYPE_DEDICATED {
				topologyString := s.generateTopologyFromCluster(cluster)
				extraInfo += topologyString
			}
			countResp, err := s.coresvcCli.Mgmt.Cluster.CountClusters(ctx, &mgmtv1.CountClustersRequest{
				Filter: &mgmtv1.ClusterFilter{
					TenantIds:   []uint64{v.tenantID},
					ClusterType: pointer.Pointer(mgmtv1.DeployType_DEPLOY_TYPE_DEDICATED),
					Deleted:     pointer.Pointer(false),
				},
			})
			if err != nil {
				//logger.V(6).Error(err, "cannot get clusters from coresvc", "tenant_id", v.tenantID)
				return ret, fmt.Errorf("cannot get clusters from coresvc: %w, tenant_id: %d", err, v.tenantID)
			} else {
				extraInfo += fmt.Sprintf("The tenant now has %d active dedicated cluster(s).", countResp.GetCount())
			}

			v.extraInfo = extraInfo
		}

	case *backupUBEvent:
		tenant, err := s.tenantSvc.GetTenant(ctx, v.tenantID)
		if err != nil {
			return ret, fmt.Errorf("call tenantSvc.GetTenant() failed: %w", err)
		}
		if tenant.SalesForceRegion == tntent.TenantRegionUnknown {
			bizRegion := string(tenant.CountryRegion)
			v.bizRegion = &bizRegion
		} else {
			bizRegion := string(tenant.SalesForceRegion)
			v.bizRegion = &bizRegion
		}
		ret.country = tenant.Country
		v.tenantName = &tenant.Name
		v.tenantAlias = &tenant.Alias

		cluster, err := s.coresvcCli.Mgmt.GetCluster(ctx, v.clusterID)
		if err != nil {
			//logger.V(6).Error(err, "cannot get cluster from coresvc", "cluster_id", v.clusterID)
			return ret, fmt.Errorf("cannot get cluster from coresvc: %w, cluster_id: %d", err, v.clusterID)
		} else {
			v.clusterName = &cluster.ClusterName
		}

		// eventInfo
		if tenant.Kind == tntent.TenantKindInternal || tenant.SalesForceKind == tntent.TenantKindInternal {
			ret.isInternalTenant = true
		}
		if tenant.Plan == tntent.TenantPlanPoc || tenant.Plan == tntent.TenantPlanOnDemand {
			ret.isPaidTenant = true
		}
		ret.bizRegion = *v.bizRegion
		ret.isSharedCluster = cluster.GetDeployType() == mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS
		ret.tenantID = strconv.FormatUint(tenant.Id, 10)

	case *restoreUBEvent:
		tenant, err := s.tenantSvc.GetTenant(ctx, v.tenantID)
		if err != nil {
			return ret, fmt.Errorf("call tenantSvc.GetTenant() failed: %w", err)
		}
		if tenant.SalesForceRegion == tntent.TenantRegionUnknown {
			bizRegion := string(tenant.CountryRegion)
			v.bizRegion = &bizRegion
		} else {
			bizRegion := string(tenant.SalesForceRegion)
			v.bizRegion = &bizRegion
		}
		ret.country = tenant.Country
		v.tenantName = &tenant.Name
		v.tenantAlias = &tenant.Alias

		cluster, err := s.coresvcCli.Mgmt.GetCluster(ctx, v.clusterID)
		if err != nil {
			//logger.V(6).Error(err, "cannot get cluster from coresvc", "cluster_id", v.clusterID)
			return ret, fmt.Errorf("cannot get cluster from coresvc: %w, cluster_id: %d", err, v.clusterID)
		} else {
			v.clusterName = &cluster.ClusterName
		}

		// eventInfo
		if tenant.Kind == tntent.TenantKindInternal || tenant.SalesForceKind == tntent.TenantKindInternal {
			ret.isInternalTenant = true
		}
		if tenant.Plan == tntent.TenantPlanPoc || tenant.Plan == tntent.TenantPlanOnDemand {
			ret.isPaidTenant = true
		}
		ret.bizRegion = *v.bizRegion
		ret.isSharedCluster = cluster.GetDeployType() == mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS
		ret.tenantID = strconv.FormatUint(tenant.Id, 10)

	case *importUBEvent:
		tenant, err := s.tenantSvc.GetTenant(ctx, v.tenantID)
		if err != nil {
			return ret, fmt.Errorf("call tenantSvc.GetTenant() failed: %w", err)
		}
		if tenant.SalesForceRegion == tntent.TenantRegionUnknown {
			bizRegion := string(tenant.CountryRegion)
			v.bizRegion = &bizRegion
		} else {
			bizRegion := string(tenant.SalesForceRegion)
			v.bizRegion = &bizRegion
		}
		ret.country = tenant.Country
		v.tenantName = &tenant.Name
		v.tenantAlias = &tenant.Alias

		cluster, err := s.coresvcCli.Mgmt.GetCluster(ctx, v.clusterID)
		if err != nil {
			// logger.V(6).Error(err, "cannot get cluster from coresvc", "cluster_id", v.clusterID)
			return ret, fmt.Errorf("cannot get cluster from coresvc: %w, cluster_id: %d", err, v.clusterID)
		} else {
			v.clusterName = &cluster.ClusterName
		}

		// eventInfo
		if tenant.Kind == tntent.TenantKindInternal || tenant.SalesForceKind == tntent.TenantKindInternal {
			ret.isInternalTenant = true
		}
		if tenant.Plan == tntent.TenantPlanPoc || tenant.Plan == tntent.TenantPlanOnDemand {
			ret.isPaidTenant = true
		}
		ret.bizRegion = *v.bizRegion
		ret.isSharedCluster = cluster.GetDeployType() == mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS
		ret.tenantID = strconv.FormatUint(tenant.Id, 10)

	case *cdcUBEvent:
		tenant, err := s.tenantSvc.GetTenant(ctx, v.tenantID)
		if err != nil {
			return ret, fmt.Errorf("call tenantSvc.GetTenant() failed: %w", err)
		}
		if tenant.SalesForceRegion == tntent.TenantRegionUnknown {
			bizRegion := string(tenant.CountryRegion)
			v.bizRegion = &bizRegion
		} else {
			bizRegion := string(tenant.SalesForceRegion)
			v.bizRegion = &bizRegion
		}
		ret.country = tenant.Country
		v.tenantName = &tenant.Name
		v.tenantAlias = &tenant.Alias

		cluster, err := s.coresvcCli.Mgmt.GetCluster(ctx, v.clusterID)
		if err != nil {
			// logger.V(6).Error(err, "cannot get cluster from coresvc", "cluster_id", v.clusterID)
			return ret, fmt.Errorf("cannot get cluster from coresvc: %w, cluster_id: %d", err, v.clusterID)
		} else {
			v.clusterName = &cluster.ClusterName
		}

		// eventInfo
		if tenant.Kind == tntent.TenantKindInternal || tenant.SalesForceKind == tntent.TenantKindInternal {
			ret.isInternalTenant = true
		}
		if tenant.Plan == tntent.TenantPlanPoc || tenant.Plan == tntent.TenantPlanOnDemand {
			ret.isPaidTenant = true
		}
		ret.bizRegion = *v.bizRegion
		ret.isSharedCluster = cluster.GetDeployType() == mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS
		ret.tenantID = strconv.FormatUint(tenant.Id, 10)
	case *dmJobUBEvent:
		tenant, err := s.tenantSvc.GetTenant(ctx, v.tenantID)
		if err != nil {
			return ret, fmt.Errorf("call tenantSvc.GetTenant() failed: %w", err)
		}
		if tenant.SalesForceRegion == tntent.TenantRegionUnknown {
			bizRegion := string(tenant.CountryRegion)
			v.bizRegion = &bizRegion
		} else {
			bizRegion := string(tenant.SalesForceRegion)
			v.bizRegion = &bizRegion
		}
		ret.country = tenant.Country
		v.tenantName = &tenant.Name
		v.tenantAlias = &tenant.Alias

		cluster, err := s.coresvcCli.Mgmt.GetCluster(ctx, v.clusterID)
		if err != nil {
			// logger.V(6).Error(err, "cannot get cluster from coresvc", "cluster_id", v.clusterID)
			return ret, fmt.Errorf("cannot get cluster from coresvc: %w, cluster_id: %d", err, v.clusterID)
		} else {
			v.clusterName = &cluster.ClusterName
		}

		// eventInfo
		if tenant.Kind == tntent.TenantKindInternal || tenant.SalesForceKind == tntent.TenantKindInternal {
			ret.isInternalTenant = true
		}
		if tenant.Plan == tntent.TenantPlanPoc || tenant.Plan == tntent.TenantPlanOnDemand {
			ret.isPaidTenant = true
		}
		ret.bizRegion = *v.bizRegion
		ret.isSharedCluster = cluster.GetDeployType() == mgmtv1.DeployType_DEPLOY_TYPE_SERVERLESS
		ret.tenantID = strconv.FormatUint(tenant.Id, 10)
	default:
		return ret, fmt.Errorf("not a registered ub event")
	}

	return ret, nil
}

func (s *svc) queueGTMEmail(
	ctx context.Context,
	lead *ubent.GTMLead,
	matchResult gtm.MatchResult,
	emailContent *gtm.EmailContent,
) (*ubent.GTMSentEmail, error) {
	suppressed, err := s.repo.IsGTMEmailSuppressed(ctx, lead.TenantID, GTMCampaignQualifiedSignupFollowup)
	if err != nil {
		return nil, err
	}
	if suppressed {
		s.logger.Info("skip GTM email queue because campaign is suppressed",
			"tenant_id", lead.TenantID,
			"campaign_name", GTMCampaignQualifiedSignupFollowup,
		)
		return nil, nil
	}

	nowFn := s.nowFn
	if nowFn == nil {
		nowFn = time.Now
	}
	randIntnFn := s.randIntnFn
	if randIntnFn == nil {
		randIntnFn = rand.Intn
	}

	scheduledAt, err := computeGTMEmailScheduledAt(nowFn(), s.conf.GTM.EmailDelay, randIntnFn)
	if err != nil {
		return nil, err
	}

	queuedEmail := &ubent.GTMSentEmail{
		LeadID:             lead.ID,
		TenantID:           lead.TenantID,
		ToAddress:          lead.Email,
		Subject:            emailContent.Subject,
		Body:               emailContent.HTMLBody,
		CCAddresses:        pointer.Pointer(strings.Join(s.conf.GTM.InternalCCList, ",")),
		UseCaseProfile:     matchResult.UseCaseProfile,
		CustomerReferences: pointer.Pointer(strings.Join(matchResult.CustomerReferences, ",")),
		ScheduledAt:        scheduledAt.UTC(),
		Status:             "queued",
		AttemptCount:       0,
	}

	if err := s.repo.CreateGTMSentEmail(ctx, queuedEmail); err != nil {
		return nil, err
	}

	return queuedEmail, nil
}

func (s *svc) persistGTMLeadMatch(ctx context.Context, lead *ubent.GTMLead, matchResult gtm.MatchResult) error {
	if err := s.repo.UpdateGTMLeadUseCaseMatch(ctx, lead.ID, matchResult.UseCaseProfile, matchResult.Reason); err != nil {
		return err
	}

	lead.MatchedUseCase = &matchResult.UseCaseProfile
	lead.UseCaseMatchReason = &matchResult.Reason
	return nil
}

var gtmEmailScheduleLoc = time.FixedZone("UTC-7", -7*60*60)

func gtmEmailScheduleLocation() *time.Location {
	return gtmEmailScheduleLoc
}

func computeGTMEmailScheduledAt(base time.Time, delay time.Duration, randIntnFn func(int) int) (time.Time, error) {
	computed := base.Add(delay).In(gtmEmailScheduleLocation())
	if isWithinLABusinessHours(computed) {
		return computed, nil
	}

	jitterMinutes := 0
	if randIntnFn != nil {
		jitterMinutes = randIntnFn(60)
		if jitterMinutes < 0 {
			jitterMinutes = 0
		}
		if jitterMinutes >= 60 {
			jitterMinutes = 59
		}
	}

	nextBusinessMorning := nextLABusinessMorning(computed)
	if isWeekday(computed) && computed.Before(startOfLABusinessDay(computed)) {
		nextBusinessMorning = startOfLABusinessDay(computed)
	}
	return nextBusinessMorning.Add(time.Duration(jitterMinutes) * time.Minute), nil
}

func isWithinLABusinessHours(t time.Time) bool {
	if !isWeekday(t) {
		return false
	}

	start := startOfLABusinessDay(t)
	end := endOfLABusinessDay(t)
	return !t.Before(start) && !t.After(end)
}

func nextLABusinessMorning(t time.Time) time.Time {
	nextDay := time.Date(t.Year(), t.Month(), t.Day(), 9, 0, 0, 0, t.Location()).AddDate(0, 0, 1)
	for !isWeekday(nextDay) {
		nextDay = nextDay.AddDate(0, 0, 1)
	}
	return nextDay
}

func isWeekday(t time.Time) bool {
	return t.Weekday() != time.Saturday && t.Weekday() != time.Sunday
}

func startOfLABusinessDay(t time.Time) time.Time {
	return time.Date(t.Year(), t.Month(), t.Day(), 9, 0, 0, 0, t.Location())
}

func endOfLABusinessDay(t time.Time) time.Time {
	return time.Date(t.Year(), t.Month(), t.Day(), 17, 0, 0, 0, t.Location())
}

func (s *svc) ProcessQueuedGTMEmails(ctx context.Context) error {
	if !s.conf.GTM.Enabled {
		return nil
	}

	logger := log.FromContextOrDef(ctx, s.logger)
	now := time.Now().UTC()

	queued, err := s.repo.ListDueGTMSentEmails(ctx, now, 100)
	if err != nil {
		return err
	}

	for _, email := range queued {
		locked, err := s.repo.MarkGTMSentEmailSending(ctx, email.ID)
		if err != nil {
			logger.Error(err, "failed to mark queued email as sending", "id", email.ID)
			continue
		}
		if !locked {
			continue
		}

		to := email.ToAddress
		if mockTo := strings.TrimSpace(s.conf.GTM.MockToAddress); mockTo != "" {
			to = mockTo
		}

		attemptedAt := time.Now().UTC()
		resp, err := s.sendGridCli.SendPlainEmail(ctx, sendgridcli.SendPlainEmailRequest{
			From:     s.conf.GTM.FromAddress,
			FromName: s.conf.GTM.FromName,
			To:       to,
			CC:       splitCommaList(email.CCAddresses),
			Subject:  email.Subject,
			HTMLBody: email.Body,
		})
		if err != nil {
			logger.Error(err, "failed to send queued email", "id", email.ID)
			_ = s.repo.MarkGTMSentEmailFailed(ctx, email.ID, attemptedAt, err.Error())
			continue
		}

		if err := s.repo.MarkGTMSentEmailSent(ctx, email.ID, attemptedAt, resp.MessageID); err != nil {
			logger.Error(err, "failed to mark queued email as sent", "id", email.ID)
			continue
		}
	}

	return nil
}

func splitCommaList(s *string) []string {
	if s == nil {
		return nil
	}
	raw := strings.TrimSpace(*s)
	if raw == "" {
		return nil
	}

	parts := strings.Split(raw, ",")
	out := make([]string, 0, len(parts))
	for _, p := range parts {
		p = strings.TrimSpace(p)
		if p != "" {
			out = append(out, p)
		}
	}
	return out
}