package mysql

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"time"

	"github.com/actiontech/dtle/driver/oracle/extractor"

	"github.com/actiontech/dtle/g"

	"github.com/actiontech/dtle/driver/common"
	"github.com/actiontech/dtle/driver/kafka"
	mysql "github.com/actiontech/dtle/driver/mysql"
	"github.com/armon/go-metrics"
	"github.com/pkg/errors"

	"github.com/hashicorp/go-hclog"
	"github.com/hashicorp/nomad/plugins/drivers"
)

type taskHandle struct {
	logger g.LoggerType

	// stateLock syncs access to all fields below
	stateLock sync.RWMutex

	taskConfig  *drivers.TaskConfig
	procState   drivers.TaskState
	startedAt   time.Time
	completedAt time.Time
	exitResult  *drivers.ExitResult

	runner DriverHandle

	ctx        context.Context
	cancelFunc context.CancelFunc
	waitCh     chan *drivers.ExitResult
	stats      *common.TaskStatistics

	driverConfig *common.MySQLDriverConfig
	shutdown     bool
}

func newDtleTaskHandle(ctx context.Context, logger g.LoggerType, cfg *drivers.TaskConfig, state drivers.TaskState, started time.Time) *taskHandle {
	h := &taskHandle{
		logger:      logger,
		stateLock:   sync.RWMutex{},
		taskConfig:  cfg,
		procState:   state,
		startedAt:   started,
		completedAt: time.Time{},
		exitResult:  nil,
		waitCh:      make(chan *drivers.ExitResult),
	}
	h.ctx, h.cancelFunc = context.WithCancel(ctx)
	go h.watchWaitCh()
	return h
}

func (h *taskHandle) watchWaitCh() {
	select {
	case r := <-h.waitCh:
		h.stateLock.Lock()
		h.exitResult = r
		h.cancelFunc()
		h.stateLock.Unlock()
	case <-h.ctx.Done():
	}
}

func (h *taskHandle) TaskStatus() (*drivers.TaskStatus, error) {
	h.stateLock.RLock()
	defer h.stateLock.RUnlock()

	m := map[string]string{}

	if h.runner != nil {
		stat, err := h.runner.Stats()
		if err != nil {
			return nil, errors.Wrap(err, "runner.Stats")
		}
		m["GtidSet"] = stat.CurrentCoordinates.GtidSet
	}

	// TODO Cannot get InspectTask -> TaskStatus called by any API.
	// See https://github.com/hashicorp/nomad/issues/4848
	return &drivers.TaskStatus{
		ID:               h.taskConfig.ID,
		Name:             h.taskConfig.Name,
		State:            h.procState,
		StartedAt:        h.startedAt,
		CompletedAt:      h.completedAt,
		ExitResult:       h.exitResult,
		DriverAttributes: m,
	}, nil
}

// used when h.runner has not been setup
func (h *taskHandle) onError(err error) {
	common.WriteWaitCh(h.waitCh, &drivers.ExitResult{
		ExitCode:  common.TaskStateDead,
		Signal:    0,
		OOMKilled: false,
		Err:       err,
	})
}

func (h *taskHandle) IsRunning() bool {
	h.stateLock.RLock()
	defer h.stateLock.RUnlock()
	return h.procState == drivers.TaskStateRunning
}

func (h *taskHandle) resumeTask(d *Driver) error {
	var err error
	h.runner, err = h.NewRunner(d)
	if err != nil {
		return err
	}
	go h.runner.Run()
	return nil
}

func (h *taskHandle) run(d *Driver) {
	h.stateLock.Lock()
	if h.exitResult == nil {
		h.exitResult = &drivers.ExitResult{}
	}
	h.procState = drivers.TaskStateRunning
	h.stateLock.Unlock()

	consulJobInfo, err := d.storeManager.GetJobInfo(h.taskConfig.JobName)
	if nil != err {
		h.onError(errors.Wrap(err, "get pause status from consul failed"))
		return
	}

	if consulJobInfo.JobStatus == common.DtleJobStatusPaused {
		h.logger.Info("job is paused. not starting the runner")
	} else {
		err = h.resumeTask(d)
		if err != nil {
			h.onError(err)
			return
		}
	}

	go func() {
		duration := time.Duration(d.config.StatsCollectionInterval) * time.Second
		t := time.NewTimer(0)
		for {
			select {
			case <-h.ctx.Done():
				if !t.Stop() {
					<-t.C
				}
				return
			case <-t.C:
				if h.runner != nil {
					s, err := h.runner.Stats()
					if err != nil {
						// ignore
					} else {
						h.stats = s
						if d.config.PublishMetrics {
							h.logger.Trace("emitStats")
							h.emitStats(s)
						}
					}
				}
				t.Reset(duration)
			}
		}
	}()
}

