package cdc

// Database access should be done through this file.
import (
	"github.com/pingcap/errors"
	"github.com/tidbcloud/metrics-server/central/db"
	"github.com/tidbcloud/metrics-server/central/db/entity"
	"gorm.io/gorm"
)

type CdcFilter struct {
	ClusterID uint64
	Status    entity.CDCStatus
	Statuss   []entity.CDCStatus
	Deleted   *bool
	Creating  *bool
}

type CDCDao struct {
}

// fetches the entities of a QuerySet
func (d *CDCDao) FindCDCTasks(filter CdcFilter) ([]entity.CDCTask, error) {
	var (
		err      error
		cdcTasks []entity.CDCTask
	)

	err = db.Transaction(func(tx *gorm.DB) error {
		tx = tx.Preload("Cluster")

		if filter.ClusterID != 0 {
			tx = tx.Where("cdc_tasks.cluster_id = ?", filter.ClusterID)
		}

		if filter.Status != "" {
			tx = tx.Where("cdc_tasks.status = ?", filter.Status)
		}

		if len(filter.Statuss) > 0 {
			tx = tx.Where("cdc_tasks.status IN (?)", filter.Statuss)
		}

		if filter.Deleted != nil {
			if *filter.Deleted {
				tx = tx.Where("cdc_tasks.status IN (?)", entity.DeletedCdcStatus)
			} else {
				tx = tx.Where("cdc_tasks.status NOT IN (?)", entity.DeletedCdcStatus)
			}
		}

		if filter.Creating != nil {
			if *filter.Creating {
				tx = tx.Where("cdc_tasks.status IN (?)", entity.CreatingCdcStatus)
			} else {
				tx = tx.Where("cdc_tasks.status NOT IN (?)", entity.CreatingCdcStatus)
			}
		}

		qs := entity.NewCDCTaskQuerySet(tx)

		err = qs.OrderDescByCreatedAt().All(&cdcTasks)
		if err != nil {
			return errors.AddStack(err)
		}

		return nil
	})
	if err != nil {
		return nil, err
	}

	return cdcTasks, nil
}

// if cdc task is `deleting` only can change to `deleted` for prevent data
func (d *CDCDao) UpdateCdcTask(task entity.CDCTask, schemas []entity.CDCTaskDBSchemaField) error {
	return db.Transaction(func(tx *gorm.DB) error {
		haveStatus := false
		for _, s := range schemas {
			if s == entity.CDCTaskDBSchema.Status {
				haveStatus = true
			}
		}
		if haveStatus && task.Status != entity.CDCStatusDeleted {
			tmpTask, err := d.GetCdcTaskByID(task.ID, task.ClusterID, task.Cluster.ProjectID)
			if err != nil {
				return err
			}
			if tmpTask.Status == entity.CDCStatusDeleting {
				task.Status = entity.CDCStatusDeleting
			}
		}
		return task.Update(tx, schemas...)
	})
}

func (d *CDCDao) GetCdcSqlTask(taskID entity.ID) (*entity.CDCSQLTask, error) {
	var err error
	var sqlTask entity.CDCSQLTask
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewCDCSQLTaskQuerySet(tx).
			CDCTaskIDEq(taskID).
			One(&sqlTask)
	})
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get sql task")
	}
	return &sqlTask, nil
}

func (d *CDCDao) GetCdcKafkaTask(taskID entity.ID) (*entity.CDCKafkaTask, error) {
	var err error
	var kafkaTask entity.CDCKafkaTask
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewCDCKafkaTaskQuerySet(tx).
			CDCTaskIDEq(taskID).
			One(&kafkaTask)
	})
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get sql task")
	}
	return &kafkaTask, nil
}

func (d *CDCDao) GetCdcTaskByID(taskID, clusterID, projectID entity.ID) (*entity.CDCTask, error) {
	var (
		err  error
		task entity.CDCTask
	)
	err = db.Transaction(func(tx *gorm.DB) error {
		return entity.NewCDCTaskQuerySet(tx).
			PreloadCluster().
			IDEq(taskID).
			One(&task)
	})
	if err != nil {
		return nil, errors.Annotatef(err, "failed to get cdc task")
	}
	if task.Cluster.ID != clusterID {
		return nil, errors.Errorf("cluster %d does not have cdc task %d", clusterID, taskID)
	}
	if task.Cluster.ProjectID != projectID {
		return nil, errors.Errorf("project %d does not have cluster %d", projectID, clusterID)
	}
	return &task, nil
}