package test

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/schrodinger/manager/operator/proc/k8s"
	"github.com/pingcap/schrodinger/pkg/storage"
	"github.com/pingcap/schrodinger/pkg/types"
	"github.com/pingcap/schrodinger/util"
)

const maxRetryCount = 5

// Test is new test
type Test struct {
	Proc *k8s.Proc

	storage *storage.Storage
	process *types.TestProcess

	closed int32

	operator Operator
}

// Operator provides a special interface for control test proc
type Operator interface {
	// GetProcByName gets k8s proc by name
	GetProcByName(namespace string, name string) (*k8s.Proc, error)
	// StartProc starts k8s
	StartProc(ctx context.Context, p *k8s.Proc) error
	// StopProc stops k8s
	StopProc(ctx context.Context, p *k8s.Proc) error
	// DestroyProc deletes proc from k8s
	DestroyProc(p *k8s.Proc) error
	// Status returns k8s.status
	Status(p *k8s.Proc) k8s.PState
	// GetProcLog gets log
	GetProcLog(p *k8s.Proc, lines int64) (string, error)
	// CheckProcIsStarted returns true if it has been started
	CheckProcIsStarted(p *k8s.Proc) (bool, error)
}

// NewTest creates test
func NewTest(
	testProc *types.TestProcess,
	s *storage.Storage,
	operator Operator,
) *Test {
	return &Test{
		Proc:     createTestProc(testProc, testProc.Namespace),
		process:  testProc,
		storage:  s,
		operator: operator,
	}
}

// Start starts test
func (t *Test) Start(ctx context.Context) error {
	defer t.store()
	log.Infof("start test %s in [namespace %s]", t.Name(), t.Namespace())
	atomic.StoreInt32(&t.closed, 0)
	t.process.Ctime = time.Now().Format(util.TimeFormat)
	t.SetState(types.TestStateCreating)
	err := t.operator.StartProc(ctx, t.Proc)
	if err != nil {
		t.SetState(types.TestStateFailed)
		return errors.Trace(err)
	}

	t.SetState(types.TestStateRunning)

	log.Infof("start test %s in [namespace %s] success", t.Name(), t.Namespace())
	return nil
}

// Stop stops test
func (t *Test) Stop(ctx context.Context) error {
	defer t.store()
	log.Infof("stop test %s in [namespace %s]", t.Name(), t.Namespace())
	if !atomic.CompareAndSwapInt32(&t.closed, 0, 1) {
		return errors.MethodNotAllowedf("case %s was already closed", t.Name())
	}

	err := t.operator.StopProc(ctx, t.Proc)
	if err != nil {
		log.Errorf("stop %s in [namespace %s] failed %v", t.Name(), t.Namespace(), err)
		return errors.Trace(err)
	}

	if !t.IsError() {
		t.SetState(types.TestStateStopped)
	}

	log.Infof("stop test %s in [namespace %s] success", t.Name(), t.Namespace())
	return nil
}

// Destroy destroys test
func (t *Test) Destroy(ctx context.Context) error {
	defer t.store()
	err := t.operator.DestroyProc(t.Proc)
	if err != nil {
		return errors.Trace(err)
	}

	t.SetState(types.TestStateDestroyed)
	return nil
}

// GetLogs gets logs for test
func (t *Test) GetLogs(lines int64) string {
	defer t.store()
	caseLog, err := t.operator.GetProcLog(t.Proc, lines)
	if err != nil {
		caseLog = fmt.Sprintf("[namespace] %s case %s get log error %v", t.Namespace(), t.Name(), err)
	}

	return util.FatalLogAndBelow(caseLog)
}

func (t *Test) isClosed() bool {
	return atomic.LoadInt32(&t.closed) == 1
}

// SyncStatus syncs status
func (t *Test) SyncStatus(ctx context.Context) {
	defer t.store()
	err := util.RetryOnError(ctx, maxRetryCount, func() error {
		t.Proc.State = t.operator.Status(t.Proc)
		return nil
	})
	if err != nil {
		log.Errorf("[%s][%s] error %s", t.Proc.Namespace, t.Proc.Name, err.Error())
		t.Proc.State = k8s.ProcTimeout
	}

	log.Infof("[%s][%s] status is %s", t.Proc.Namespace, t.Proc.Name, t.Proc.State)
	t.process.Utime = time.Now().Format(util.TimeFormat)

	switch t.Proc.State {
	case k8s.ProcCreatePodError, k8s.ProcFatal, k8s.ProcKilled,
		k8s.ProcKilling, k8s.ProcStartError, k8s.ProcUnknown,
		k8s.ProcTimeout, k8s.ProcErrorExited:
		t.SetState(types.TestStateFailed)
	case k8s.ProcNormalExited:
		t.SetState(types.TestStateFinished)
	}
}

func (t *Test) store() {
	if err := t.storage.UpdateTestProcess(t.process); err != nil {
		log.Errorf("failed to update %s test process %s, error: %v", t.Namespace(), t.Name(), err)
	}
}

// SetReport sets report
func (t *Test) SetReport(report *types.TestReport) {
	defer t.store()
	t.process.Report = report
}

// IsError determines is case error
func (t *Test) IsError() bool {
	return t.process.Status == types.TestStateFailed
}

// IsDestroy return true if the test is destroyed
func (t *Test) IsDestroy() bool {
	return t.process.Status == types.TestStateDestroyed
}

// Identify returns test identify
func (t *Test) Identify() int64 {
	return t.process.ID
}

// IsFinished determines is test if finished
func (t *Test) IsFinished() bool {
	return t.process.Status == types.TestStateFinished
}

// MarkRemoved mark test removed
func (t *Test) MarkRemoved() {
	t.process.Mark = types.MarkRemoved
}

// MarkOnline mark test online
func (t *Test) MarkOnline() {
	t.process.Mark = types.MarkOnline
}

// MarkOffline marks test offline
func (t *Test) MarkOffline() {
	t.process.Mark = types.MarkOffline
}

// Mark returns test mark
func (t *Test) Mark() types.Mark {
	return t.process.Mark
}

// IsRemoved returns true if test is removed
func (t *Test) IsRemoved() bool {
	return t.process.Mark == types.MarkRemoved
}

// IsOnline returns true if test is online
func (t *Test) IsOnline() bool {
	return t.process.Mark == types.MarkOnline
}

// IsOffline returns true if test is offline
func (t *Test) IsOffline() bool {
	return t.process.Mark == types.MarkOffline
}

// SetState sets state of test
func (t *Test) SetState(state types.TestState) {
	defer t.store()
	t.process.SetState(state)
}

// Status returns test status
func (t *Test) Status() types.TestState {
	return t.process.Status
}

// ProcStatus return proc status
func (t *Test) ProcStatus() k8s.PState {
	return t.Proc.State
}

// Name returns test name
func (t *Test) Name() string {
	return t.process.Name
}

// Namespace returns test namespace
func (t *Test) Namespace() string {
	return t.process.Namespace
}