package workload

import (
	"context"
	"fmt"
	"os"
	"strconv"
	"time"

	"github.com/chromedp/chromedp"
	"github.com/tidbcloud/pov-service/internal/common/constant"
	"github.com/tidbcloud/pov-service/internal/common/errors"
	"github.com/tidbcloud/pov-service/internal/common/options"
	"github.com/tidbcloud/pov-service/internal/common/params"
	"github.com/tidbcloud/pov-service/internal/pkg/icontext"
	"github.com/tidbcloud/pov-service/internal/pkg/log"
	"github.com/tidbcloud/pov-service/pkg/browser"

	"github.com/tidbcloud/pov-service/internal/config"
	"github.com/tidbcloud/pov-service/internal/model"
)

func (s *svcImpl) RunWorkloadTest(
	ctx context.Context,
	param params.RunWorkload,
	startedBy string,
) (model.WorkloadTest, error) {
	tmpl, err := s.GetWorkload(ctx, param.WorkloadID)
	if err != nil {
		return model.WorkloadTest{}, err
	}
	for i := range param.GetParams() {
		param.GetParams()[i].IsUserDefined = true
	}

	// Set max_connection_number
	maxConParam := param.GetParam(string(constant.MaxConnectionNumber))
	if maxConParam == nil {
		return model.WorkloadTest{}, fmt.Errorf("max_connection_number is required")
	}
	maxCon, err := strconv.Atoi(maxConParam.Value)
	if err != nil {
		return model.WorkloadTest{}, err
	}
	trafficParams, err := s.getTrafficMeta(ctx, param.WorkloadID, maxCon)
	if err != nil {
		return model.WorkloadTest{}, err
	}
	for _, tp := range trafficParams {
		param.SetParam(tp.Name, tp.Value)
	}

	isAuto := true
	if param.IsAuto != nil {
		isAuto = *param.IsAuto
	}
	defaultWlName := tmpl.DefaultParamValue(string(constant.WorkloadName))
	if defaultWlName != nil {
		param.SetParam(string(constant.WorkloadName), defaultWlName.(string))
	}

	test := model.NewWorkloadTest(
		param.GetTestName(),
		param.WorkloadID,
		param.GetParams(),
		isAuto,
		startedBy,
	)
	test.WorkloadName = tmpl.Name

	test, err = s.workloadTestRepo.Create(ctx, test)
	if err == nil {
		if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
			ctx,
			startedBy,
			model.OperateTypeCreateWorkloadTest,
			test.ID,
		)); logErr != nil {
			// Don't return error for audit log for robust, or it should be included in a txn
			log.Log.Errorf(logErr, "failed to create audit log for workload test %s", test.ID)
		}
	}

	return test, err
}

func (s *svcImpl) GetWorkloadTest(ctx context.Context, ID string) (model.WorkloadTest, error) {
	return s.workloadTestRepo.Get(ctx, ID)
}

func (s *svcImpl) CreateWorkloadTest(
	ctx context.Context,
	params params.RunWorkload,
	startedBy string,
) (model.WorkloadTest, error) {
	tmpl, err := s.GetWorkload(ctx, params.WorkloadID)
	if err != nil {
		return model.WorkloadTest{}, err
	}

	test := model.NewWorkloadTest(
		params.GetTestName(), params.WorkloadID, params.GetParams(), false, startedBy,
	)
	test.WorkloadName = tmpl.Name

	test, err = s.workloadTestRepo.Create(ctx, test)
	if err == nil {
		if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
			ctx,
			startedBy,
			model.OperateTypeCreateWorkloadTest,
			test.ID,
		)); logErr != nil {
			// Don't return error for audit log for robust, or it should be included in a txn
			log.Log.Errorf(logErr, "failed to create audit log for workload test %s", test.ID)
		}
	}

	return test, err
}

