package opworker

import (
	"sync"

	"github.com/pingcap/dbaas/central/db"
	"github.com/pingcap/dbaas/central/db/entity"
	api "github.com/pingcap/dbaas/central/infra-api"
	"github.com/pingcap/dbaas/central/model/backup"
	billingworker "github.com/pingcap/dbaas/central/model/billing/worker"
	"github.com/pingcap/dbaas/central/model/cdc"
	"github.com/pingcap/dbaas/central/model/cluster"
	pimport "github.com/pingcap/dbaas/central/model/import"
	"github.com/pingcap/dbaas/central/model/replication"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	mailv2 "github.com/tidbcloud/pkgv2/mail/v2"
	"github.com/tidbcloud/pkgv2/worker"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

type Manager struct {
	clusterModel     *cluster.Model
	backupModel      *backup.Model
	importModel      *pimport.Model
	replicationModel *replication.Model
	cdcModel         *cdc.Model
	billingModel     *billingworker.BgModel
	backgroundWorker sync.Map
	lck              sync.Mutex
	api              *api.API
	mailFacade       mailv2.Facade
}

func New(clusterModel *cluster.Model, backupModel *backup.Model, importModel *pimport.Model, replicationModel *replication.Model, cdcModel *cdc.Model, billingModel *billingworker.BgModel, api *api.API, mailFacade mailv2.Facade) *Manager {
	return &Manager{
		clusterModel:     clusterModel,
		backupModel:      backupModel,
		importModel:      importModel,
		replicationModel: replicationModel,
		cdcModel:         cdcModel,
		billingModel:     billingModel,
		backgroundWorker: sync.Map{},
		lck:              sync.Mutex{},
		api:              api,
		mailFacade:       mailFacade,
	}
}

func (c *Manager) StartBackgroundWorkers() error {
	var tenants []entity.Tenant
	err := db.Transaction(func(tx *gorm.DB) error {
		return entity.NewTenantQuerySet(tx).StatusEq(entity.TenantStatusActive).All(&tenants)
	})
	if err != nil {
		return errors.Annotate(err, "retrieve active tenants failed")
	}
	for _, t := range tenants {
		if err := c.StartBackgroundWorkerForTenant(t.ID); err != nil {
			return err
		}
	}

	w, err := NewGlobalBgWorker(c)
	if err != nil {
		return err
	}

	w.Start()
	return nil
}

func (c *Manager) StartBackgroundWorkerForTenant(tenantID entity.ID) error {
	c.lck.Lock()
	defer c.lck.Unlock()

	_, ok := c.backgroundWorker.Load(tenantID.Uint64())
	if ok {
		log.Info("background worker already started, skip", zap.Uint64("tenantID", tenantID.Uint64()))
		return nil
	}
	log.Info("starting background worker", zap.Uint64("tenantID", tenantID.Uint64()))

	cbg, err := c.clusterModel.BgContext(tenantID)
	if err != nil {
		return errors.Annotate(err, "clusterModel")
	}
	bbg, err := c.backupModel.BgContext(tenantID)
	if err != nil {
		return errors.Annotate(err, "backupModel")
	}
	ibg, err := c.importModel.BgContext(tenantID)
	if err != nil {
		return errors.Annotate(err, "importModel")
	}

	rbg, err := c.replicationModel.BgContext(tenantID)
	if err != nil {
		return errors.Annotate(err, "replicationModel")
	}

	dbg, err := c.cdcModel.BgContext(tenantID)
	if err != nil {
		return errors.Annotate(err, "cdcModel")
	}

	billingBg := c.billingModel.BgContext(tenantID)

	w, err := NewBgWorker(cbg, bbg, ibg, rbg, dbg, billingBg, tenantID, c.api)
	if err != nil {
		return errors.Annotate(err, "opworker")
	}
	c.backgroundWorker.Store(tenantID.Uint64(), w)
	w.Start()
	return nil
}

func (c *Manager) StopBackgroundWorkerForTenant(tenantID entity.ID) error {
	c.lck.Lock()
	defer c.lck.Unlock()

	log.Info("stopping background worker", zap.Uint64("tenantID", tenantID.Uint64()))
	w, ok := c.backgroundWorker.Load(tenantID)
	if ok {
		c.backgroundWorker.Delete(tenantID)
		go func() {
			w.(worker.Worker).Stop()
			log.Info("background worker is stopped", zap.Uint64("tenantID", tenantID.Uint64()))
		}()
		return nil
	}
	return errors.Errorf("tenant %d worker is not found", tenantID)
}

func (c *Manager) GetBackgroundWorker(tenantID entity.ID) worker.Worker {
	w, found := c.backgroundWorker.Load(tenantID)
	if !found {
		return nil
	}
	return w.(worker.Worker)
}