package opworker

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/account/commons/eventbus"

	"github.com/tidbcloud/account/internel/svc/org"
	"github.com/tidbcloud/account/internel/svc/org/handler"

	"github.com/tidbcloud/account/thirdparty/billing2"

	"github.com/google/uuid"
	"github.com/pingcap/errors"
	"github.com/spf13/viper"
	"github.com/tidbcloud/account/central/utils"
	"github.com/tidbcloud/account/commons/config"
	"github.com/tidbcloud/account/commons/database"
	accountpb "github.com/tidbcloud/account/idl/pbgen/proto/account"
	"github.com/tidbcloud/account/internel/dao/entity"
	"github.com/tidbcloud/account/internel/dao/project"
	"github.com/tidbcloud/account/thirdparty/account"
	"github.com/tidbcloud/account/thirdparty/infra"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/worker"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

type GlobalWorker struct {
	logger  *zap.Logger
	workers []*worker.PeriodWorker
	manager *Manager
}

func (w *GlobalWorker) Start() {
	for _, item := range w.workers {
		item.Start()
	}
}

func NewGlobalBgWorker(manager *Manager) (*GlobalWorker, error) {
	gWorker := &GlobalWorker{
		logger:  tilog.With(zap.String("worker", "global")),
		manager: manager,
	}

	cleanUpTenantWorker := CleanupTenantWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "cleanUpTenant")),
			manager: manager,
		},
	}

	cleanUpProjectWorker := CleanupProjectWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "cleanUpProject")),
			manager: manager,
		},
	}

	PocApplicationWorker := PocApplicationWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "pocApplication")),
			manager: manager,
		},
	}

	syncTenantWorker := syncTenantWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "syncTenantWorker")),
			manager: manager,
		},
	}

	SQSMonitorWorker := SQSMonitorWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "SQSMonitor")),
			manager: manager,
		},
	}
	DBUserWorker := DBUserWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "DBUserWorker")),
			manager: manager,
		},
	}
	DBUserCheckWorker := DBUserCheckWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "DBUserCheckWorker")),
			manager: manager,
		},
	}

	DynamoDBDataValidationWorker := DynamoDBDataValidationWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "DynamoDBDataValidationWorker")),
			manager: manager,
		},
	}
	MspCustomerConvertWorker := MspCustomerConvertWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "MspCustomerConvertWorker")),
			manager: manager,
		},
	}

	roleBindingMonitorWorker := RoleBindingMonitorWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "RoleBindingMonitorWorker")),
			manager: manager,
		},
	}
	mspClearUserServicesAgreementWorker := MspClearUserServicesAgreementWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "MspClearUserServicesAgreementWorker")),
			manager: manager,
		},
	}

	//clusterRelationMonitorWorker := ClusterRelationMonitorWorker{
	//	GlobalWorker: GlobalWorker{
	//		logger:  tilog.With(zap.String("worker", "ClusterRelationMonitorWorker")),
	//		manager: manager,
	//	},
	//}
	//clusterRelationSyncWorker := ClusterRelationSyncWorker{
	//	GlobalWorker: GlobalWorker{
	//		logger:  tilog.With(zap.String("worker", "ClusterRelationSyncWorker")),
	//		manager: manager,
	//	},
	//}

	cleanupUserSessionWorker := CleanupUserSessionWorker{
		GlobalWorker: GlobalWorker{
			logger:  tilog.With(zap.String("worker", "CleanupUserSessionWorker")),
			manager: manager,
		},
	}

	accountAPI := account.GetHandler()
	DynamoDBDataValidationWorkerInterval := time.Minute * 30
	if config.IsDevEnv() || config.IsStagingEnv() {
		DynamoDBDataValidationWorkerInterval = time.Hour * 72
	}

	if !config.IsLocalEnv() {

		gWorker.workers = []*worker.PeriodWorker{
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "cleanUpTenant")), time.Minute*30, withMetrics(0, "cleanUpTenant", cleanUpTenantWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "cleanUpProject")), time.Minute*1, withMetrics(0, "cleanUpProject", cleanUpProjectWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "pocApplication")), time.Hour*1, withMetrics(0, "pocApplication", PocApplicationWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "syncTenantWorker")), time.Hour*1, withMetrics(0, "syncTenantWorker", syncTenantWorker.Run)),

			worker.NewPeriodWorker(tilog.With(zap.String("worker", "SQSMonitor")), time.Minute*3, withMetrics(0, "SQSMonitor", SQSMonitorWorker.Run)),
			//worker.NewPeriodWorkerWithDelay(tilog.With(zap.String("worker", "checkTenant")), time.Minute*1, time.Second*10, withMetrics(0, "checkTenant", TenantMonitorWorker{GlobalWorker: *gWorker}.Run)),
			worker.NewPeriodWorkerWithDelay(tilog.With(zap.String("worker", "marketplaceSubscribeSync")), time.Second*30, time.Second*30,
				withMetrics(0, "marketplaceSubscribeSync", MarketplaceWorker{GlobalWorker: *gWorker, accountAPI: accountAPI}.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "DBUserWorker")), time.Second*5, withMetrics(0, "DBUser", DBUserWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "DBUserCheckWorker")), time.Minute*5, withMetrics(0, "DBUserCheckWorker", DBUserCheckWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "DynamoDBDataValidationWorker")), DynamoDBDataValidationWorkerInterval, withMetrics(0, "DynamoDBDataValidationWorker", DynamoDBDataValidationWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "MspCustomerConvertWorker")), time.Second*30, withMetrics(0, "MspCustomerConvertWorker", MspCustomerConvertWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "MspClearUserServicesAgreementWorker")), time.Minute*1, withMetrics(0, "MspClearUserServicesAgreementWorker", mspClearUserServicesAgreementWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "RoleBindingMonitorWorker")), time.Hour*10, withMetrics(0, "RoleBindingMonitorWorker", roleBindingMonitorWorker.Run)),
			//worker.NewPeriodWorker(tilog.With(zap.String("worker", "ClusterRelationMonitorWorker")), time.Hour*12, withMetrics(0, "ClusterRelationMonitorWorker", clusterRelationMonitorWorker.Run)),
			//worker.NewPeriodWorker(tilog.With(zap.String("worker", "ClusterRelationSyncWorker")), time.Hour*24, withMetrics(0, "ClusterRelationSyncWorker", clusterRelationSyncWorker.Run)),
			worker.NewPeriodWorker(tilog.With(zap.String("worker", "CleanupUserSessionWorker")), time.Hour*24, withMetrics(0, "CleanupUserSessionWorker", cleanupUserSessionWorker.Run)),
		}
	}

	// local test
	if config.IsLocalEnv() {

		gWorker.workers = []*worker.PeriodWorker{

			worker.NewPeriodWorker(tilog.With(zap.String("worker", "MspCustomerConvertWorker")), time.Minute*10, withMetrics(0, "MspCustomerConvertWorker", MspCustomerConvertWorker.Run)),
			//worker.NewPeriodWorker(tilog.With(zap.String("worker", "MspClearUserServicesAgreementWorker")), time.Minute*2, withMetrics(0, "MspClearUserServicesAgreementWorker", mspClearUserServicesAgreementWorker.Run)),
		}
	}

	return gWorker, nil
}

