package store

import (
	"errors"
	"net/http"
	"time"

	"github.com/doug-martin/goqu/v9"
	"github.com/doug-martin/goqu/v9/exp"
	"github.com/jmoiron/sqlx"
	"golang.org/x/sync/errgroup"

	"github.com/pingcap/test-infra/common/model"
	reqM "github.com/pingcap/test-infra/common/model/tcms/request"
	"github.com/pingcap/test-infra/services/tcms/service/common"
	serror "github.com/pingcap/test-infra/services/tcms/service/common/errors"
	"github.com/pingcap/test-infra/services/tcms/utils"
	"github.com/pingcap/test-infra/services/tcms/utils/sql"
	"github.com/pingcap/test-infra/services/tcms/utils/timing"
)

const (
	TABLE_PLAN_EXECUTIONS_COLUMNS = "plan_executions.id,plan_executions.plan_id,plan_executions.trigger_id,plan_executions.priority,plan_executions.creator,plan_executions.created_at,plan_executions.updated_at,plan_executions.started_at,plan_executions.completed_at,plan_executions.status,plan_executions.output,plan_executions.note,plan_executions.reason,plan_executions.retry_cnt,plan_executions.rerun_id,plan_executions.steps,plan_executions.meta,plan_executions.argo_url,plan_executions.variables,plan_executions.meta_params,plan_executions.resourcepool_id"
)

type planExecutionService struct {
	db      *sql.Database
	config  *utils.Config
	planSvc *planService
}

func (s *planExecutionService) init(db *sql.Database, planSvc *planService) {
	s.db = db
	s.planSvc = planSvc
}

type planExecCreateOpts struct {
	// required field
	PlanID, TriggerID, CreatorID model.ID
	Priority                     model.ExecPriority
	CreatedAt                    time.Time
	RerunID                      model.ID
	Steps                        model.StringSlice
	VariableBinds                model.PlanVarBindSlice
	MetaParams                   model.PlanVarBindSlice
}

func (s *planExecutionService) getPlanID(ctx utils.Context, id model.ID) (*model.ID, error) {
	var planID model.ID
	err := s.db.Get(&planID, "SELECT plan_id FROM plan_executions WHERE id=?", id)
	if err != nil {
		ctx.Log().With("planExecID", id).Errorf(err, "PlanExecutionExist failed")
		return nil, err
	}
	return &planID, nil
}

func (s *planExecutionService) GetStatus(ctx utils.Context, id model.ID) (*model.ExecutionStatus, error) {
	var status model.ExecutionStatus
	err := s.db.Get(&status, "SELECT status FROM plan_executions WHERE id=?", id)
	if err != nil {
		if sql.IsErrNoRows(err) {
			return nil, nil
		}
		ctx.Log().Errorf(err, "GetPlanExecutionStatus failed")
		return nil, err
	}

	return &status, nil
}

func (s *planExecutionService) IsAllFinished(ctx utils.Context, createdAt time.Time) (bool, error) {
	var isAllFinished bool
	query, args, _ := sqlx.In("SELECT NOT EXISTS(SELECT * FROM plan_executions WHERE created_at=? and status IN (?))", createdAt, model.IncompletedExecutionStates)
	err := s.db.GetContext(ctx.Request().Context(), &isAllFinished, query, args...)
	if err != nil {
		ctx.Log().Errorf(err, "IsAllFinished failed")
		return false, err
	}
	return isAllFinished, nil
}

func (s *planExecutionService) ListFailedSteps(ctx utils.Context, id model.ID) (steps []string, err error) {
	// except success and omit
	failedStatus := append(model.ExecutionErrorStatus, model.ExecutionStatusCancelled, model.ExecutionStatusSkipped)
	query, args, _ := sqlx.In(`SELECT step_name FROM case_executions WHERE root_plan_exec_id = ? AND status IN (?) LIMIT ?`, id, failedStatus, model.MaxPlanStepsLength)
	if err := s.db.Select(&steps, query, args...); err != nil {
		ctx.Log().Errorf(err, "ListFailedSteps failed")
		return nil, err
	}
	return steps, nil
}

