package store

import (
	"fmt"
	"math"
	"net/http"
	"strings"
	"time"

	"github.com/doug-martin/goqu/v9"
	"github.com/jmoiron/sqlx"
	"github.com/robfig/cron/v3"
	"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"
)

const (
	TABLE_TRIGGERS_COLUMNS = "triggers.id,triggers.name,triggers.type,triggers.creator,triggers.priority,triggers.summary,triggers.release_branch,triggers.created_at,triggers.updated_at,triggers.last_triggered_at,triggers.expr,triggers.ts,triggers.plan_id,triggers.plan_ids,triggers.variables,triggers.active"
)

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

type triggerService struct {
	db          *sql.Database
	planSvc     *planService
	planExecSvc *planExecutionService
}

func (s *triggerService) List(ctx utils.Context, opts common.TriggerQueryOpts) ([]*model.TestTrigger, int, error) {
	c := ctx.Request().Context()

	where := goqu.Ex{}
	if opts.PlanID != nil {
		where["plan_id"] = *opts.PlanID
	}
	if opts.Type != nil {
		where["type"] = *opts.Type
	}
	if opts.Creator != nil {
		where["creator"] = *opts.Creator
	}
	if opts.ReleaseBranch != "" {
		where["release_branch"] = opts.ReleaseBranch
	}
	if opts.IsActive != nil {
		where["active"] = *opts.IsActive
	}
	b := sql.Builder.From("triggers").Where(where).Prepared(true)

	g := errgroup.Group{}

	var total int
	query_, args_, _ := b.Select(goqu.COUNT("*")).ToSQL()
	g.Go(func() (err error) {
		if err := s.db.GetContext(c, &total, query_, args_...); err != nil {
			ctx.Log().Errorf(err, "select total failed")
			return err
		}
		return nil
	})

	var triggers []*model.TestTrigger
	query, args, _ := b.Select(goqu.L(TABLE_TRIGGERS_COLUMNS), goqu.L("COALESCE(users.email, '') AS creator_email")).
		LeftJoin(goqu.T("users"), goqu.On(goqu.I("users.id").Eq(goqu.I("triggers.creator")))).
		Order(goqu.I("triggers.created_at").Desc(), goqu.I("triggers.id").Asc()).
		Limit(uint(opts.Count)).
		Offset(uint(opts.Offset)).
		ToSQL()
	err := s.db.SelectContext(c, &triggers, query, args...)
	if err != nil {
		ctx.Log().Errorf(err, "select trigger failed")
		return nil, 0, err
	}

	if err := g.Wait(); err != nil {
		return nil, 0, err
	}
	return triggers, total, nil
}