// TenantMonitorWorker check the tenant status, then start or stop the tenant background worker
type TenantMonitorWorker struct{ GlobalWorker }

func (w TenantMonitorWorker) Run(ctx context.Context) error {
	ctx, span := otel.Tracer(config.TraceComponentNameWorker).Start(ctx, "TenantMonitorWorker")
	defer span.End()
	tilog.Info(ctx, "start to check the worker status")
	var tenants []entity.Tenant
	err := database.Transaction(func(tx *gorm.DB) error {
		return entity.NewTenantQuerySet(tx).StatusNe(entity.TenantStatusCleared).All(&tenants)
	})
	if err != nil {
		return err
	}
	var errs error
	for _, t := range tenants {
		bgw := w.manager.GetBackgroundWorker(t.ID)
		if t.Status == entity.TenantStatusActive {
			if bgw == nil {
				errs = multierr.Append(errs, w.manager.StartBackgroundWorkerForTenant(t.ID))
			}
		} else {
			if bgw != nil {
				errs = multierr.Append(errs, w.manager.StopBackgroundWorkerForTenant(t.ID))
			}
		}
	}
	return errs
}

type CleanupProjectWorker struct{ GlobalWorker }

func (w CleanupProjectWorker) Run(ctx context.Context) error {
	ctx, span := otel.Tracer(config.TraceComponentNameWorker).Start(ctx, "CleanupProjectWorker")
	defer span.End()
	logger := tilog.AppendCtxFields(ctx, w.logger)
	logger.Info("start to cleanup deleted project")

	var projects []entity.Project
	err := database.Transaction(func(tx *gorm.DB) error {
		return entity.NewProjectQuerySet(tx).
			StatusEq(entity.ProjectStatusDeleted).
			All(&projects)
	})
	if err != nil {
		return errors.Annotate(err, "failed to query deleted projects")
	}

	var errs error
	for _, p := range projects {
		logger.Info("start to cleanup project resources", zap.Uint64("projectID", p.ID.Uint64()))

		clusters, err := infra.GetHandler().GetProjectClustersNotInStatus(ctx, p.ID.Uint64(), core.ClusterStatus_CLUSTER_STATUS_DELETECOMPLETE)
		if err != nil {
			tilog.Error(ctx, "failed to query the list of non-cleared clusters under the project", zap.Uint64("projectID", p.ID.Uint64()), zap.Error(err))
			errs = multierr.Append(errs, errors.Annotatef(err, "failed to query the list of non-cleared clusters under the project %d", p.ID))
			continue
		}

		if len(clusters) > 0 {
			clustersIDList := make([]string, 0, len(clusters))
			for _, c := range clusters {
				clustersIDList = append(clustersIDList, fmt.Sprintf("%d", c.Id))
			}

			clusterIDs := strings.Join(clustersIDList, ", ")
			logger.Info("Before deleting the project, please confirm that the following clusters and their backups in the recycle bin have been deleted.", zap.Uint64("projectID", p.ID.Uint64()), zap.String("clustersIDs", clusterIDs))
			continue
		}

		err = infra.GetHandler().CleanProjectResource(ctx, p.ID.Uint64())
		if err != nil {
			errs = multierr.Append(errs, errors.Annotatef(err, "failed to delete project %d resource", p.ID))
			continue
		}

		err = database.Transaction(func(tx *gorm.DB) error {
			return project.ClearProject(ctx, tx, p.ID)
		})
		if err != nil {
			errs = multierr.Append(errs, err)
			continue
		}
	}
	return errs
}