func (s *planExecutionService) List(ctx utils.Context, opts common.PlanExecQueryOpts) ([]*model.TestPlanExecution, int, error) {
	const MaxOffset = 1000
	if opts.Offset > MaxOffset {
		return nil, 0, serror.ErrMaxOffset(int(opts.Offset), MaxOffset)
	} else if opts.Count == 0 {
		return nil, 0, errors.New("expected count")
	}
	c := ctx.Request().Context()
	defer timing.DB(c, "plan-exec:list").End()

	where := []exp.Expression{}
	if len(opts.IDs) > 0 {
		where = append(where, goqu.I("plan_executions.id").In(opts.IDs))
	}
	if len(opts.PlanIDs) > 0 {
		where = append(where, goqu.I("plan_executions.plan_id").In(opts.PlanIDs))
	}
	if len(opts.TriggerIDs) > 0 {
		where = append(where, goqu.I("plan_executions.trigger_id").In(opts.TriggerIDs))
	}
	if len(opts.Status) > 0 {
		where = append(where, goqu.I("plan_executions.status").In(opts.Status))
	}
	if opts.OnlyCron {
		subqueryTrigger := sql.Builder.
			Select("*").
			From("triggers").
			Where(goqu.L("plan_executions.trigger_id=triggers.id AND triggers.type='" + string(model.TriggerTypeCron) + "'"))

		where = append(where, goqu.Func("EXISTS", subqueryTrigger))
	}
	if opts.HasLivingTestbed || opts.ResourcePoolID != nil {
		subqueryTestbeds := sql.Builder.
			Select("plan_execution_id").
			From("testbed_infos")

		if opts.HasLivingTestbed {
			subqueryTestbeds = subqueryTestbeds.Where(goqu.C("status").In([]model.TestBedInfoStatus{
				model.TestBedInfoStatusInitiating,
				model.TestBedInfoStatusReady,
			}))
		}

		if opts.ResourcePoolID != nil {
			subqueryTestbeds = subqueryTestbeds.Where(goqu.I("resource_pool_id").Eq(*opts.ResourcePoolID))
		}

		where = append(where, goqu.I("plan_executions.id").In(subqueryTestbeds))
	}
	if opts.TriagePhase != nil {
		subqueryTriages := sql.Builder.
			Select("plan_execution_id").
			From("triages").
			Where(goqu.I("triage_phase").Eq(*opts.TriagePhase))

		where = append(where, goqu.I("plan_executions.id").In(subqueryTriages))
	}
	if opts.FromStartedAt != nil {
		where = append(where, goqu.I("plan_executions.started_at").Gte(opts.FromStartedAt))
	}
	if opts.ToStartedAt != nil {
		where = append(where, goqu.I("plan_executions.started_at").Lt(opts.ToStartedAt))
	}
	if opts.CreatedAt != nil {
		where = append(where, goqu.I("plan_executions.created_at").Eq(opts.CreatedAt))
	}
	if opts.FromUpdatedAt != nil {
		where = append(where, goqu.I("plan_executions.updated_at").Gt(opts.FromUpdatedAt))
	}
	b := sql.Builder.From(goqu.T("plan_executions")).Where(where...).Prepared(true)

	g := errgroup.Group{}

	// Get total count.
	var total int
	b_ := b // prevent race
	g.Go(func() (err error) {
		query, args, _ := b_.Select(goqu.COUNT("*")).ToSQL()
		if err := s.db.GetContext(c, &total, query, args...); err != nil {
			ctx.Log().Errorf(err, "select total failed")
			return err
		}
		return nil
	})

	// Get items.
	var items []*model.TestPlanExecution
	if opts.SortedBy != "" {
		if opts.SortedDescending {
			b = b.Order(goqu.I(string(opts.SortedBy)).Desc(), goqu.I("id").Desc())
		} else {
			b = b.Order(goqu.I(string(opts.SortedBy)).Asc(), goqu.I("id").Asc())
		}

	}
	query, args, _ := b.Select(
		goqu.L(TABLE_PLAN_EXECUTIONS_COLUMNS),
		goqu.L("COALESCE(plans.name, '') AS plan_name"),
		goqu.L("COALESCE(users.email, '') AS creator_email"),
	).
		LeftJoin(goqu.T("plans"), goqu.On(goqu.I("plans.id").Eq(goqu.I("plan_executions.plan_id")))).
		LeftJoin(goqu.T("users"), goqu.On(goqu.I("users.id").Eq(goqu.I("plan_executions.creator")))).
		Limit(opts.Count).
		Offset(opts.Offset).
		ToSQL()
	if err := s.db.SelectContext(c, &items, query, args...); err != nil {
		ctx.Log().Errorf(err, "select plan-execution failed")
		return nil, 0, err
	}
	if err := g.Wait(); err != nil {
		return nil, 0, err
	}
	return items, total, nil
}