func (s *svcImpl) ListWorkloadTests(ctx context.Context, param params.ListWorkloadTests) ([]model.WorkloadTest, error) {
	opt := param.ToListOption()
	return s.workloadTestRepo.List(ctx, opt)
}

func (s *svcImpl) StopWorkloadTest(
	ctx context.Context,
	ID string) (model.WorkloadTest, error) {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return model.WorkloadTest{}, fmt.Errorf("user email not found")
	}

	test, err := s.workloadTestRepo.Get(ctx, ID)
	if err != nil {
		return model.WorkloadTest{}, err
	}
	if test.Status != model.ExecStatusRunning {
		return model.WorkloadTest{}, fmt.Errorf("workload test %s is not running", ID)
	}
	test.Status = model.ExecStatusStopping

	test, err = s.workloadTestRepo.Update(ctx, test, options.ShouldWait())

	if err == nil {
		if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
			ctx,
			userEmail,
			model.OperateTypeStopWorkloadTest,
			test.ID,
		)); logErr != nil {
			log.Log.Errorf(logErr, "failed to create audit log for stopping workload test %s", test.ID)
		}
	}

	return test, err

}

func (s *svcImpl) ScaleCluster(ctx context.Context, ID string, replicas int) error {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return fmt.Errorf("user email not found")
	}

	test, err := s.workloadTestRepo.Get(ctx, ID)
	if err != nil {
		return err
	}
	if test.Status != model.ExecStatusRunning {
		return fmt.Errorf("workload test %s is not running", ID)
	}
	if test.Cluster == nil {
		return fmt.Errorf("workload test %s has no cluster", ID)
	}

	err = s.Workload.ScaleCluster(ctx, *test.Cluster, replicas)
	if err == nil {
		if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
			ctx,
			userEmail,
			model.OperateTypeScaleCluster,
			test.ID,
		).SetDetailMap(map[string]string{
			"replicas": strconv.Itoa(replicas),
		}),
		); logErr != nil {
			log.Log.Errorf(logErr, "failed to create audit log for scaling workload test %s", test.ID)
		}
	}

	return err

}

func (s *svcImpl) UpdateClusterRU(
	ctx context.Context, testID string, minRU, maxRU int64) error {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return fmt.Errorf("user email not found")
	}

	test, err := s.workloadTestRepo.Get(ctx, testID)
	if err != nil {
		return err
	}
	if test.Status != model.ExecStatusRunning {
		return fmt.Errorf("workload test %s is not running", testID)
	}
	if test.Cluster == nil {
		return fmt.Errorf("workload test %s has no cluster", testID)
	}

	// Validate RU values
	if minRU <= 0 || maxRU <= 0 {
		return fmt.Errorf("minRU and maxRU must be greater than 0")
	}
	if minRU > maxRU {
		return fmt.Errorf("minRU cannot be greater than maxRU")
	}

	// Store old values for audit log
	oldMinRU := test.Cluster.MinRU
	oldMaxRU := test.Cluster.MaxRU
	if oldMinRU == minRU && oldMaxRU == maxRU {
		log.Log.Infof("UpdateClusterRU invalid: minRU/maxRU is already %d/%d", minRU, maxRU)
		return nil
	}

	// Call RMS API to update cluster RU
	err = s.Workload.UpdateClusterRU(ctx, *test.Cluster, minRU, maxRU)
	if err != nil {
		return fmt.Errorf("failed to update cluster RU: %w", err)
	}

	// Update cluster MinRU and MaxRU in database
	test.Cluster.MinRU = minRU
	test.Cluster.MaxRU = maxRU
	maxRUstr := strconv.FormatInt(maxRU, 10)
	minRUstr := strconv.FormatInt(minRU, 10)
	test.Params.Set(string(constant.MaxRU), maxRUstr)
	test.Params.Set(string(constant.MinRU), minRUstr)

	_, err = s.workloadTestRepo.Update(ctx, test, options.ShouldWait())
	if err == nil {
		msg := icontext.OperateMessageFromCtx(ctx)
		// Create audit log
		if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
			ctx,
			userEmail,
			model.OperateTypeUpdateClusterRU,
			test.ID,
		).SetUpdateDetails(
			model.NewUpdateTuple(string(constant.MinRU), strconv.FormatInt(oldMinRU, 10), strconv.FormatInt(minRU, 10)),
			model.NewUpdateTuple(string(constant.MaxRU), strconv.FormatInt(oldMaxRU, 10), strconv.FormatInt(maxRU, 10)),
		).AppendMessage(msg),
		); logErr != nil {
			log.Log.Errorf(logErr, "failed to create audit log for updating cluster RU for workload test %s", test.ID)
		}
	}

	return err
}