type CleanupTenantWorker struct{ GlobalWorker }

func (w CleanupTenantWorker) Run(ctx context.Context) error {
	ctx, span := otel.Tracer(config.TraceComponentNameWorker).Start(ctx, "CleanupTenantWorker")
	defer span.End()
	tilog.Info(ctx, "start to cleanup deleted resources")
	var tenants []entity.Tenant
	err := database.Transaction(func(tx *gorm.DB) error {
		return entity.NewTenantQuerySet(tx).
			StatusEq(entity.TenantStatusDeleted).
			All(&tenants)
	})
	if err != nil {
		return err
	}
	var errs error
	for _, t := range tenants {
		tilog.Info(ctx, "start to cleanup tenant resources", zap.Uint64("tenantID", t.ID.Uint64()))
		err := infra.GetHandler().CleanTenantResource(ctx, t.ID.Uint64())
		if err != nil {
			errs = multierr.Append(errs, err)
			continue
		}
		err = database.Transaction(func(tx *gorm.DB) error {
			return org.UpdateTenantStatus(tx, t.ID, entity.TenantStatusCleared)
		})
		if err != nil {
			errs = multierr.Append(errs, errors.Annotatef(err, "update tenant %d status failed", t.ID))
			continue
		}
		tilog.Info(ctx, "tenant status changed from deleted to cleared", zap.Uint64("tenantID", t.ID.Uint64()))
		//sentry.CaptureMessage(fmt.Sprintf("tenant %d status changed from deleted to cleared", t.ID.Uint64()))
	}
	return errs
}

type MarketplaceWorker struct {
	GlobalWorker
	accountAPI account.AccountAPIHandler
}