func (s *planExecutionService) Get(ctx utils.Context, id model.ID) (*model.TestPlanExecution, error) {
	log := ctx.Log().With("planExecutionID", id)

	exec := &model.TestPlanExecution{}
	err := s.db.Get(exec, `
	SELECT
	  `+TABLE_PLAN_EXECUTIONS_COLUMNS+`,
	  COALESCE(plans.name, '') AS plan_name,
	  COALESCE(users.email, '') AS creator_email
	FROM plan_executions 
	LEFT JOIN plans ON plans.id=plan_executions.plan_id
	LEFT JOIN users ON users.id=plan_executions.creator
	WHERE plan_executions.id=?`,
		id,
	)
	if err != nil {
		if sql.IsErrNoRows(err) {
			return nil, nil
		}
		log.Errorf(err, "GetPlanExecution failed")
		return nil, err
	}

	return exec, nil
}

func (s *planExecutionService) batchCreate(
	ctx utils.Context,
	tx *sqlx.Tx,
	execs []*planExecCreateOpts,
) ([]model.ID, error) {
	if len(execs) == 0 {
		return nil, nil
	}
	var execer sqlx.Execer = s.db
	if tx != nil {
		execer = tx
	}

	b := sql.Builder.
		Insert("plan_executions").
		Cols("plan_id", "trigger_id", "creator", "rerun_id", "steps", "priority", "created_at", "output", "note", "variables", "meta_params").
		Prepared(true)
	for _, c := range execs {
		b = b.Vals(goqu.Vals{
			c.PlanID,
			c.TriggerID,
			c.CreatorID,
			c.RerunID,
			c.Steps,
			c.Priority,
			c.CreatedAt,
			"",
			"",
			c.VariableBinds,
			c.MetaParams,
		})
	}

	query, args, _ := b.ToSQL()
	r, err := execer.Exec(query, args...)
	if err != nil {
		ctx.Log().Errorf(err, "BatchCreate plan-execs failed")
		return nil, err
	}
	firstRowID, err := r.LastInsertId()
	if err != nil {
		return nil, err
	}
	ids := make([]model.ID, len(execs))
	for i := 0; i < len(execs); i++ {
		/*
			NOTE: TiDB guarantees the consecutive autoID in a batch insert.
			See https://github.com/pingcap/tidb/pull/11876.
		*/
		ids[i] = model.ID(firstRowID + int64(i))
	}
	return ids, nil
}