func (s *svcImpl) InjectFailure(
	ctx context.Context,
	testID string,
	duration string,
) error {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return fmt.Errorf("user email not found")
	}

	test, err := s.workloadTestRepo.Get(ctx, testID)
	if err != nil {
		return err
	}
	if test.Status != model.ExecStatusRunning {
		return fmt.Errorf("workload test %s is not running", testID)
	}
	if test.Cluster == nil {
		return fmt.Errorf("workload test %s has no cluster", testID)
	}

	// Call RMS API to inject failure
	err = s.Workload.InjectFailure(ctx, *test.Cluster, duration)
	if err != nil {
		return fmt.Errorf("failed to inject failure: %w", err)
	}

	// Create audit log
	msg := icontext.OperateMessageFromCtx(ctx)
	if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
		ctx,
		userEmail,
		model.OperateTypeInjectFailure,
		test.ID,
	).SetDetailMap(map[string]string{
		"failure_injection_duration":  duration,
		"failure_injection_component": "tidb", // currently only tidb is supported, so hardcode it now
	}).AppendMessage(msg),
	); logErr != nil {
		log.Log.Errorf(logErr, "failed to create audit log for injecting failure for workload test %s", test.ID)
	}

	return nil
}

func (s *svcImpl) RollbackFailure(
	ctx context.Context,
	testID string,
) error {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return fmt.Errorf("user email not found")
	}

	test, err := s.workloadTestRepo.Get(ctx, testID)
	if err != nil {
		return err
	}
	if test.Status != model.ExecStatusRunning {
		return fmt.Errorf("workload test %s is not running", testID)
	}
	if test.Cluster == nil {
		return fmt.Errorf("workload test %s has no cluster", testID)
	}

	// Call RMS API to rollback failure
	err = s.Workload.RollbackFailure(ctx, *test.Cluster)
	if err != nil {
		return fmt.Errorf("failed to rollback failure: %w", err)
	}

	// Create audit log
	msg := icontext.OperateMessageFromCtx(ctx)
	if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
		ctx,
		userEmail,
		model.OperateTypeRollbackFailure,
		test.ID,
	).SetDetailMap(map[string]string{
		"failure_rollback_component": "tidb", // currently only tidb is supported, so hardcode it now
	}).AppendMessage(msg),
	); logErr != nil {
		log.Log.Errorf(logErr, "failed to create audit log for rolling back failure for workload test %s", test.ID)
	}

	return nil
}

