package opworker

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

	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/dbutil"
	"github.com/pingcap/dbaas/central/db/entity"
	infraapi "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/model/cluster"
	"github.com/pingcap/dbaas/central/model/tenant/handler"
	"github.com/pingcap/dbaas/central/model/util"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/pkgv2/log"
	mailv2 "github.com/tidbcloud/pkgv2/mail/v2"
	pubsub2 "github.com/tidbcloud/pkgv2/pubsub"
	"github.com/tidbcloud/pkgv2/worker"
	"go.uber.org/multierr"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

var (
	globalPublisher pubsub2.Publisher
)

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:  log.With(zap.String("worker", "global")),
		manager: manager,
	}

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

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

	infraCli := infraapi.GetInfraAPI()
	notificationMgr := util.NewNotificationMgr()
	dbProvider := db.NewProvider()
	clusterRepo := dbutil.NewClusterRepo(dbProvider)
	tenantRepo := dbutil.NewTenantRepo(dbProvider)
	userRepo := dbutil.NewUserRepo(dbProvider)
	clusterTrafficUsageRepo := dbutil.NewClusterTrafficUsageRepoRepo(dbProvider)

	// TODO(chenfei): use DI
	// TODO(chenfei): should be a global parameter
	globalPublisher = pubsub2.NewPublisher(
		cluster.NewDevTierNetworkThresholdSwitchProcessor(infraCli),
		cluster.NewDevTierNetworkThrottledNotifyProcessor(tenantRepo, userRepo, dbProvider, notificationMgr, manager.mailFacade),
	)

	gWorker.workers = []*worker.PeriodWorker{
		worker.NewPeriodWorker(log.With(zap.String("worker", "cleanUpTenant")), time.Minute*30, withMetrics(0, "cleanUpTenant", cleanUpTenantWorker.Run)),
		worker.NewPeriodWorker(log.With(zap.String("worker", "cleanUpProject")), time.Minute*1, withMetrics(0, "cleanUpProject", cleanUpProjectWorker.Run)),
		worker.NewPeriodWorkerWithDelay(log.With(zap.String("worker", "checkTenant")), time.Minute*1, time.Second*10, withMetrics(0, "checkTenant", TenantMonitorWorker{GlobalWorker: *gWorker}.Run)),
		// a worker for dev tier cluster. clean the idle cluster and warn the network threshold exceeded cluster
		worker.NewPeriodWorker(
			log.With(zap.String("worker", "dev_tier_cluster")),
			time.Minute*10,
			withMetrics(0, "dev_tier_cluster", NewDevTierClusterWorker(*gWorker, manager.mailFacade, infraCli, notificationMgr, dbProvider, tenantRepo, userRepo, clusterRepo, clusterTrafficUsageRepo, globalPublisher).Run),
		),
		worker.NewPeriodWorkerWithDelay(log.With(zap.String("worker", "marketplaceSubscribeSync")), time.Minute*1, time.Minute*1,
			withMetrics(0, "marketplaceSubscribeSync", MarketplaceWorker{GlobalWorker: *gWorker, mailFacade: manager.mailFacade}.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 {
	w.logger.Info("start to check the worker status")
	var tenants []entity.Tenant
	err := db.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 {
	logger := log.AppendCtxFields(ctx, w.logger)
	logger.Info("start to cleanup deleted project")

	var projects []entity.Project
	err := db.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()))

		var clusters []entity.Cluster
		err = db.Transaction(func(tx *gorm.DB) error {
			clusters, err = dbutil.GetProjectNotClearedClusters(tx, p.ID)
			return err
		})
		if err != nil {
			log.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 := w.manager.api.DeleteProjectNetworkPeers(context.TODO(), p.ID.Uint64())
		if err != nil {
			errs = multierr.Append(errs, errors.Annotatef(err, "failed to delete NetworkPeer of project %d", p.ID))
			continue
		}
		err = w.manager.api.DeleteProjectNetworks(context.TODO(), p.ID.Uint64())
		if err != nil {
			errs = multierr.Append(errs, errors.Annotatef(err, "failed to delete Network of project %d", p.ID))
			continue
		}

		err = db.Transaction(func(tx *gorm.DB) error {
			return dbutil.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 {
	w.logger.Info("start to cleanup deleted resources")
	var tenants []entity.Tenant
	err := db.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 {
		w.logger.Info("start to cleanup tenant resources", zap.Uint64("tenantID", t.ID.Uint64()))
		err := w.manager.api.DeleteTenantNetworks(context.TODO(), t.ID.Uint64())
		if err != nil {
			errs = multierr.Append(errs, err)
			continue
		}
		err = db.Transaction(func(tx *gorm.DB) error {
			return dbutil.UpdateTenantStatus(tx, t.ID, entity.TenantStatusCleared)
		})
		if err != nil {
			errs = multierr.Append(errs, errors.Annotatef(err, "update tenant %d status failed", t.ID))
			continue
		}
	}
	return errs
}

type MarketplaceWorker struct {
	GlobalWorker
	mailFacade mailv2.Facade
}

func (w MarketplaceWorker) ProcessSubscribeChange(ctx context.Context, input handler.SubscribeChange) (bool, error) {
	w.logger.Info("Process subscribe change.", zap.String("accountCode", input.AccountCode), zap.String("status", input.SubscribeStatus))
	if input.SubscribeStatus == handler.SubscribeStatusSubscribed { //nolint
		// Freeze the tenant.
		//nolint
	}
	if input.SubscribeStatus == handler.SubscribeStatusUnsubscribed { //nolint
		// UnFreeze the tenant.

		// just notify the bizOps and rd for now
		tenant := &entity.Tenant{}
		err := db.Transaction(func(tx *gorm.DB) error {
			err := entity.NewTenantQuerySet(tx).ThirdPartyAccountProviderEq(input.AccountProvider).ThirdPartyAccountCodeEq(input.AccountCode).One(tenant)
			if err != nil {
				return err
			}
			return nil
		})
		if err != nil {
			w.logger.Info(fmt.Sprintf("cant' find tenant by accountCode. provider=%s,accountCode=%s", input.AccountProvider, input.AccountCode))
			return false, nil
		}
		err = w.notifyOpsAndRD(ctx, tenant, input)
		return err == nil, err
	}
	return true, nil
}

func (w MarketplaceWorker) notifyOpsAndRD(ctx context.Context, tenant *entity.Tenant, input handler.SubscribeChange) error {
	var tos []string
	if config.Global().ServerConfig.BizOpsTeamEmail != "" {
		tos = append(tos, config.Global().ServerConfig.BizOpsTeamEmail)
	}
	if tenant.AccountManager != "" {
		tos = append(tos, tenant.AccountManager)
	}
	vars := map[string]string{
		"accountProvider": input.AccountProvider,
		"accountCode":     input.AccountCode,
		"orgID":           fmt.Sprint(tenant.ID.Uint64()),
		"orgName":         tenant.Alias,
	}
	opts := mailv2.NewDefaultSendMailOpts()
	// temporary cc to RD
	opts.CC = []string{
		"fei.chen@pingcap.com",
		"liuzhi@pingcap.com",
	}
	if err := w.mailFacade.SendMailWithTemplate(ctx, tos, config.CustomerStatusUpdateSubject, config.TNCustomerUpdateSubscribeStatusChanged, vars, opts); err != nil {
		return fmt.Errorf("send mail to ops failed: %w", err)
	}
	return nil
}

func (w MarketplaceWorker) Run(ctx context.Context) error {
	if !config.Global().MarketPlaceConfig.EnableBackground {
		w.logger.Info("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
}