func (s *planExecutionService) Update(ctx utils.Context, id model.ID, exec *reqM.PlanExecutionUpdateRequest) (*model.TestPlanExecution, error) {
	log := ctx.Log().With("planExecutionID", id)

	_, err := s.db.Exec(`UPDATE plan_executions SET started_at=?, completed_at=?, status=?, output=?, priority=?, retry_cnt=?, meta=?, argo_url=?, resourcepool_id=? WHERE id=?`,
		exec.StartedAt,
		exec.CompletedAt,
		exec.Status,
		exec.Output,
		exec.Priority,
		exec.RetryCnt,
		exec.Meta,
		exec.ArgoURL,
		exec.ResourcepoolID,
		id,
	)
	if err != nil {
		log.Errorf(err, "UpdatePlanExecution failed")
		return nil, err
	}
	return s.Get(ctx, id)
}

func (s *planExecutionService) UpdateNote(ctx utils.Context, id model.ID, note string) error {
	_, err := s.db.Exec(`UPDATE plan_executions SET note=? WHERE id=?`, note, id)
	if err != nil {
		ctx.Log().Errorf(err, "update plan-execution note failed")
		return err
	}
	return nil
}

func (s *planExecutionService) UpdateReason(ctx utils.Context, id model.ID, reason model.ExecFailReason) error {
	_, err := s.db.Exec(`UPDATE plan_executions SET reason=? WHERE id=?`, reason, id)
	if err != nil {
		ctx.Log().Errorf(err, "update plan-execution reason failed")
		return err
	}
	return nil
}

func (s *planExecutionService) Cancel(ctx utils.Context, planID *model.ID, ids []model.ID) error {
	if planID == nil && len(ids) == 0 {
		return serror.NewMsg(http.StatusBadRequest, "expect one of planID or plan-exec-ids")
	}

	where := goqu.Ex{}
	if planID != nil {
		where["plan_id"] = *planID
	}
	if len(ids) > 0 {
		where["id"] = ids
	}
	where["status"] = []model.ExecutionStatus{model.ExecutionStatusPending, model.ExecutionStatusPreparing, model.ExecutionStatusPrepareBackoff, model.ExecutionStatusRunning}
	query, args, _ := sql.Builder.Update("plan_executions").Prepared(true).Set(
		map[string]interface{}{"status": string(model.ExecutionStatusCancelling)},
	).Where(where).ToSQL()
	if _, err := s.db.ExecContext(ctx.Request().Context(), query, args...); err != nil {
		ctx.Log().With("planID", planID, "ids", ids).Errorf(err, "update plan-exec status failed")
		return err
	}
	return nil
}