func (s *svcImpl) UpdateTrafficMaxConn(
	ctx context.Context,
	testID string,
	maxConNumber int,
) error {
	log.Log.Infof("UpdateTrafficMaxConn: %s, %d", testID, maxConNumber)
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return fmt.Errorf("user email not found")
	}

	tx, err := s.workloadTestRepo.BeginTx(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	test, err := s.workloadTestRepo.GetAndLockForUpdateInTx(ctx, tx, testID, options.ShouldWait())
	if err != nil {
		return err
	}
	if test.Phase != model.WorkloadTestPhaseRunWorkload {
		return fmt.Errorf("MaxConnectionNumber can only be updated when workload test is running")
	}

	workloadTask := test.CurrentPhaseTask()
	if workloadTask.Phase != model.WorkloadTaskPhaseRunning {
		return fmt.Errorf("MaxConnectionNumber can only be updated when workload task is running")
	}

	oldValue := ""
	oldValuePram, ok := test.GetParam(string(constant.MaxConnectionNumber))
	if ok {
		oldValue = oldValuePram.(string)
	}
	if oldValue == strconv.Itoa(maxConNumber) {
		log.Log.Infof("UpdateTrafficMaxConn invalid: MaxConnectionNumber is already %d", maxConNumber)
		return nil
	}
	test.Params.Set(string(constant.MaxConnectionNumber), strconv.Itoa(maxConNumber))

	trafficParams, err := s.getTrafficMeta(ctx, test.WorkloadID, maxConNumber)
	if err != nil {
		return err
	}
	for _, param := range trafficParams {
		test.Params.Set(param.Name, param.Value)
	}

	workloadTask.Phase = model.WorkloadTaskPhaseUpdating

	err = s.workloadTestRepo.UpdateInTx(ctx, tx, test)
	if err == nil {
		msg := icontext.OperateMessageFromCtx(ctx)

		if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
			ctx,
			userEmail,
			model.OperateTypeConfigTrafficConn,
			test.ID,
		).SetUpdateDetail(
			string(constant.MaxConnectionNumber), oldValue, strconv.Itoa(maxConNumber),
		).AppendMessage(msg),
		); logErr != nil {
			log.Log.Errorf(logErr, "failed to create audit log for scaling workload test %s", test.ID)
		}
	}

	return tx.Commit()
}

func (s *svcImpl) UpdateWorkloadTest(
	ctx context.Context,
	testID string,
	params params.UpdateWorkloadTest,
) error {
	ctx = icontext.NewOperateAtContext(ctx, time.Now().UTC())
	if err := s.UpdateTrafficMaxConn(ctx, testID, params.MaxConn); err != nil {
		return err
	}

	return s.UpdateClusterRU(ctx, testID, params.MinRU, params.MaxRU)
}

func (s *svcImpl) SampleDBs(
	ctx context.Context,
	testID string,
	options model.SampleDBOptions) (model.DBSampleResult, error) {
	test, err := s.workloadTestRepo.Get(ctx, testID)
	if err != nil {
		return model.DBSampleResult{}, err
	}
	if !test.IsClusterCreated() {
		return model.DBSampleResult{}, fmt.Errorf("cluster creation is not finished")
	}
	cluster := test.Cluster
	if cluster == nil {
		return model.DBSampleResult{}, fmt.Errorf("cluster not found")
	}
	return s.Workload.SampleDBs(ctx, *cluster, options)
}

func (s *svcImpl) SampleData(
	ctx context.Context,
	testID string,
	options model.SampleDataOptions) (model.TableQueryResult, error) {
	test, err := s.workloadTestRepo.Get(ctx, testID)
	if err != nil {
		return model.TableQueryResult{}, err
	}
	if !test.IsClusterCreated() {
		return model.TableQueryResult{}, fmt.Errorf("cluster creation is not finished")
	}
	cluster := test.Cluster
	if cluster == nil {
		return model.TableQueryResult{}, fmt.Errorf("cluster not found")
	}
	return s.Workload.SampleData(ctx, *cluster, options)
}

func (s *svcImpl) ListAuditLog(
	ctx context.Context,
	testID string,
) ([]model.AuditLog, error) {
	_, err := s.workloadTestRepo.Get(ctx, testID)
	if err != nil {
		return nil, err
	}

	return s.auditLogRepo.List(
		ctx,
		options.NewListAuditLogOption().
			SetOrderBy("operate_at").
			AddFilter(options.WorkloadTestObjectTypes()).
			AddFilter(options.ObjectID(testID)),
	)
}

