package store

import (
	"context"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/creasty/defaults"
	"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_PLANS_COLUMNS is all the columns of `plans` table.
	TABLE_PLANS_COLUMNS = "plans.id,plans.type,plans.name,plans.deleted_at,plans.creator,plans.created_at,plans.updated_at,plans.wait_timeout,plans.test_bed,plans.pipeline,plans.variables,plans.meta_params,plans.fail_fast,plans.is_steps,plans.gh_repo,plans.gh_branch,plans.gh_commit,plans.gh_commit_ts,plans.gh_path,plans.gh_schedule,plans.interrupt_action"
)

type planService struct {
	db      *sql.Database
	config  *utils.Config
	caseSvc *caseService
}

func (s *planService) init(db *sql.Database, config *utils.Config, caseSvc *caseService) {
	s.db = db
	s.config = config
	s.caseSvc = caseSvc
}

func (s *planService) AllExist(ctx utils.Context, ids map[model.ID]struct{}) (bool, error) {
	if len(ids) == 0 {
		return true, nil
	}

	ids_ := make([]model.ID, len(ids))
	i := 0
	for id := range ids {
		ids_[i] = id
		i++
	}

	var matchCount int
	query, args, _ := sqlx.In("SELECT COUNT(*) FROM plans WHERE id IN (?) AND deleted_at=0", ids_)
	err := s.db.Get(&matchCount, query, args...)
	if err != nil {
		return false, err
	}
	return matchCount == len(ids), nil
}

func (s *planService) List(ctx utils.Context, opts common.PlanQueryOpts) ([]*model.TestPlan, int, error) {
	c := ctx.Request().Context()
	defer timing.DB(c, "plan:list")

	stmt := sql.Builder.From("plans").Prepared(true)
	if !opts.IncludeDeleted {
		stmt = stmt.Where(goqu.L("plans.deleted_at=0"))
	}
	if opts.Type != "" {
		stmt = stmt.Where(goqu.I("plans.type").Eq(opts.Type))
	}
	if opts.Branch != "" {
		stmt = stmt.Where(goqu.I("plans.gh_branch").Eq(opts.Branch))
	}
	if len(opts.IDs) > 0 {
		stmt = stmt.Where(goqu.I("plans.id").In(opts.IDs))
	}
	if opts.FromUpdatedAt != nil {
		stmt = stmt.Where(goqu.I("plans.updated_at").Gt(opts.FromUpdatedAt))
	}
	if len(opts.Tags) > 0 {
		// join table `plan_tags`
		tagFilter := sql.Builder.
			Select(goqu.COUNT("*").Eq(len(opts.Tags))).
			From("plan_tags").
			Where(goqu.L("cases.id=plan_tags.case_id"), goqu.I("plan_tags.name").In(opts.Tags)).
			Prepared(true)
		stmt = stmt.Where(tagFilter)
	}

	g := errgroup.Group{}

	var total int
	stmt_ := stmt
	g.Go(func() (err error) {
		query, args, _ := stmt_.Select(goqu.COUNT("*")).ToSQL()
		if err := s.db.GetContext(c, &total, query, args...); err != nil {
			return err
		}
		return nil
	})

	plans := []*model.TestPlan{}
	// The cols that will be returned to client.
	cols := `plans.id,
	plans.type,
	plans.name,
	plans.creator,
	plans.fail_fast,
	plans.gh_repo,
	plans.gh_branch,
	plans.gh_commit,
	plans.gh_path,
	plans.gh_schedule,
	plans.created_at,
	plans.updated_at`
	if opts.AllFields {
		cols = TABLE_PLANS_COLUMNS
	}

	// Order by
	orderBy := []exp.OrderedExpression{}
	if opts.SortedBy != "" {
		orderBy = []exp.OrderedExpression{goqu.I(string(opts.SortedBy)).Asc(), goqu.I("plans.id").Asc()}
		if opts.SortedDescending {
			orderBy = []exp.OrderedExpression{goqu.I(string(opts.SortedBy)).Desc(), goqu.I("plans.id").Desc()}
		}
	}

	// perform query
	query, args, _ := stmt.Select(
		goqu.L(cols),
		goqu.L("COALESCE(users.email, '') AS creator_email"),
	).
		LeftJoin(goqu.T("users"), goqu.On(goqu.I("users.id").Eq(goqu.I("plans.creator")))).
		Order(orderBy...).
		Limit(uint(opts.Count)).
		Offset(uint(opts.Offset)).
		ToSQL()
	if err := s.db.SelectContext(c, &plans, query, args...); err != nil {
		ctx.Log().Errorf(err, "list cases failed")
		return nil, 0, err
	}
	if err := g.Wait(); err != nil {
		return nil, 0, err
	}
	return plans, total, nil
}