const (
	ThirdPartyNormalSubscribed   = "tenant-third-party-normal-subscribed"
	ThirdPartyNormalUnsubscribed = "tenant-third-party-normal-unsubscribed"
)

type SubscribeEvent struct {
	TenantID uint64 `json:"tenant_id"`
	Action   string `json:"action"`
}

const (
	SubscribeAction   = "subscribe"
	UnsubscribeAction = "unsubscribe"
)

func (w MarketplaceWorker) getTenantByAccountCode(ctx context.Context, tx *gorm.DB, provider string, code string) (*entity.Tenant, error) {
	var tenants []entity.Tenant
	err := entity.NewTenantQuerySet(tx).ThirdPartyAccountProviderEq(provider).ThirdPartyAccountCodeEq(code).StatusNotIn(entity.TenantStatusCleared, entity.TenantStatusDeleted).All(&tenants)
	if err != nil {
		return nil, err
	}
	if len(tenants) == 0 {
		tilog.Info(ctx, fmt.Sprintf("cant' find tenant by accountCode. provider=%s,accountCode=%s", provider, code))
		return nil, err
	}
	return &tenants[0], nil
}

func (w MarketplaceWorker) processSubscribed(ctx context.Context, input handler.SubscribeChange) error {
	// DB + event publish must succeed; notifications are best-effort so a
	// transient notify failure does not roll back the subscribe state and
	// re-deliver the SQS message.
	var (
		tenant        *entity.Tenant
		needNotifyOps bool
		notifyReason  string
	)
	err := database.Transaction(func(tx *gorm.DB) error {
		t, err := w.getTenantByAccountCode(ctx, tx, input.AccountProvider, input.AccountCode)
		if err != nil {
			return err
		}
		if t == nil {
			return nil
		}
		tenant = t
		if input.FreeTrial {
			if !tenant.ThirdPartyFreeTrial {
				// oddity: existing paid tenant reverting to free trial — alert only
				needNotifyOps = true
				notifyReason = "Fatal: can not change non free trial to free trial subscribe"
			}
			return nil
		}
		if err := entity.NewTenantQuerySet(tx).IDEq(tenant.ID).GetUpdater().SetThirdPartyFreeTrial(input.FreeTrial).Update(); err != nil {
			return err
		}
		return eventbus.Get().PublishEventWithTransaction(ThirdPartyNormalSubscribed, &eventbus.Event{
			EventID:   uuid.NewString(),
			EventType: ThirdPartyNormalSubscribed,
			EventData: SubscribeEvent{
				TenantID: tenant.ID.Uint64(),
				Action:   SubscribeAction,
			},
		}, tx)
	})
	if err != nil {
		return err
	}
	if tenant == nil || !needNotifyOps {
		return nil
	}
	if err := w.notifyOpsAndRD(ctx, tenant, input, notifyReason); err != nil {
		w.logger.Error("notify ops and rd failed on subscribed", zap.Uint64("tenantID", tenant.ID.Uint64()), zap.Error(err))
	}
	return nil
}

var Provider2ProviderNameMap = map[string]string{
	handler.AWSAccountProvider:      "AWS",
	handler.GCPAccountProvider:      "GCP",
	handler.AliCloudAccountProvider: "Alibaba Cloud",
	handler.AzureAccountProvider:    "Azure",
}