func (h *taskHandle) NewRunner(d *Driver) (runner DriverHandle, err error) {
	ctx := &common.ExecContext{
		Subject:  h.taskConfig.JobName,
		StateDir: d.config.DataDir,
	}

	logger := h.logger
	// create a new log file for debug job
	if d.config.DebugJob == h.taskConfig.JobName && hclog.LevelFromString(d.config.LogLevel) != hclog.Debug {
		logger = setupLogger(d.config.LogFile, "dtle_debug_job.log")
		logger.SetLevel(hclog.Debug)
	}

	switch common.TaskTypeFromString(h.taskConfig.Name) {
	case common.TaskTypeSrc:
		if h.driverConfig.SrcOracleConfig != nil {
			h.logger.Debug("found oracle src", "SrcOracleConfig", h.driverConfig.SrcOracleConfig)
			runner, err = extractor.NewExtractorOracle(ctx, h.driverConfig, logger, d.storeManager, h.waitCh, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewExtractor")
			}
		} else {
			e, err := mysql.NewExtractor(ctx, h.driverConfig, logger, d.storeManager, h.waitCh, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewOracleExtractor")
			}
			runner = e
			if h.driverConfig.TwoWaySync {
				ctx2 := &common.ExecContext{
					Subject:  ctx.Subject + "_dtrev",
					StateDir: d.config.DataDir,
				}
				cfg2 := &common.MySQLDriverConfig{
					DtleTaskConfig: common.DtleTaskConfig{
						DestType: "mysql",
					},
				}
				e.RevApplier, err = mysql.NewApplier(ctx2, cfg2, logger, d.storeManager, d.config.NatsAdvertise,
					h.waitCh, d.eventer, h.taskConfig, h.ctx)
			}

		}
	case common.TaskTypeDest:
		h.logger.Debug("found dest", "allConfig", h.driverConfig)
		switch strings.ToLower(h.driverConfig.DestType) {
		case "kafka":
			runner, err = kafka.NewKafkaRunner(ctx, logger,
				d.storeManager, d.config.NatsAdvertise, h.waitCh, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewKafkaRunner")
			}
		case "mysql":
			runner, err = mysql.NewApplier(ctx, h.driverConfig, logger, d.storeManager,
				d.config.NatsAdvertise, h.waitCh, d.eventer, h.taskConfig, h.ctx)
			if err != nil {
				return nil, errors.Wrap(err, "NewApplier")
			}
		case "":
			return nil, fmt.Errorf("DestType for dest task is empty")
		default:
			return nil, fmt.Errorf("unknown DestType for dest task")
		}
	case common.TaskTypeUnknown:
		return nil, fmt.Errorf("unknown processor type: %+v", h.taskConfig.Name)
	}
	return runner, err
}