func (s *svcImpl) TriggerCreateCluster(
	ctx context.Context,
	testID string) (model.WorkloadTest, error) {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return model.WorkloadTest{}, fmt.Errorf("user email not found")
	}

	tx, err := s.workloadTestRepo.BeginTx(ctx)
	if err != nil {
		return model.WorkloadTest{}, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback()

	test, err := s.workloadTestRepo.GetAndLockForUpdateInTx(ctx, tx, testID)
	if err != nil {
		return model.WorkloadTest{}, err
	}

	if test.Phase != model.WorkloadTestPhasePending || test.IsAuto {
		return model.WorkloadTest{}, errors.NewInvalidArgument("workload test is not ready to create cluster")
	}

	log.Log.Infof("Trigger creating cluster for test %s", testID)

	test.ProceedCreateCluster()

	if err := s.workloadTestRepo.UpdateInTx(ctx, tx, test); err != nil {
		log.Log.Errorf(err, "failed to update workload test %s to create cluster phase", testID)
		return model.WorkloadTest{}, fmt.Errorf("failed to update workload test %s to create cluster phase: %w", testID, err)
	}

	// Commit the transaction - this releases the lock
	if err := tx.Commit(); err != nil {
		log.Log.Errorf(err, "failed to commit transaction")
		return model.WorkloadTest{}, fmt.Errorf("failed to commit transaction: %w", err)
	}

	if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
		ctx,
		userEmail,
		model.OperateTypePrepareData,
		test.ID,
	)); logErr != nil {
		log.Log.Errorf(logErr, "failed to create audit log for triggering creating cluster for %s", test.ID)
	}

	return test, nil
}

func (s *svcImpl) TriggerPrepareData(
	ctx context.Context,
	testID string) (model.WorkloadTest, error) {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return model.WorkloadTest{}, fmt.Errorf("user email not found")
	}

	tx, err := s.workloadTestRepo.BeginTx(ctx)
	if err != nil {
		return model.WorkloadTest{}, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback()

	test, err := s.workloadTestRepo.GetAndLockForUpdateInTx(ctx, tx, testID)
	if err != nil {
		return model.WorkloadTest{}, err
	}

	if ok := test.Ready2PrepareData(); !ok || test.IsAuto {
		return model.WorkloadTest{}, errors.NewInvalidArgument("workload test is not ready to prepare data")
	}

	log.Log.Infof("Trigger prepare data for test %s", testID)

	test.ProceedPrepareData()

	if err := s.workloadTestRepo.UpdateInTx(ctx, tx, test); err != nil {
		log.Log.Errorf(err, "failed to update workload test %s to prepare phase", testID)
		return model.WorkloadTest{}, fmt.Errorf("failed to update workload test %s to prepare phase: %w", testID, err)
	}

	// Commit the transaction - this releases the lock
	if err := tx.Commit(); err != nil {
		log.Log.Errorf(err, "failed to commit transaction")
		return model.WorkloadTest{}, fmt.Errorf("failed to commit transaction: %w", err)
	}

	if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
		ctx,
		userEmail,
		model.OperateTypePrepareData,
		test.ID,
	)); logErr != nil {
		log.Log.Errorf(logErr, "failed to create audit log for triggering preparing data for %s", test.ID)
	}

	return test, nil
}