func (s *triggerService) Trigger(ctx utils.Context, ty model.TriggerType, triggerAt time.Time, count int) (int, error) {
	log := ctx.Log().With("type", ty)

	triggers := []*model.TestTrigger{}
	err := s.db.Transaction(ctx.Request().Context(), func(tx *sqlx.Tx) error {
		// 1. Get triggers that need to be triggered now.
		if err := tx.Select(&triggers,
			"SELECT id, plan_id, plan_ids, creator, priority, expr, ts, variables FROM triggers WHERE active=1 AND type=? AND ts <= ? ORDER BY created_at asc LIMIT ?",
			ty,
			triggerAt.Unix(),
			count,
		); err != nil {
			log.Errorf(err, "failed to list triggers")
			return err
		}

		length := len(triggers)
		if length == 0 {
			return nil
		}
		ids := make([]model.ID, length)
		for i, t := range triggers {
			ids[i] = t.ID
		}
		log.Infof("%d triggers will be triggered, ids= %v", length, ids)

		// 2. Set last_triggered_at for all these triggers.
		stmt := strings.Builder{}
		stmt.WriteString("UPDATE triggers SET last_triggered_at=?")
		// (For ONCE and RELEASE) Mark as inactve
		if ty == model.TriggerTypeOnce || ty == model.TriggerTypeRelease {
			stmt.WriteString(", active=0")
		}
		stmt.WriteString(" WHERE id IN (?)")
		query, args, _ := sqlx.In(stmt.String(), triggerAt.Unix(), ids)
		if _, err := tx.Exec(query, args...); err != nil {
			log.Errorf(err, "SetTriggersLastTriggeredAt failed")
			return err
		}

		// 3. (Only for CRON) Set trigger-timestamp to the next trigger time.
		if ty == model.TriggerTypeCron {
			for _, t := range triggers {
				log = log.With("triggerID", t.ID)
				nextTriggerTS, err := s.parseTriggerExpr(model.TriggerTypeCron, t.Expr)
				if err != nil {
					log.Errorf(err, "!!! parse cron schedule failed, schedule= %s !!!", t.Expr)
					return err
				}
				if nextTriggerTS == t.Ts {
					continue
				}
				if _, err := tx.Exec("UPDATE triggers SET ts=? where id=?", nextTriggerTS, t.ID); err != nil {
					log.Errorf(err, "SetTriggerTs failed, ts= %d", nextTriggerTS)
					return err
				}
			}
		}

		return nil
	})
	if err != nil {
		log.Errorf(err, "trigger failed")
		return 0, err
	}

	// 4. Create corresponding plan-executions.
	planExecs := []*planExecCreateOpts{}
	for _, t := range triggers {
		var (
			priority model.ExecPriority
			pids     []model.ID
		)

		if t.ExecPriority != model.ExecPriorityDefault {
			priority = t.ExecPriority
		} else {
			// detect priority by trigger type
			switch ty {
			case model.TriggerTypeRelease:
				priority = model.ExecPriorityHigh
			case model.TriggerTypeOnce:
				priority = model.ExecPriorityUserDebug
			default:
				priority = model.ExecPriorityDefault
			}
		}

		// a trigger may ref to multiple plans
		if len(t.PlanIDs) > 0 {
			pids = t.PlanIDs
		} else if t.PlanID != 0 {
			pids = []model.ID{t.PlanID}
		} else {
			continue
		}

		for _, pid := range pids {
			planExecs = append(planExecs, &planExecCreateOpts{
				PlanID:        pid,
				TriggerID:     t.ID,
				CreatorID:     t.CreatedBy.ID,
				Priority:      priority,
				CreatedAt:     triggerAt,
				VariableBinds: t.VariableBinds,
			})
		}
	}
	if _, err := s.planExecSvc.batchCreate(ctx, nil, planExecs); err != nil {
		log.Errorf(err, "failed create plan-execs")
		return 0, err
	}

	return len(triggers), err
}

func (s *triggerService) AutoComplete(ctx utils.Context, name string, ty model.TriggerType, count int) ([]*model.TestTrigger, error) {
	t := []*model.TestTrigger{}
	err := s.db.Select(&t, "SELECT id, name, type, creator, created_at, updated_at FROM triggers WHERE type=? AND name LIKE ? ORDER BY name LIMIT ?",
		ty,
		sql.SanitizeLike(name)+"%",
		count,
	)
	if err != nil {
		ctx.Log().With("triggerType", ty, "triggerName", name).Errorf(err, "AutoCompleteTrigger failed")
		return nil, err
	}

	return t, nil
}

func (s *triggerService) Get(ctx utils.Context, id model.ID) (*model.TestTrigger, error) {
	log := ctx.Log().With("triggerID", id)

	t := &model.TestTrigger{}
	err := s.db.Get(t, `
	SELECT 
		`+TABLE_TRIGGERS_COLUMNS+`,
		COALESCE(users.email, '') AS creator_email
	FROM triggers LEFT JOIN users ON users.id=triggers.creator WHERE triggers.id=?`, id)
	if err != nil {
		if sql.IsErrNoRows(err) {
			return nil, nil
		}
		log.Errorf(err, "GetTrigger failed")
		return nil, err
	}

	return t, nil
}