func (s *planExecutionService) Rerun(ctx utils.Context, id model.ID, steps []string) (model.ID, error) {
	log := ctx.Log().With("id", id)

	// check the original plan-exec
	original, err := s.Get(ctx, id)
	if err != nil {
		log.Errorf(err, "get plan-exec failed")
		return 0, err
	} else if original == nil {
		return 0, serror.NewMsg(http.StatusNotFound, "plan-exec #%d not found", id)
	} else if !original.Status.IsCompleted() {
		return 0, serror.NewMsg(http.StatusBadRequest, "plan-exec #%d is not completed, current status = %v", id, original.Status)
	}
	rerunID := id
	if original.RerunID != 0 {
		rerunID = original.RerunID
	}
	log.With("oldStatus", original.Status).Infof("rerun plan-exec")

	// check whether these steps is exists
	// TODO(gaozifeng): Use the history plan.
	refPlanID := original.PlanIDAndName.ID
	if len(steps) > 0 {
		originalSteps, err := s.planSvc.getSteps(ctx, refPlanID)
		if err != nil {
			log.Errorf(err, "get plan failed")
			return 0, err
		} else if len(steps) == 0 {
			return 0, serror.NewMsg(http.StatusNotFound, "plan #%d has no steps", refPlanID)
		}

		var requiredSteps = utils.ToMap[string, bool](steps)
		for _, step := range originalSteps {
			if _, ok := requiredSteps[step.Name]; ok {
				requiredSteps[step.Name] = true
			}
		}
		notExistSteps := []string{}
		for step, isExists := range requiredSteps {
			if !isExists {
				notExistSteps = append(notExistSteps, step)
			}
		}
		if len(notExistSteps) > 0 {
			return 0, serror.NewMsg(
				http.StatusBadRequest,
				"rerun plan-execution failed, pipeline-steps %v not found in original plan %s, please consider rerunning the entire plan instead of parts of it",
				notExistSteps,
				s.planSvc.TCMSURL(refPlanID),
			)
		}
	}

	newExec := &planExecCreateOpts{
		PlanID:        refPlanID,
		TriggerID:     original.TriggerID,
		CreatorID:     ctx.User().ID,
		Priority:      original.Priority,
		CreatedAt:     time.Now(),
		RerunID:       rerunID,
		Steps:         steps,
		VariableBinds: original.VariableBinds,
		MetaParams:    original.MetaParams,
	}
	var execIDs []model.ID

	err = s.db.Transaction(ctx.Request().Context(), func(tx *sqlx.Tx) error {
		if execIDs, err = s.batchCreate(ctx, tx, []*planExecCreateOpts{newExec}); err != nil {
			log.Errorf(err, "rerun plan-exec failed")
			return err
		}

		// FIXME: Archived data is changed, might cause some confusions
		if original.TriggerID != 0 {
			_, err = tx.Exec(`UPDATE plan_executions SET trigger_id=0 WHERE id=?`,
				id,
			)
			if err != nil {
				log.Errorf(err, "set original plan execution TriggerID to 0 failed")
				return err
			}
		}

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

	return execIDs[0], nil
}

func (s *planExecutionService) CreateOneShotPlan(ctx utils.Context, plan *reqM.PlanCreateRequest) (p *model.TestPlan, e model.ID, err error) {
	if ctx.User().ID != 0 &&
		plan.Priority > model.ExecPriorityUserDebug {
		return nil, 0, serror.NewMsg(http.StatusBadRequest, "normal user not allowed to set priority higher than ExecPriorityUserDebug")
	}
	plan.Creator = ctx.User().ID
	err = s.db.Transaction(ctx.Request().Context(), func(tx *sqlx.Tx) error {
		p, e, err = s.CreateOneShotPlanTx(ctx, plan, tx)
		return err
	})
	return
}

func (s *planExecutionService) CreateOneShotPlanTx(ctx utils.Context, plan *reqM.PlanCreateRequest, db *sqlx.Tx) (*model.TestPlan, model.ID, error) {
	tp, err := s.planSvc.Create(ctx, plan)
	if err != nil {
		return nil, 0, err
	}
	if ctx.DryRun() {
		return tp, 0, nil
	}
	execs := []*planExecCreateOpts{
		{
			PlanID:    tp.ID,
			TriggerID: 0,
			CreatorID: plan.Creator,
			Priority:  model.ExecPriorityUserDebug,
			CreatedAt: time.Now(),
		},
	}

	execIDs, err := s.batchCreate(ctx, db, execs)
	if err != nil {
		ctx.Log().Errorf(err, "create oneshot plan's plan exec failed")
		return nil, 0, err
	}
	return tp, execIDs[0], nil
}

func (s *planExecutionService) CreateExecTx(ctx utils.Context, exec model.TestPlanExecution, db *sqlx.Tx) (model.ID, error) {
	execs := []*planExecCreateOpts{
		{
			PlanID:        exec.ID,
			TriggerID:     exec.TriggerID,
			CreatorID:     exec.CreatedBy.ID,
			Priority:      exec.Priority,
			CreatedAt:     exec.CreatedAt.Time(),
			VariableBinds: exec.VariableBinds,
		},
	}
	execIDs, err := s.batchCreate(ctx, db, execs)
	if err != nil {
		ctx.Log().Errorf(err, "create plan exec failed")
		return 0, err
	}
	return execIDs[0], nil
}