func (s *svcImpl) TriggerRunTraffic(
	ctx context.Context,
	testID string) (model.WorkloadTest, error) {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return model.WorkloadTest{}, fmt.Errorf("user email not found")
	}

	tx, err := s.workloadTestRepo.BeginTx(ctx)
	if err != nil {
		return model.WorkloadTest{}, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback()

	test, err := s.workloadTestRepo.GetAndLockForUpdateInTx(ctx, tx, testID)
	if err != nil {
		return model.WorkloadTest{}, err
	}

	if ok := test.Ready2RunTraffic(); !ok || test.IsAuto {
		return model.WorkloadTest{}, errors.NewInvalidArgument("workload test is not ready to run traffic")
	}

	log.Log.Infof("Trigger running traffic for test %s", testID)

	test.ProceedRunTraffic()

	if err := s.workloadTestRepo.UpdateInTx(ctx, tx, test); err != nil {
		log.Log.Errorf(err, "failed to update workload test %s to run traffic phase", testID)
		return model.WorkloadTest{}, fmt.Errorf("failed to update workload test %s to run traffic phase: %w", testID, err)
	}

	// Commit the transaction - this releases the lock
	if err := tx.Commit(); err != nil {
		log.Log.Errorf(err, "failed to commit transaction")
		return model.WorkloadTest{}, fmt.Errorf("failed to commit transaction: %w", err)
	}

	if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
		ctx,
		userEmail,
		model.OperateTypeRunTraffic,
		test.ID,
	)); logErr != nil {
		log.Log.Errorf(logErr, "failed to create audit log for triggering running traffic for %s", test.ID)
	}

	return test, nil
}

func (s *svcImpl) PreserveWorkloadSample(
	ctx context.Context,
	workloadID int64,
	cluster model.TiDBCluster,
) error {
	// Begin txn to save the latest sample and delete outdated samples
	tx, err := s.workloadSampleRepo.BeginTx(ctx)
	if err != nil {
		return fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback()

	// Check if workload sample already exists(preserved in 24h by default)
	legacySample, err := s.workloadSampleRepo.LockWorkloadSchemaInTx(ctx, tx, workloadID)
	if err != nil {
		return err
	}
	if legacySample.DBCount > 0 && !legacySample.IsOutdated() {
		log.Log.Infof("Preserved sample data for workload %d already exists", workloadID)
		return nil
	}

	log.Log.Infof("Preserving sample data for workload %d", workloadID)
	sampleResult, err := s.Workload.SampleDBs(ctx, cluster, model.SampleDBOptions{
		DBLimit:    defaultPreserveDBLimit,
		TableLimit: defaultPreserveTableLimit,
		WithData:   true,
		RowOptions: model.SampleDataOptions{
			Limit: defaultPreserveRowLimit,
		},
	})
	if err != nil {
		return err
	}
	if len(sampleResult.Databases) == 0 {
		return fmt.Errorf("no databases found")
	}

	newVer, err := s.workloadSampleRepo.CreateInTx(ctx, tx, workloadID, sampleResult.Databases)
	if err != nil {
		return err
	}
	if err = s.workloadSampleRepo.DeleteOutdatedInTx(ctx, tx, workloadID, newVer); err != nil {
		return err
	}

	return tx.Commit()
}

// DownloadPDF returns the PDF file of the workload
// It's a sync call now which may take few seconds to complete.
// And can be optimized with predownloading and uploading to S3 instead.
// TODO: preserve PDF file to S3 after workload test execution to provide better performance and capability
func (s *svcImpl) DownloadTestPDF(
	ctx context.Context,
	testID string,
	opts browser.PrintPDFOptions,
) (*os.File, error) {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return nil, fmt.Errorf("user email not found")
	}

	// Extract localStorage from context before creating allocator
	storages := browser.LocalStorageFromContext(ctx)

	allocator, cancel := browser.NewHeadlessAllocator(ctx)
	defer cancel()

	timeoutCtx, timeoutCancel := allocator.NewTimeOutContext(40 * time.Second)
	defer timeoutCancel()

	// Create a single chromedp context for both login and PDF generation
	browserCtx, browserCancel := chromedp.NewContext(timeoutCtx)
	defer browserCancel()

	// Re-inject localStorage into the chromedp context if available
	if len(storages) > 0 {
		browserCtx = browser.NewLocalStorageContext(browserCtx, storages)
	}

	if err := browser.Navigate(
		browserCtx,
		config.PoVLocalStorageSettingPageAddress()); err != nil {
		return nil, err
	}

	// Generate PDF using the same browser context (with session preserved)
	pdfBytes, err := browser.PrintPDF(browserCtx, browser.PrintPDFOptions{
		Address:       opts.Address,
		ReadySelector: opts.ReadySelector,
		Width:         opts.Width,
		Height:        opts.Height,
	})
	if err != nil {
		return nil, err
	}

	tmpfile, err := os.CreateTemp("", "tiworkload_test_report_")
	if err != nil {
		return nil, err
	}
	err = os.WriteFile(tmpfile.Name(), pdfBytes, 0644)

	if err == nil {
		if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
			ctx,
			userEmail,
			model.OperateTypeDownloadWorkloadTestPDF,
			testID,
		)); logErr != nil {
			// Don't return error for audit log for robust, or it should be included in a txn
			log.Log.Errorf(logErr, "failed to create audit log for downloading pdf of test %s", testID)
		}
	}

	return tmpfile, err
}