// getCreator returns nil if trigger is not exist.
func (s *triggerService) getCreator(ctx utils.Context, id model.ID) (creator *model.ID, err error) {
	err = s.db.GetContext(ctx.Request().Context(), &creator, "SELECT creator FROM triggers WHERE id=?", id)
	if err != nil {
		if sql.IsErrNoRows(err) {
			return nil, nil
		}
		return nil, err
	}

	return
}

func (s *triggerService) Create(ctx utils.Context, t *reqM.TriggerCreateRequest) (*model.TestTrigger, error) {
	log := ctx.Log().With("triggerType", t.Type, "triggerName", t.Name)

	// 1. create inline oneshot plans
	inlinePlanIDs := make([]model.ID, len(t.InlinePlans))
	creator := ctx.User().ID
	if len(t.InlinePlans) > 0 {
		// TODO(gaozifeng): Max 10 concurrency?
		g := errgroup.Group{}
		for i, plan := range t.InlinePlans {
			i, plan := i, plan
			g.Go(func() (err error) {
				pid, err := s.planSvc.create(ctx, &reqM.PlanCreateRequest{
					TestPlanBase:                 plan.TestPlanBase,
					Type:                         model.TestPlanKindOneShot,
					PlanCreateRequestOneshotSpec: plan.PlanCreateRequestOneshotSpec,
					Creator:                      creator,
				})
				if err != nil {
					log.Errorf(err, "create inline plans failed")
					return err
				}
				inlinePlanIDs[i] = pid
				return nil
			})
		}
		if err := g.Wait(); err != nil {
			return nil, err
		}
		t.PlanIDs = append(t.PlanIDs, inlinePlanIDs...)
	}

	// 2. create trigger
	triggerID, err := s.create(ctx, nil, t)
	if err != nil {
		log.Errorf(err, "create trigger failed")
		return nil, err
	}
	if err != nil {
		if len(inlinePlanIDs) > 0 {
			if err := s.planSvc.delete(ctx, inlinePlanIDs, true); err != nil {
				log.Errorf(err, "delete inline plans failed")
			}
		}
		return nil, err
	}
	return s.Get(ctx, triggerID)
}