func (s *planService) AutoComplete(ctx utils.Context, ty model.TestPlanType, name string, count int) ([]*model.TestPlan, error) {
	plans := []*model.TestPlan{}
	err := s.db.Select(&plans, "SELECT id, name, type, creator, created_at, updated_at, deleted_at FROM plans WHERE type = ? AND name LIKE ? AND deleted_at=0 ORDER BY name LIMIT ?",
		ty,
		sql.SanitizeLike(name)+"%",
		count,
	)
	if err != nil {
		ctx.Log().With("planName", name).Errorf(err, "AutoCompletePlan failed")
		return nil, err
	}

	return plans, nil
}

func (s *planService) Get(ctx utils.Context, id model.ID) (*model.TestPlan, error) {
	defer timing.DB(ctx.Request().Context(), "plan:get").End()
	// get plan
	p := &model.TestPlan{}
	err := s.db.Get(p, `
	SELECT 
		`+TABLE_PLANS_COLUMNS+`,
		COALESCE(users.email, '') AS creator_email,
		(SELECT GROUP_CONCAT(plan_tags.name SEPARATOR ',') FROM plan_tags WHERE plan_tags.plan_id=plans.id) AS tags
	FROM plans 
	LEFT JOIN users ON plans.creator=users.id
	WHERE plans.id=?`,
		id,
	)
	if err != nil {
		if sql.IsErrNoRows(err) {
			return nil, nil
		}
		ctx.Log().With("planID", id).Errorf(err, "get plan failed")
		return nil, err
	}

	// get pipeline
	if len(p.Pipeline) == 0 {
		p.Pipeline, err = s.getSteps(ctx, id)
		if err != nil {
			return nil, err
		}
	}
	if p.IsSteps {
		p.Steps = p.Pipeline
		p.Pipeline = nil
	}
	return p, nil
}

func (s *planService) GetReleasePlans(ctx utils.Context, branch string, maxPlan ...int) (planIDs []model.ID, err error) {
	max := "100"
	if len(maxPlan) > 0 {
		max = strconv.Itoa(maxPlan[0])
	}
	err = s.db.Select(&planIDs, "SELECT id FROM plans WHERE gh_branch=? AND EXISTS (SELECT * FROM plan_tags t WHERE t.plan_id=plans.id AND t.name='release') LIMIT "+max, branch)
	if err != nil {
		ctx.Log().With("branch", branch).Errorf(err, "GetReleasePlans failed")
	}
	return
}

func (s *planService) Create(ctx utils.Context, p *reqM.PlanCreateRequest) (*model.TestPlan, error) {
	pid, err := s.create(ctx, p)
	if err != nil {
		return nil, err
	}

	if ctx.DryRun() {
		return &model.TestPlan{}, nil
	}

	plan, err := s.Get(ctx, pid)
	if err != nil {
		return nil, err
	}
	return plan, nil
}

func (s *planService) Delete(ctx utils.Context, id model.ID) error {
	creator, err := s.getCreator(ctx, id)
	if err != nil {
		ctx.Log().Errorf(err, "GetPlanCreator failed")
		return err
	} else if creator == nil {
		return serror.NewMsg(http.StatusNotFound, "plan #%d not found", id)
	} else if *creator != ctx.User().ID {
		return serror.NewMsg(http.StatusForbidden, "the creator is not you")
	}

	if err := s.delete(ctx, []model.ID{id}, false); err != nil {
		return err
	}
	return nil
}