func (h *taskHandle) emitStats(ru *common.TaskStatistics) {
	const srcFullFactor float32 = 4.5
	const dstFullFactor float32 = 5
	const srcIncrFactor float32 = 19
	const dstIncrFactor float32 = 9.5

	labels := []metrics.Label{{"task_name", fmt.Sprintf("%s_%s", h.taskConfig.JobName, h.taskConfig.Name)}}

	metrics.SetGaugeWithLabels([]string{"network", "in_msgs"}, float32(ru.MsgStat.InMsgs), labels)
	metrics.SetGaugeWithLabels([]string{"network", "out_msgs"}, float32(ru.MsgStat.OutMsgs), labels)
	metrics.SetGaugeWithLabels([]string{"network", "in_bytes"}, float32(ru.MsgStat.InBytes), labels)
	metrics.SetGaugeWithLabels([]string{"network", "out_bytes"}, float32(ru.MsgStat.OutBytes), labels)
	switch common.TaskTypeFromString(h.taskConfig.Name) {
	case common.TaskTypeSrc:
		metrics.SetGaugeWithLabels([]string{"buffer", "event_queue_size"}, float32(ru.BufferStat.BinlogEventQueueSize), labels)
		metrics.SetGaugeWithLabels([]string{"buffer", "src_queue_size"}, float32(ru.BufferStat.ExtractorTxQueueSize), labels)
		metrics.SetGaugeWithLabels([]string{"buffer", "send_by_timeout"}, float32(ru.BufferStat.SendByTimeout), labels)
		metrics.SetGaugeWithLabels([]string{"buffer", "send_by_size_full"}, float32(ru.BufferStat.SendBySizeFull), labels)

		metrics.SetGaugeWithLabels([]string{"memory.full_kb_est"}, float32(ru.MemoryStat.Full)*srcFullFactor/1024, labels)
		metrics.SetGaugeWithLabels([]string{"memory.incr_kb_est"}, float32(ru.MemoryStat.Incr)*srcIncrFactor/1024, labels)
	case common.TaskTypeDest:
		metrics.SetGaugeWithLabels([]string{"buffer", "dest_queue_size"}, float32(ru.BufferStat.ApplierMsgQueueSize), labels)
		metrics.SetGaugeWithLabels([]string{"buffer", "dest_queue2_size"}, float32(ru.BufferStat.ApplierTxQueueSize), labels)

		metrics.SetGaugeWithLabels([]string{"memory.full_kb_est"}, float32(ru.MemoryStat.Full)*dstFullFactor/1024, labels)
		metrics.SetGaugeWithLabels([]string{"memory.incr_kb_est"}, float32(ru.MemoryStat.Incr)*dstIncrFactor/1024, labels)
	case common.TaskTypeUnknown:
	}

	metrics.SetGaugeWithLabels([]string{"memory.full_kb_count"}, float32(ru.MemoryStat.Full)/1024, labels)
	metrics.SetGaugeWithLabels([]string{"memory.incr_kb_count"}, float32(ru.MemoryStat.Incr)/1024, labels)

	if ru.TableStats != nil {
		metrics.SetGaugeWithLabels([]string{"table", "insert"}, float32(ru.TableStats.InsertCount), labels)
		metrics.SetGaugeWithLabels([]string{"table", "update"}, float32(ru.TableStats.UpdateCount), labels)
		metrics.SetGaugeWithLabels([]string{"table", "delete"}, float32(ru.TableStats.DelCount), labels)
	}

	if ru.DelayCount != nil {
		// TODO
		//metrics.SetGaugeWithLabels([]string{"delay", "num"}, float32(ru.DelayCount.Num), labels)
		metrics.SetGaugeWithLabels([]string{"delay", "time"}, float32(ru.DelayCount.Time), labels)
	}

	if ru.ThroughputStat != nil {
		metrics.SetGaugeWithLabels([]string{"throughput", "num"}, float32(ru.ThroughputStat.Num), labels)
		metrics.SetGaugeWithLabels([]string{"throughput", "time"}, float32(ru.ThroughputStat.Time), labels)
	}

	if nil != ru.HandledTxCount.AppliedTxCount {
		metrics.SetGaugeWithLabels([]string{"dest_applied_incr_tx_count"}, float32(*ru.HandledTxCount.AppliedTxCount), labels)
	}
	if nil != ru.HandledTxCount.ExtractedTxCount {
		metrics.SetGaugeWithLabels([]string{"src_extracted_incr_tx_count"}, float32(*ru.HandledTxCount.ExtractedTxCount), labels)
	}

	if nil != ru.HandledQueryCount.AppliedQueryCount {
		metrics.SetGaugeWithLabels([]string{"dest_applied_incr_query_count"}, float32(*ru.HandledQueryCount.AppliedQueryCount), labels)
	}
	if nil != ru.HandledQueryCount.ExtractedQueryCount {
		metrics.SetGaugeWithLabels([]string{"src_extracted_incr_query_count"}, float32(*ru.HandledQueryCount.ExtractedQueryCount), labels)
	}
}

func (h *taskHandle) Destroy() {
	if h.shutdown {
		return
	}
	h.stateLock.Lock()
	h.shutdown = true
	h.stateLock.Unlock()

	common.WriteWaitCh(h.waitCh, &drivers.ExitResult{
		ExitCode:  0,
		Signal:    0,
		OOMKilled: false,
		Err:       nil,
	})

	if h.runner != nil {
		err := h.runner.Shutdown()
		if err != nil {
			h.logger.Error("error in h.runner.Shutdown", "err", err)
		}
	}
}

func (h *taskHandle) GetExitResult() *drivers.ExitResult {
	h.stateLock.Lock()
	defer h.stateLock.Unlock()
	if h.exitResult == nil {
		return &drivers.ExitResult{
			ExitCode:  0,
			Signal:    0,
			OOMKilled: false,
			Err:       nil,
		}
	} else {
		return h.exitResult.Copy()
	}
}

type DriverHandle interface {
	Run()

	// Shutdown is used to stop the task.
	// Do not send ExitResult in Shutdown().
	// pause API will call Shutdown and the task should not exit.
	Shutdown() error

	// Stats returns aggregated stats of the driver
	Stats() (*common.TaskStatistics, error)

	Finish1() error
}