func (w MarketplaceWorker) processUnsubscribed(ctx context.Context, input handler.SubscribeChange) error {
	// Phase 1: DB + event publish must succeed. Notifications are best-effort
	// and must NOT roll back DB state — otherwise a transient email/SendGrid
	// failure would re-deliver the SQS message and storm the customer.
	var (
		tenant    *entity.Tenant
		skip      bool
		needEmail bool
	)
	err := database.Transaction(func(tx *gorm.DB) error {
		t, err := w.getTenantByAccountCode(ctx, tx, input.AccountProvider, input.AccountCode)
		if err != nil {
			return err
		}
		if t == nil {
			return nil
		}
		tenant = t
		isSkip, err := w.isSkipUnsubscribed(ctx, tenant)
		if err != nil {
			w.logger.Error(fmt.Sprintf("User unsubscribed  failed. tenantId = %d, provider=%s, accountCode=%s", tenant.ID, input.AccountProvider, input.AccountCode), zap.Error(err))
			return err

		}
		if isSkip {
			skip = true
			w.logger.Info(fmt.Sprintf("User unsubscribed hit skip acccount lis. tenantId = %d, provider=%s, accountCode=%s", tenant.ID, input.AccountProvider, input.AccountCode))
			return nil
		}

		// log error and trigger alert
		if input.SubscribeStatus == handler.SubscribeStatusUnsubscribed {
			// user unsubscribed
			w.logger.Info(fmt.Sprintf("User unsubscribed. tenantId = %d, provider=%s, accountCode=%s", tenant.ID, input.AccountProvider, input.AccountCode))
		} else {
			// user subscribe failed on AWS Marketplace, eg: Payment method invalid
			w.logger.Error(fmt.Sprintf("User subscribe failed. tenantId = %d, provider=%s, accountCode=%s", tenant.ID, input.AccountProvider, input.AccountCode))
		}
		if !input.FreeTrial {
			err = entity.NewTenantQuerySet(tx).IDEq(tenant.ID).GetUpdater().SetThirdPartyFreeTrial(true).Update()
			if err != nil {
				return err
			}
			err = eventbus.Get().PublishEventWithTransaction(ThirdPartyNormalUnsubscribed, &eventbus.Event{
				EventID:   uuid.NewString(),
				EventType: ThirdPartyNormalUnsubscribed,
				EventData: SubscribeEvent{
					TenantID: tenant.ID.Uint64(),
					Action:   UnsubscribeAction,
				},
			}, tx)
			if err != nil {
				return err
			}
			needEmail = true
		}
		return nil
	})
	if err != nil {
		return err
	}
	if tenant == nil || skip || !needEmail {
		return nil
	}

	// Phase 2: notifications are best-effort. A failure here only logs —
	// the SQS message has already been / will be deleted by the caller based
	// on the nil error above, so the customer is not paged again.
	if err := w.notifyOpsAndRD(ctx, tenant, input, "unsubscribed"); err != nil {
		w.logger.Error("notify ops and rd failed on unsubscribed", zap.Uint64("tenantID", tenant.ID.Uint64()), zap.Error(err))
	}
	provider := Provider2ProviderNameMap[input.AccountProvider]
	if err := w.sendUnsubEmail(ctx, tenant, provider); err != nil {
		w.logger.Error("send unsub email failed", zap.Uint64("tenantID", tenant.ID.Uint64()), zap.String("provider", provider), zap.Error(err))
	}
	return nil
}

func (w MarketplaceWorker) isSkipUnsubscribed(ctx context.Context, tenant *entity.Tenant) (bool, error) {
	var cfg map[string]bool
	logger := log.WithContextL(ctx)
	parseErr := viper.UnmarshalKey("skipUnsubscribedMap", &cfg)
	if parseErr != nil {
		logger.Error("parse skipUnsubscribedMap failed.", zap.Error(parseErr))
		return false, parseErr
	}
	return cfg[strconv.FormatUint(uint64(tenant.ID), 10)], nil
}
func (w MarketplaceWorker) notifyOpsAndRD(ctx context.Context, tenant *entity.Tenant, input handler.SubscribeChange, reason string) error {
	var toEmails []string
	if config.Global().ServerConfig.BizOpsTeamEmail != "" {
		toEmails = []string{config.Global().ServerConfig.BizOpsTeamEmail}
	}
	tenant.AccountManager = utils.GetOrgAccountManager(ctx, tenant.ID.Uint64())
	if tenant.AccountManager != "" {
		toEmails = append(toEmails, tenant.AccountManager)
	}
	lastBillsKey, currentBillsKey := "lastMonthBills", "currentMonthBills"
	vars := map[string]string{
		"accountProvider": input.AccountProvider,
		"accountCode":     input.AccountCode,
		"subscribeStatus": input.SubscribeStatus,
		"freeTrial":       fmt.Sprint(input.FreeTrial),
		"reason":          reason,
		"orgID":           fmt.Sprint(tenant.ID.Uint64()),
		"orgName":         tenant.Name,
		lastBillsKey:      "Unknown",
		currentBillsKey:   "Unknown",
	}
	if input.SubscribeStatus == handler.SubscribeStatusUnsubscribed || input.SubscribeStatus == handler.SubscribeStatusSubscribeFailed {
		now := time.Now().UTC()
		for offset, key := range []string{currentBillsKey, lastBillsKey} {
			dt := time.Date(now.Year(), now.Month(), 1, 0, 0, 0, 0, time.UTC).AddDate(0, -offset, 0)
			monthBills, err := billing2.GetHandler().GetMonthlyBills(context.Background(), tenant.ID.Uint64(), dt.Year(), dt.Month())
			if err != nil {
				log.Error(ctx, "failed to get month bills on customer unsubscribed", zap.Error(err), zap.Uint64("tenantID", tenant.ID.Uint64()), zap.Time("dt", dt))
				continue
			}
			vars[key] = monthBills.InvoiceCosts.CostPrice.String()
		}
	}
	ccEmails := []string{
		"nizheming@pingcap.com",
		"yichao.zhang@pingcap.com",
	}

	_, err := w.accountAPI.SendEmail(ctx, &accountpb.SendEmailReq{
		OrgId:        tenant.ID.Uint64(),
		Scene:        string(entity.NotificationSceneCustomerStatusUpdate),
		Identifier:   uuid.New().String(),
		Subject:      config.CustomerStatusUpdateSubject,
		Emails:       toEmails,
		CcEmails:     ccEmails,
		TemplateId:   config.SendgridTemplateName2id[config.TNCustomerUpdateSubscribeStatusChanged],
		TemplateVars: vars,
	})

	if err != nil {
		return fmt.Errorf("send mail to ops failed: %w", err)
	}

	return nil
}