func (w *svcImpl) ListScriptTemplates(ctx context.Context) ([]model.Script, error) {
	return nil, nil
}

func (w *svcImpl) ActivateScriptTemplate(ctx context.Context, id int64) error {
	return nil
}

func (s *svcImpl) InstructWorkloadTest(
	ctx context.Context,
	testID string,
	params params.InstructWorkloadTest,
) error {
	userEmail := icontext.UserEmailFromCtx(ctx)
	if userEmail == "" {
		return fmt.Errorf("user email not found")
	}

	test, err := s.GetWorkloadTest(ctx, testID)
	if err != nil {
		return err
	}
	if params.SQLStatement != "" {
		err = s.ExecuteSQL(ctx, *test.Cluster, params.SQLStatement)
		if err == nil {
			msg := icontext.OperateMessageFromCtx(ctx)

			if logErr := s.auditLogRepo.Create(ctx, model.NewOperationLog(
				ctx,
				userEmail,
				model.OperateTypeExecuteSQL,
				testID,
			).SetDetailMap(map[string]string{
				string(constant.SQLStatement): params.SQLStatement,
			}).AppendMessage(msg),
			); logErr != nil {
				// Don't return error for audit log for robust, or it should be included in a txn
				log.Log.Errorf(logErr, "failed to create audit log for downloading pdf of test %s", testID)
			}
		}

	}

	return err
}

func (w *svcImpl) getTrafficMeta(
	ctx context.Context,
	workloadID int64,
	maxConnNumber int,
) (ret model.WorkloadParams, err error) {
	calTrafficClientMeta := func(
		maxCon int,
		// defaultCon is the recommended threads of single clients
		defaultCon int,
	) (count int, threads int) {
		if defaultCon > maxCon {
			count = 1
			threads = maxCon
			return
		}

		// calculate the client count
		count = maxCon / defaultCon

		// if there is remain, add one client
		remain := maxCon % defaultCon
		if remain > 0 {
			count++
		}
		// Just ignore the remain threads
		threads = maxCon / count

		return
	}
	ret = make(model.WorkloadParams, 0)

	// Set max_connection_number
	workloadTemplate, err := w.GetWorkload(ctx, workloadID)
	if err != nil {
		return
	}
	defaultMaxConValue := workloadTemplate.DefaultParamValue(string(constant.MaxConnectionNumber))
	if defaultMaxConValue == nil {
		return
	}
	var defaultMaxCon int
	switch defaultMaxConValue := defaultMaxConValue.(type) {
	case int:
		defaultMaxCon = defaultMaxConValue
	case string:
		defaultMaxCon, err = strconv.Atoi(defaultMaxConValue)
		if err != nil {
			return
		}
	}

	count, threads := calTrafficClientMeta(maxConnNumber, defaultMaxCon)
	ret.Set(string(constant.ClientCount), strconv.Itoa(count))
	ret.Set(string(constant.Thread), strconv.Itoa(threads))

	return
}