package executor

import (
	"fmt"
	"io"
	"sync"
	"time"

	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/base"
)

const (
	EventDataUnknown        EventDataTypeHint = ""
	EventDataInt            EventDataTypeHint = "int"
	EventDataBool           EventDataTypeHint = "bool"
	EventDataStr            EventDataTypeHint = "string"
	EventDataStrList        EventDataTypeHint = "[]string"
	EventDataError          EventDataTypeHint = "error"
	EventDataExecTask       EventDataTypeHint = "*executor.ExecutorTask"
	EventDataActionResult   EventDataTypeHint = "*executor.ActionTaskStatus"
	EventDataActionTask     EventDataTypeHint = "*executor.ActionTask"
	EventDataActionTaskList EventDataTypeHint = "[]*executor.ActionTask"
	EventDataDag            EventDataTypeHint = "dag.Dag"
	EventDataDagAction      EventDataTypeHint = "*dag.DagAction"
	EventDataDagActionList  EventDataTypeHint = "[]*dag.DagAction"
	EventDataExecTaskList   EventDataTypeHint = "[]*executor.ExecutorTask"

	EventAny   EventName = "any"
	EventDebug EventName = "debug"

	EventExecutorStart                     EventName = "executor.start"
	EventExecutorDone                      EventName = "executor.done"
	EventExecutorRunLoopStart              EventName = "executor.run-loop.start"
	EventExecutorRunLoopOnStopPolicyNotify EventName = "executor.run-loop.on-case.stop-policy-notify"
	EventExecutorRunLoopOnExecTaskSubmit   EventName = "executor.run-loop.on-case.exec-task.submit"
	EventExecutorRunLoopOnExecTaskAbort    EventName = "executor.run-loop.on-case.exec-task.abort"
	EventExecutorRunLoopOnActionResult     EventName = "executor.run-loop.on-case.action-result"

	EventExecTaskLoadUnfinishedStart EventName = "exec-task.load-unfinished.start"
	EventExecTaskLoadUnfinishedDone  EventName = "exec-task.load-unfinished.done"
	EventExecTaskListCount           EventName = "exec-task.list-count"
	EventExecTaskListEmpty           EventName = "exec-task.list-empty"
	EventExecTaskStall               EventName = "exec-task.stall"
	EventExecTaskStucked             EventName = "exec-task.stucked"

	EventExecTaskSubmitPresend EventName = "exec-task.submit.pre-send"
	EventExecTaskSubmitSent    EventName = "exec-task.submit.sent"

	EventExecTaskAbortPresend EventName = "exec-task.abort.pre-send"
	EventExecTaskAbortSent    EventName = "exec-task.abort.sent"

	// the `start` event didn't use yet
	EventExecTaskStart EventName = "exec-task.start"
	EventExecTaskDone  EventName = "exec-task.done"
	EventExecTaskError EventName = "exec-task.error"

	EventExecTaskAddToRunning         EventName = "exec-task.add-to-running"
	EventExecTaskQueueActionsStart    EventName = "exec-task.queue-actions.start"
	EventExecTaskDuplicated           EventName = "exec-task.duplicated"
	EventExecTaskDuplicatedRunning    EventName = "exec-task.duplicated.running"
	EventExecTaskDuplicatedFinished   EventName = "exec-task.duplicated.finished"
	EventExecTaskDuplicatedSubmitting EventName = "exec-task.duplicated.submitting"

	EventGetReadyActions EventName = "dag-action.get-ready-list"
	EventNewReadyActions EventName = "dag-action.new-ready-list"

	EventDagActionStgSetDone     EventName = "dag-action.stg.set-done"
	EventDagActionStgQueryIsDone EventName = "dag-action.stg.query-is-done"

	EventActionTasksToBeRun EventName = "action-task.to-be-run-list"
	EventActionTaskStart    EventName = "action-task.start"

	EventSchedulerAdd                   EventName = "scheduler.add"
	EventSchedulerPop                   EventName = "scheduler.pop"
	EventSchedulerIdlePop               EventName = "scheduler.pop-when-idle"
	EventSchedulerTickPop               EventName = "scheduler.pop-when-tick"
	EventSchedulerReschedule            EventName = "scheduler.reschedule"
	EventSchedulerSkipRescheduleOnAbort EventName = "scheduler.reschedule.skip-aborting-task"
	EventExecTaskTerminalError          EventName = "exec-task.terminal-error"
	EventSchedulerSkip                  EventName = "scheduler.skip"

	EventSchedulerSendAbortSignalToAction EventName = "action.send-abort-signal"

	EventExecStgError      EventName = "exec-stg.error"
	EventDagActionStgError EventName = "dag-action-stg.error"
)