func (s *planService) getCreator(ctx utils.Context, id model.ID) (creator *model.ID, err error) {
	err = s.db.GetContext(ctx.Request().Context(), &creator, "SELECT creator FROM plans WHERE id=? AND deleted_at=0", id)
	if err != nil {
		if sql.IsErrNoRows(err) {
			return nil, nil
		}
		return nil, err
	}

	return
}

// getSteps returns steps of plan.
func (s *planService) getSteps(ctx utils.Context, planID model.ID) ([]model.PipelineStep, error) {
	steps := []model.PipelineStep{}
	err := s.db.Select(&steps, `
	SELECT 
		s.plan_id, s.name, s.seq, s.timeout, s.after, s.arguments, s.ref_case_id, COALESCE(cases.name, '') AS ref_case_name, s.ref_case_image 
	FROM 
		plan_steps s
	LEFT JOIN
		cases ON s.ref_case_id=cases.id 
	WHERE s.plan_id = ?
	ORDER BY s.seq ASC`,
		planID,
	)
	if err != nil {
		ctx.Log().Errorf(err, "query plan_steps failed")
		return nil, err
	}
	if len(steps) == 0 {
		// maybe these steps is stored in `plans` table
		var pipeline model.Pipeline
		if err = s.db.Select(&pipeline, "SELECT pipeline FROM plans WHERE id = ?", planID); err != nil {
			ctx.Log().Errorf(err, "query pipeline failed")
			return nil, err
		}
		steps = []model.PipelineStep(pipeline)
	}
	return steps, nil
}

func (s *planService) insertSteps(tx *sqlx.Tx, planID model.ID, pipeline []model.PipelineStep, deletedAt ...int64) error {
	if len(pipeline) == 0 {
		return nil
	}
	var deleted int64
	if len(deletedAt) > 0 {
		deleted = deletedAt[0]
	}

	b := sql.Builder.
		Insert("plan_steps").
		Cols("plan_id", "name", "seq", "timeout", "after", "arguments", "ref_case_id", "ref_case_image", "deleted_at").
		Prepared(true)
	for _, s := range pipeline {
		b = b.Vals(goqu.Vals{planID, s.Name, s.Seq, s.Timeout, s.After, s.Arguments, s.CaseID, s.CaseImage, deleted})
	}
	query, args, _ := b.ToSQL()
	if _, err := tx.Exec(query, args...); err != nil {
		return err
	}
	return nil
}

// NOTE: Inline-case is not searchable, updatable and could not be referred by any other plan, so we mark inline-case as deleted when creating.
func (s *planService) createInlineCases(ctx utils.Context, planName string, inlineCases []*model.TestCaseBase, pipeline []model.PipelineStep) error {
	deletedAt := time.Now().UnixNano()
	newCases := make([]*reqM.CaseCreateRequest, len(inlineCases))
	case2seq := make(map[string]int, len(inlineCases))
	for i, c := range inlineCases {
		newCases[i] = &reqM.CaseCreateRequest{
			TestCaseBase: *c,
			TestCaseMeta: model.TestCaseMeta{
				Summary: fmt.Sprintf("Inline case for plan %s.", planName),
			},
			DesignerID: ctx.User().ID,
			DeletedAt:  deletedAt,
		}
		case2seq[c.Name] = i
	}
	ids, err := s.caseSvc.BatchCreate(ctx, nil, newCases)
	if err != nil {
		ctx.Log().With("planName", planName).Errorf(err, "create inline cases failed")
		return err
	}

	// assign inline-case's id to plan-step
	for i := range pipeline {
		step := &pipeline[i]
		if seq, ok := case2seq[step.CaseName]; ok {
			step.CaseID = ids[seq]
		}
	}
	return nil
}