func (s *triggerService) create(ctx utils.Context, tx *sqlx.Tx, t *reqM.TriggerCreateRequest) (model.ID, error) {
	var execur sqlx.Execer = s.db
	if tx != nil {
		execur = tx
	}

	ts, err := s.parseTriggerExpr(t.Type, t.Expr)
	if err != nil {
		ctx.Log().Warnf("parse trigger expr failed, err= %v", err)
		return 0, err
	}

	r, err := execur.Exec(`INSERT INTO triggers (name, type, creator, priority, summary, release_branch, expr, ts, plan_id, plan_ids, variables, active)
	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		t.Name,
		t.Type,
		t.Creator,
		t.ExecPriority,
		t.Summary,
		t.ReleaseBranch,
		t.Expr,
		ts,
		t.PlanID,
		t.PlanIDs,
		t.VariableBinds,
		t.Active,
	)
	if err != nil {
		ctx.Log().With("refPlanID", t.PlanID).Errorf(err, "CreateTrigger failed")
		return 0, err
	}
	id, err := r.LastInsertId()
	if err != nil {
		ctx.Log().Errorf(err, "CreateTrigger get last insert id failed")
		return 0, err
	}
	return model.ID(id), nil
}

func (s *triggerService) AddPlanIDs(ctx utils.Context, id model.ID, newPlanIDs []model.ID) error {
	log := ctx.Log().With("trigger_id", id)

	var trigger model.TestTrigger
	err := s.db.Transaction(ctx.Request().Context(), func(tx *sqlx.Tx) error {
		// 1. Get trigger planIDs
		if err := tx.Get(&trigger,
			"SELECT type, plan_ids FROM triggers WHERE id = ?",
			id,
		); err != nil {
			if sql.IsErrNoRows(err) {
				log.Warnf("no trigger has id: %d", id)
				return serror.New(http.StatusNotFound)
			}
			log.Errorf(err, "failed to get trigger planIDs")
			return err
		}

		if trigger.Type != model.TriggerTypeRelease {
			return serror.NewMsg(http.StatusBadRequest, "unexpected trigger id: type != 'RELEASE'")
		}

		// 2. Set plan_ids for the trigger.
		planIDs := trigger.PlanRef.PlanIDs

		// Filter out the planIDs already in the trigger.PlanRef
		newPlanIDs = utils.Except(newPlanIDs, planIDs)
		if len(newPlanIDs) == 0 {
			// NO-OP
			return nil
		}

		cnt := len(planIDs) + len(newPlanIDs)
		allPlanIDs := utils.Union(planIDs, newPlanIDs)
		if len(allPlanIDs) != cnt {
			return serror.NewMsg(http.StatusBadRequest, "unexpected newPlanIDs: duplicated planID")
		}

		_, err := tx.Exec(`UPDATE triggers SET plan_ids=?, last_triggered_at=? WHERE id=?`,
			utils.FormatCSVID(allPlanIDs),
			time.Now().Unix(),
			id,
		)
		if err != nil {
			log.Errorf(err, "update trigger's planIDs failed")
			return err
		}

		return nil
	})

	if err != nil {
		return err
	}

	// 4. Create corresponding plan-executions.
	var (
		planExecs []*planExecCreateOpts
		priority  = model.ExecPriorityHigh
	)
	timeNow := time.Now()

	for _, pid := range newPlanIDs {
		planExecs = append(planExecs, &planExecCreateOpts{
			PlanID:    model.ID(pid),
			TriggerID: id,
			CreatorID: ctx.User().ID,
			Priority:  priority,
			CreatedAt: timeNow,
		})
	}

	if _, err := s.planExecSvc.batchCreate(ctx, nil, planExecs); err != nil {
		log.Errorf(err, "failed create plan-execs")
		return err
	}

	return err
}

func (s *triggerService) Delete(ctx utils.Context, id model.ID) error {
	log := ctx.Log().With("id", id)

	creator, err := s.getCreator(ctx, id)
	if err != nil {
		log.Errorf(err, "GetTriggerCreator failed")
		return err
	} else if creator == nil {
		return nil
	} else if *creator != ctx.User().ID {
		return serror.NewMsg(http.StatusForbidden, "the creator is not you")
	}

	query, args, _ := sqlx.In("DELETE FROM triggers WHERE id=?", id)
	_, err = s.db.Exec(query, args...)
	if err != nil {
		log.Errorf(err, "DeleteTrigger failed")
		return err
	}

	return nil
}

// parseTriggerExpr parses `TestTrigger.Expr` to UNIX Timestamp or cron string.
func (s *triggerService) parseTriggerExpr(ty model.TriggerType, expr string) (triggerTS int64, err error) {
	const errMsg = "parse trigger expr failed, only RFC3339 and cron string are accepted"

	switch ty {
	case model.TriggerTypeOnce, model.TriggerTypeRelease:
		if expr == "" {
			// will be trigger immediately if expr is empty
			triggerTS = 0
		} else {
			time_, err := time.Parse(time.RFC3339, expr)
			if err != nil {
				return 0, serror.NewMsg(http.StatusBadRequest, errMsg)
			}
			triggerTS = time_.Unix()
		}

		return triggerTS, nil
	case model.TriggerTypeCron:
		if expr == "" {
			// will never be triggered if cron schedule is empty
			triggerTS = math.MaxInt64
		} else {
			var sched cron.Schedule
			if expr == "@weekly" {
				// run @weekly test at 00:00 on Saturday
				sched, err = cron.ParseStandard("0 0 * * 6")
			} else {
				sched, err = cron.ParseStandard(expr)
			}

			if err != nil {
				return 0, serror.NewMsg(http.StatusBadRequest, errMsg)
			}
			triggerTS = sched.Next(time.Now()).Unix()
		}

		return triggerTS, nil
	}

	return 0, serror.New(http.StatusBadRequest, fmt.Errorf("unknown trigger type: %s", ty))
}