type (
	EventName         string
	EventDataTypeHint string
)

type EventData struct {
	DataName string
	TypeHint EventDataTypeHint
	Object   base.Cloneable
}

func (ed EventData) Clone() EventData {
	return EventData{
		DataName: ed.DataName,
		TypeHint: ed.TypeHint,
		Object:   ed.Object.CloneableClone(),
	}
}

func EventDataWithType(typeHint EventDataTypeHint, object base.Cloneable) EventData {
	return EventData{
		TypeHint: typeHint,
		Object:   object,
	}
}

func EventDataWithName(dataName string, object base.Cloneable) EventData {
	return EventData{
		DataName: dataName,
		Object:   object,
	}
}

func EventDataWithNameType(dataName string, typeHint EventDataTypeHint, object base.Cloneable) EventData {
	return EventData{
		DataName: dataName,
		TypeHint: typeHint,
		Object:   object,
	}
}

// (most of) events will be notified from main thread (aka, run thread), don't do heavy things in receiver
type EventReceiver interface {
	OnEvent(name EventName, data ...EventData)
}

type EventHub interface {
	RegisterReceiver(receiver EventReceiver)
}

type SyncEventHub struct {
	receivers     []EventReceiver
	asyncReceiver *AsyncEventHub
	locker        sync.Mutex
}

func NewSyncEventHub() *SyncEventHub {
	return &SyncEventHub{}
}

func (hub *SyncEventHub) GetAsyncHub() EventHub {
	hub.locker.Lock()
	defer hub.locker.Unlock()
	if hub.asyncReceiver == nil {
		hub.asyncReceiver = NewAsyncEventHub()
		// put async-receiver at the first place
		hub.receivers = append([]EventReceiver{hub.asyncReceiver}, hub.receivers...)
	}
	return hub.asyncReceiver
}

func (hub *SyncEventHub) RegisterReceiver(receiver EventReceiver) {
	hub.locker.Lock()
	defer hub.locker.Unlock()
	hub.receivers = append(hub.receivers, receiver)
}

func (hub *SyncEventHub) OnEvent(name EventName, data ...EventData) {
	hub.locker.Lock()
	defer hub.locker.Unlock()
	for _, receiver := range hub.receivers {
		receiver.OnEvent(name, data...)
	}
}

type AsyncEventHub struct {
	receivers []EventReceiver
	locker    sync.Mutex
}

func NewAsyncEventHub() *AsyncEventHub {
	return &AsyncEventHub{}
}

func (hub *AsyncEventHub) RegisterReceiver(receiver EventReceiver) {
	hub.locker.Lock()
	defer hub.locker.Unlock()
	hub.receivers = append(hub.receivers, receiver)
}

func (hub *AsyncEventHub) OnEvent(name EventName, data ...EventData) {
	hub.locker.Lock()
	defer hub.locker.Unlock()
	if len(hub.receivers) == 0 {
		return
	}
	// TODO: clone data to avoid race
	var dataClone []EventData
	for _, it := range data {
		dataClone = append(dataClone, it.Clone())
	}
	for _, receiver := range hub.receivers {
		receiver.OnEvent(name, dataClone...)
	}
}

type GeneralLogger struct {
	prefix     string
	w          io.Writer
	timeFormat string
}

func NewGeneralLogger(loggerName string, w io.Writer) *GeneralLogger {
	prefix := loggerName
	if prefix != "" {
		prefix = "[" + loggerName + "] "
	}
	return &GeneralLogger{
		prefix:     prefix,
		w:          w,
		timeFormat: time.DateTime,
	}
}