func (s *planService) create(ctx utils.Context, p *reqM.PlanCreateRequest) (model.ID, error) {
	defer timing.DB(ctx.Request().Context(), "plan:create").End()
	log := ctx.Log().With("planName", p.Name)
	defaults.MustSet(p)

	// 1. Create inline cases and assign case id to steps
	pipeline := p.Pipeline
	p.IsSteps = len(p.Steps) > 0
	if p.IsSteps {
		pipeline = p.Steps
	}
	if err := s.completeSteps(ctx, p.Name, p.InlineCases, pipeline); err != nil {
		return 0, err
	}

	if ctx.DryRun() {
		return 0, nil
	}

	var deletedAt int64 = 0
	// Oneshot plan is not updatable, searchable and couldn't have triggers,
	// thus we mark oneshot plan as deleted when create.
	if p.Type == model.TestPlanKindOneShot {
		deletedAt = time.Now().UnixNano()
	}

	var id model.ID
	err := s.db.Transaction(ctx.Request().Context(), func(tx *sqlx.Tx) (err error) {
		// 2. insert plan
		var plansPipeline model.Pipeline
		if p.Type == model.TestPlanKindOneShot {
			// We store the pipeline in `plans` but not `plan_steps` table.
			// Because there is no need to search for it.
			plansPipeline = pipeline
		}
		r, err := tx.Exec(`INSERT INTO plans (type, name, creator, test_bed, pipeline, variables, meta_params, fail_fast, is_steps, deleted_at, wait_timeout, interrupt_action) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			p.Type,
			p.Name,
			p.Creator,
			p.TestBed,
			plansPipeline,
			p.Variables,
			p.MetaParams,
			p.FailFast,
			p.IsSteps,
			deletedAt,
			p.WaitTimeout,
			p.InterruptAction,
		)
		if err != nil {
			if sql.IsErrDuplicate(err) {
				return serror.NewMsg(http.StatusBadRequest, "plan is already exist, name= %s", p.Name)
			}
			log.Errorf(err, "CreatePlan failed")
			return err
		}
		id_, err := r.LastInsertId()
		if err != nil {
			log.Errorf(err, "CreatePlan get last insert id failed")
			return err
		}
		id = model.ID(id_)

		// 3. insert plan's steps
		if p.Type == model.TestPlanKindOneShot {
			// We store the pipeline in `plans` but not `plan_steps` table.
			return nil
		}
		if err := s.insertSteps(tx, id, pipeline, deletedAt); err != nil {
			log.Errorf(err, "CreatePlan insert steps failed")
			return err
		}
		return nil
	})
	return id, err
}

func (s *planService) delete(ctx utils.Context, ids []model.ID, hard bool) error {
	log := ctx.Log().With("planIDs", ids)

	if len(ids) == 0 {
		return nil
	}

	// hard delete
	if hard {
		return s.db.Transaction(context.Background(), func(tx *sqlx.Tx) error {
			// delete plans
			query, args, _ := sqlx.In(`DELETE FROM plans WHERE id IN (?)`, ids)
			if _, err := tx.Exec(query, args...); err != nil {
				log.Errorf(err, "delete old gh plans failed")
				return err
			}

			// delete plans' steps
			query, args, _ = sqlx.In("DELETE FROM plan_steps WHERE plan_id IN (?)", ids)
			if _, err := tx.Exec(query, args...); err != nil {
				log.Errorf(err, "DeletePlan delete steps failed")
				return err
			}

			// delete plans' tags
			query, args, _ = sqlx.In("DELETE FROM plan_tags WHERE plan_id IN (?)", ids)
			if _, err := tx.Exec(query, args...); err != nil {
				log.Errorf(err, "DeletePlan delete tags failed")
				return err
			}

			// delete corresponding triggers
			return s.deleteTrigger(ctx, tx, ids)
		})
	}

	// soft delete
	deletedAt := time.Now().Unix()
	return s.db.Transaction(context.Background(), func(tx *sqlx.Tx) error {
		query, args, _ := sqlx.In(`UPDATE plans SET deleted_at=? WHERE id IN (?)`, deletedAt, ids)
		if _, err := tx.Exec(query, args...); err != nil {
			log.Errorf(err, "delete old gh plans failed")
			return err
		}

		query, args, _ = sqlx.In("UPDATE plan_steps SET deleted_at=? WHERE plan_id IN (?)", deletedAt, ids)
		if _, err := tx.Exec(query, args...); err != nil {
			log.Errorf(err, "DeletePlan delete steps failed")
			return err
		}

		// delete plans' tags
		query, args, _ = sqlx.In("DELETE FROM plan_tags WHERE plan_id IN (?)", ids)
		if _, err := tx.Exec(query, args...); err != nil {
			log.Errorf(err, "DeletePlan delete tags failed")
			return err
		}

		// delete corresponding triggers
		return s.deleteTrigger(ctx, tx, ids)
	})
}

// deleteTrigger deletes the corresponding triggers of these `planIDs`.
func (s *planService) deleteTrigger(ctx utils.Context, tx *sqlx.Tx, planIDs []model.ID) error {
	if len(planIDs) == 0 {
		return nil
	}

	var execer sqlx.Execer = s.db
	if tx != nil {
		execer = tx
	}
	query, args, _ := sqlx.In("DELETE FROM triggers WHERE plan_id IN (?)", planIDs)
	if _, err := execer.Exec(query, args...); err != nil {
		return err
	}
	return nil
}

type caseBase struct {
	*model.TestCaseBase
	id         model.ID
	refBySteps []int
}

// completeSteps assigns case id to step and auto generate step name/seq.
func (s *planService) completeSteps(ctx utils.Context, planName string, inlineCases []*model.TestCaseBase, pipeline []model.PipelineStep) error {
	// Inline case
	inlineCaseNames := make(map[string]*caseBase, len(inlineCases))
	// TCMS case can ref to a case's id or name
	tcmsCaseIDs := map[model.ID]*caseBase{}
	tcmsCaseNames := map[string]*caseBase{}

	// Do some initialization.
	for _, c := range inlineCases {
		inlineCaseNames[c.Name] = &caseBase{c, 0, nil}
	}
	for seq := range pipeline {
		step := &pipeline[seq]
		// generate step name base on case name.
		if step.Name == "" {
			step.Name = s.genStepName(seq, step.CaseName)
		}
		// assign seq to each step
		step.Seq = uint(seq)

		var c *caseBase
		if step.CaseName != "" {
			if inlineCase, isRefsToInlineCase := inlineCaseNames[step.CaseName]; isRefsToInlineCase {
				c = inlineCase
			} else {
				// if the case that the step refs to is not an inlined-case, search the case in TCMS
				if _, ok := tcmsCaseNames[step.CaseName]; !ok {
					tcmsCaseNames[step.CaseName] = &caseBase{}
				}
				c = tcmsCaseNames[step.CaseName]
			}
		} else {
			if _, ok := tcmsCaseIDs[step.CaseID]; !ok {
				tcmsCaseIDs[step.CaseID] = &caseBase{}
			}
			c = tcmsCaseIDs[step.CaseID]
		}
		c.refBySteps = append(c.refBySteps, seq)
	}

	// Every inline-case must be referred by at least one step.
	for _, c := range inlineCaseNames {
		if len(c.refBySteps) == 0 {
			return serror.NewMsg(http.StatusBadRequest, "inline-case %s is not referred by any pipeline step", c.Name)
		}
	}

	g := errgroup.Group{}
	// 1. Create inline cases
	if len(inlineCases) > 0 && !ctx.DryRun() {
		g.Go(func() (err error) {
			if err := s.createInlineCases(ctx, planName, inlineCases, pipeline); err != nil {
				ctx.Log().Errorf(err, "create inline cases failed")
				return err
			}
			return nil
		})
	}
	// 2. TCMS cases that used in pipeline must exist
	if err := s.validateStepCaseExist(ctx, tcmsCaseIDs, tcmsCaseNames); err != nil {
		return err
	}
	if err := g.Wait(); err != nil {
		return err
	}

	// 3. For each step, checking the parameters mapping.
	for i := range pipeline {
		step := &pipeline[i]
		var c *caseBase
		if step.CaseName != "" {
			var isRefsToInlineCase bool
			c, isRefsToInlineCase = inlineCaseNames[step.CaseName]
			if !isRefsToInlineCase {
				// set the case id to avoid duplicate work (the id of inline-case will be set in step 1.)
				c = tcmsCaseNames[step.CaseName]
				step.CaseID = c.id
			}
		} else {
			c = tcmsCaseIDs[step.CaseID]
			step.CaseName = c.Name
		}

		if err := s.validateStepArguments(step, c); err != nil {
			return err
		}
	}

	return nil
}

func (s *planService) validateStepCaseExist(ctx utils.Context, tcmsCaseIDs map[model.ID]*caseBase, tcmsCaseNames map[string]*caseBase) error {
	g := errgroup.Group{}
	if len(tcmsCaseIDs) > 0 {
		g.Go(func() error {
			ids := make([]model.ID, len(tcmsCaseIDs))
			for id := range tcmsCaseIDs {
				ids = append(ids, id)
			}
			cases_, err := s.caseSvc.ListBasicInfo(ctx, ids, nil, true)
			if err != nil {
				return err
			}
			for _, c := range cases_ {
				b := tcmsCaseIDs[c.ID]
				b.TestCaseBase, b.id = &c.TestCaseBase, c.ID
			}
			if len(cases_) != len(ids) {
				for id, b := range tcmsCaseIDs {
					if b.TestCaseBase == nil {
						return serror.NewMsg(http.StatusBadRequest, "case #%d used in pipeline-steps %v is not found", id, b.refBySteps)
					}
				}
			}
			return nil
		})
	}
	if len(tcmsCaseNames) > 0 {
		names := make([]string, len(tcmsCaseNames))
		for n := range tcmsCaseNames {
			names = append(names, n)
		}
		cases_, err := s.caseSvc.ListBasicInfo(ctx, nil, names, true)
		if err != nil {
			return err
		}
		for _, c := range cases_ {
			b := tcmsCaseNames[c.Name]
			b.TestCaseBase, b.id = &c.TestCaseBase, c.ID
		}
		if len(cases_) != len(names) {
			for name, b := range tcmsCaseNames {
				if b.TestCaseBase == nil {
					return serror.NewMsg(http.StatusBadRequest, "case `%s` used in pipeline-steps %v is not found", name, b.refBySteps)
				}
			}
		}
	}
	if err := g.Wait(); err != nil {
		return err
	}
	return nil
}

func (s *planService) validateStepArguments(step *model.PipelineStep, c *caseBase) error {
	stepArgs := make(map[string]bool, len(step.Arguments))
	for _, a := range step.Arguments {
		stepArgs[a.Name] = false
	}

	for _, p := range c.Parameters {
		required := p.Type != model.CaseParamTypeString // required if its type is not string
		_, existed := stepArgs[p.Name]

		// required but not existed
		if required && !existed {
			return serror.NewMsg(http.StatusBadRequest, `pipeline-step "%s" should set a value for arguement "%s" (required by case "%s")`, step.Name, p.Name, c.Name)
		}

		// mark arguement as valid
		stepArgs[p.Name] = true
	}

	for arg, valid := range stepArgs {
		if !valid {
			return serror.NewMsg(http.StatusBadRequest, `pipeline-step "%s" has an invalid arguement "%s" (does not defined by case "%s")`, step.Name, arg, c.Name)
		}
	}
	return nil
}

// Plan's details URL on TCMS.
func (s *planService) TCMSURL(planID model.ID) string {
	return fmt.Sprintf("%s/dashboard/plans/%d", s.config.BaseURL, planID)
}

// format: {step_seq}-{case_name}
func (s *planService) genStepName(stepSeq int, caseName string) string {
	suffix := s.caseName2stepNameSuffix(caseName)
	return fmt.Sprintf("step-%d%s", stepSeq, suffix)
}

var (
	caseName2stepNameReplacer = strings.NewReplacer(
		`_`, `-`,
		`.`, `-`,
	)
)

func (s *planService) caseName2stepNameSuffix(caseName string) string {
	if caseName == "" {
		return ""
	}
	return "-" + caseName2stepNameReplacer.Replace(caseName)
}