func (w MarketplaceWorker) sendUnsubEmail(ctx context.Context, tenant *entity.Tenant, provider string) error {
	suspendDate := time.Now().AddDate(0, 0, 3).Format("2006-01-02")
	vars := map[string]string{
		"provider":    provider,
		"orgID":       fmt.Sprint(tenant.ID.Uint64()),
		"suspendDate": suspendDate,
	}

	_, err := w.accountAPI.SendEmailByLabels(ctx, &accountpb.SendEmailByLabelsReq{
		OrgId:        tenant.ID.Uint64(),
		Scene:        string(entity.NotificationSceneCustomerStatusUpdate),
		Identifier:   uuid.New().String(),
		TemplateId:   config.SendgridTemplateName2id[config.TNMarketplaceUnsubscriptionDetected],
		TemplateVars: vars,
		ToLabel:      accountpb.MailLabels_MAIL_LABELS_ORG_OWNER,
	})

	if err != nil {
		return fmt.Errorf("send mail to customer failed: %w", err)
	}

	return nil
}

func (w MarketplaceWorker) ProcessSubscribeChange(ctx context.Context, input handler.SubscribeChange) (ret bool, err error) {
	tilog.Info(ctx, "Process subscribe change.", zap.String("accountCode", input.AccountCode), zap.String("status", input.SubscribeStatus))
	switch input.SubscribeStatus {
	case handler.SubscribeStatusSubscribed:
		err = w.processSubscribed(ctx, input)
	case handler.SubscribeStatusUnsubscribed, handler.SubscribeStatusSubscribeFailed:
		err = w.processUnsubscribed(ctx, input)
	default:
		log.Error(
			ctx, "process unknown SubscribeStatus", zap.String("SubscribeStatus", input.SubscribeStatus),
			zap.String("AccountProvider", input.AccountProvider), zap.String("AccountCode", input.AccountCode),
		)
	}
	return err == nil, err
}

func (w MarketplaceWorker) Run(ctx context.Context) error {
	ctx, span := otel.Tracer(config.TraceComponentNameWorker).Start(ctx, "MarketplaceWorker")
	defer span.End()
	if !config.Global().MarketPlaceConfig.EnableBackground {
		tilog.Info(ctx, "marketplace worker skipped by config.")
		return nil
	}
	providers, err := handler.GetAccountHandler().GetAllAccountProviders()
	if err != nil {
		return err
	}
	var mulErrs error
	for _, provider := range providers {
		mulErrs = multierr.Append(mulErrs, provider.ProcessTenantSubscribeChange(ctx, w.ProcessSubscribeChange))
	}
	return mulErrs
}