func (logger *GeneralLogger) OnEvent(name EventName, data ...EventData) {
	timeStr := time.Now().Format(logger.timeFormat)
	str := fmt.Sprintf("[%v] %s%v", timeStr, logger.prefix, name)
	if len(data) != 0 {
		str += ":"
	}
	for _, it := range data {
		str += " "
		if it.DataName != "" {
			str += fmt.Sprintf("'%v' ", it.DataName)
		}
		if it.TypeHint != EventDataUnknown && it.TypeHint != EventDataInt && it.TypeHint != EventDataStr && it.TypeHint != EventDataStrList && it.TypeHint != EventDataBool {
			str += fmt.Sprintf("(type %v) ", it.TypeHint)
		}
		str += fmt.Sprintf("= %v;", it.Object)
	}
	_, _ = fmt.Fprintf(logger.w, "%s\n", str)
}

type EventAllowFilter struct {
	origin    EventReceiver
	allowList map[EventName]bool
}

func NewEventAllowFilter(origin EventReceiver, names ...EventName) *EventAllowFilter {
	eaf := &EventAllowFilter{
		origin:    origin,
		allowList: make(map[EventName]bool),
	}
	for _, name := range names {
		eaf.allowList[name] = true
	}
	return eaf
}

func (eaf *EventAllowFilter) OnEvent(name EventName, data ...EventData) {
	if eaf.allowList[name] {
		eaf.origin.OnEvent(name, data...)
	}
}

type EventDenyFilter struct {
	origin   EventReceiver
	denyList map[EventName]bool
}

func NewEventDenyFilter(origin EventReceiver, names ...EventName) *EventDenyFilter {
	edf := &EventDenyFilter{
		origin:   origin,
		denyList: make(map[EventName]bool),
	}
	for _, name := range names {
		edf.denyList[name] = true
	}
	return edf
}

func (eaf *EventDenyFilter) OnEvent(name EventName, data ...EventData) {
	if eaf.denyList[name] {
		return
	}
	eaf.origin.OnEvent(name, data...)
}

type DragSlow struct {
	dur time.Duration
}

func NewDragSlow(dur time.Duration) *DragSlow {
	return &DragSlow{
		dur: dur,
	}
}

func (drag *DragSlow) OnEvent(name EventName, data ...EventData) {
	time.Sleep(drag.dur)
}

type GlobalTaskProgressReceiver func(actionTaskStatus *ActionTaskStatus)

type GlobalTaskProgressNotifier struct {
	receivers []GlobalTaskProgressReceiver
}

func NewGlobalTaskProgressNotifier() *GlobalTaskProgressNotifier {
	return &GlobalTaskProgressNotifier{}
}

func (pn *GlobalTaskProgressNotifier) RegisterReceiver(receiver GlobalTaskProgressReceiver) {
	pn.receivers = append(pn.receivers, receiver)
}

func (pn *GlobalTaskProgressNotifier) OnEvent(name EventName, data ...EventData) {
	if name != EventExecutorRunLoopOnActionResult {
		return
	}
	result := data[0].Object.(any).(*ActionTaskStatus)
	for _, it := range pn.receivers {
		it(result)
	}
}

type EventsCounter struct {
	counts map[EventName]int
	locker sync.Mutex
}

func NewEventsCounter(mustHaveNames ...EventName) *EventsCounter {
	// for better dump
	counts := make(map[EventName]int)
	for _, name := range mustHaveNames {
		counts[name] = 0
	}
	return &EventsCounter{
		counts: counts,
	}
}

func (counter *EventsCounter) OnEvent(name EventName, data ...EventData) {
	counter.locker.Lock()
	defer counter.locker.Unlock()
	val := counter.counts[name]
	counter.counts[name] = val + 1
}

func (counter *EventsCounter) Dump(w io.Writer) {
	for name, cnt := range counter.counts {
		_, _ = fmt.Fprintf(w, "%v: %v\n", name, cnt)
	}
}

func (counter *EventsCounter) GetCount(name EventName) int {
	return counter.counts[name]
}