package master

import (
	"context"
	"encoding/json"
	"os"
	"os/exec"
	"syscall"
	"time"

	"analysis/logo"
	"analysis/util"
)

const (
	opRemove = "remove"
	opAdd    = "add"
)

// Notice transit to slave
type Notice struct {
	Op      string   `json:"Op"`
	Content []string `json:"Content"`
}

type transit struct {
	chNotify chan<- []byte
	cancel   context.CancelFunc
}

// NamedProc 单个进程名字和服务通道
type NamedProc struct {
	// 进程名字
	Name string
	// 进程通道
	Channels []string
	// 进程runtime
	Env string
	// 进程config file path
	Config string
	// 进程param
	Param []string
}

// TypeProc 每个Type进程的参数
type TypeProc struct {
	// 进程类型FaceDetect/Yolo
	Typ string
	// 具名进程
	SNameProc []NamedProc
}

// Worker 单个进程服务
type Worker struct {
	pid   int
	cmd   *exec.Cmd
	info  *NamedProc
	trans *transit
}

// Daemon 监控的所有子进程
type Daemon struct {
	// 每个sdk类型启动的进程数量
	workers map[string][]*Worker
}

// NewDaemon 监控
func NewDaemon() *Daemon {
	return &Daemon{
		workers: make(map[string][]*Worker),
	}
}

//求交集
func intersect(slice1, slice2 []string) []string {
	m := make(map[string]int)
	nn := make([]string, 0)
	for _, v := range slice1 {
		m[v]++
	}

	for _, v := range slice2 {
		times, _ := m[v]
		if times == 1 {
			nn = append(nn, v)
		}
	}
	return nn
}

//求差集 slice1-并集
func difference(slice1, slice2 []string) []string {
	m := make(map[string]int)
	nn := make([]string, 0)
	inter := intersect(slice1, slice2)
	for _, v := range inter {
		m[v]++
	}

	for _, value := range slice1 {
		times, _ := m[value]
		if times == 0 {
			nn = append(nn, value)
		}
	}
	return nn
}

func removeWorker(w *Worker) {
	syscall.Kill(w.pid, syscall.SIGTERM)
	w.cmd.Wait()
	w.trans.cancel()
}

func (d *Daemon) rmWorkerWith(typ string) {
	if workers, ok := d.workers[typ]; ok {
		delete(d.workers, typ)

		for _, w := range workers {
			removeWorker(w)
		}
	}
}

func (d *Daemon) rmWorkerNoType(childs []TypeProc) {
	var newTypes []string
	for _, v := range childs {
		newTypes = append(newTypes, v.Typ)
	}

	var runTypes []string
	for k := range d.workers {
		runTypes = append(runTypes, k)
	}

	// 不存在于新信息中的type, remove
	rmTypes := difference(runTypes, newTypes)
	if len(rmTypes) > 0 {
		for _, v := range rmTypes {
			d.rmWorkerWith(v)
		}
	}
}

func (d *Daemon) rmWorkerNoNamed(workers []*Worker, procs []NamedProc) []*Worker {
	var newNames []string
	for _, v := range procs {
		newNames = append(newNames, v.Name)
	}

	var runNames []string
	for _, v := range workers {
		runNames = append(runNames, v.info.Name)
	}

	// 已经不需要存在进程,remove
	rmWorkers := difference(runNames, newNames)
	for _, v := range rmWorkers {
		for _, w := range workers {
			if v == w.info.Name {
				removeWorker(w)
			}
		}
	}

	// 保留已存在的进程
	stillWorks := intersect(runNames, newNames)
	var ret []*Worker
	for _, v := range stillWorks {
		for _, w := range workers {
			if v == w.info.Name {
				ret = append(ret, w)
			}
		}
	}

	return ret
}

//////////////////////////////////////////////////////////

func getNamedProc(typ string, childs []TypeProc) []NamedProc {
	for _, v := range childs {
		if v.Typ == typ {
			return v.SNameProc
		}
	}
	return nil
}

func getNamedProcInfo(name string, procs []NamedProc) *NamedProc {
	for _, v := range procs {
		if name == v.Name {
			return &v
		}
	}
	return nil
}

func (d *Daemon) channelChanged(ctx context.Context, typs []string, childs []TypeProc) {
	for _, s := range typs {
		// 存在这个类型的进程
		if workers, ok := d.workers[s]; ok {
			child := getNamedProc(s, childs)
			if child == nil {
				continue
			}
			var newNames []string
			for _, v := range child {
				newNames = append(newNames, v.Name)
			}

			var runNames []string
			for _, v := range workers {
				runNames = append(runNames, v.info.Name)
			}

			add := difference(newNames, runNames)
			for _, c := range child {
				for _, v := range add {
					if c.Name == v {
						d.startWorker(ctx, s, &c)
					}
				}
			}

			adjust := intersect(runNames, newNames)
			for _, v := range adjust {
				proc := getNamedProcInfo(v, child)
				if proc == nil {
					continue
				}

				for _, w := range workers {
					if v == w.info.Name {
						// 找到了对应名字的进程,首先求不需要再运行的通道
						removes := difference(w.info.Channels, proc.Channels)
						if len(removes) > 0 {
							// 通知子进程关闭通道
							n := Notice{
								Op:      opRemove,
								Content: removes,
							}
							if d, err := json.Marshal(n); err == nil {
								w.trans.chNotify <- d
							}
						}

						// 其次求出新增的通道
						adds := difference(proc.Channels, w.info.Channels)
						if len(adds) > 0 {
							// 通知子进程打开通道
							n := Notice{
								Op:      opAdd,
								Content: adds,
							}
							if d, err := json.Marshal(n); err == nil {
								w.trans.chNotify <- d
							}
						}

					}
				}
			}
		}
	}
}

func (d *Daemon) startNewWorker(ctx context.Context, child TypeProc) {
	for _, v := range child.SNameProc {
		d.startWorker(ctx, child.Typ, &v)
	}
}

func (d *Daemon) adjustWorker(ctx context.Context, childs []TypeProc) {
	var newTypes []string
	for _, v := range childs {
		newTypes = append(newTypes, v.Typ)
	}

	var runTypes []string
	for k := range d.workers {
		runTypes = append(runTypes, k)
	}

	// 新类型添加
	addWorkers := difference(newTypes, runTypes)
	for _, a := range addWorkers {
		for _, v := range childs {
			if a == v.Typ {
				// start new type proc
				d.startNewWorker(ctx, v)
			}
		}
	}

	stillWorkers := intersect(newTypes, runTypes)
	// 调整已存在的进程的通道
	d.channelChanged(ctx, stillWorkers, childs)
}

func (d *Daemon) updateWorker(ctx context.Context, childs []TypeProc) {
	// 新的进程信息,首先删除掉不再运行的类型
	d.rmWorkerNoType(childs)
	// 按名字删除特定类型中不再运行的进程
	for _, v := range childs {
		if workers, ok := d.workers[v.Typ]; ok {
			nWorkers := d.rmWorkerNoNamed(workers, v.SNameProc)
			d.workers[v.Typ] = nWorkers
		}
	}

	d.adjustWorker(ctx, childs)
}

// Watch watch
func (d *Daemon) Watch(ctx context.Context, ch <-chan []TypeProc) {
	chExit := make(chan ExitInfo, 32)
	go Reap(chExit)

	for {
		select {
		case <-ctx.Done():
			return
		case i := <-chExit:
			d.reWorker(ctx, &i)
		case childs := <-ch:
			d.updateWorker(ctx, childs)
		default:
			time.Sleep(time.Second)
		}
	}
}

func (d *Daemon) reWorker(ctx context.Context, info *ExitInfo) {
	// 有退出的进程,查看是否在运行进程中,拉起
	for i, workers := range d.workers {
		found := false
		for j, w := range workers {
			if w.pid == info.Pid {
				w = d.restartWorker(ctx, w)
				d.workers[i][j] = w
				found = true
				break
			}
		}
		if found {
			break
		}
	}
}

func (d *Daemon) restartWorker(ctx context.Context, w *Worker) *Worker {
	w.cmd.Wait()
	w.cmd = runProc(ctx, w.cmd.Path, w.info.Env, w.cmd.Args[1:])
	w.pid = w.cmd.Process.Pid
	return w
}

func (d *Daemon) startWorker(ctx context.Context, typ string, info *NamedProc) {

	ipcID := "analysis-" + typ + "-" + info.Name

	args := []string{
		`-role=slave`,
		"-sdk=" + typ,
		"-id=" + ipcID,
		"-" + util.ConfigPath + "=" + info.Config,
	}

	args = append(args, info.Param...)
	cmd := runProc(ctx, "./analysis", info.Env, args)

	if cmd == nil {
		logo.Errorf("ANALYSIS START SLAVE PROC %s IPC: %s Failed\n", typ, ipcID)
		return
	}
	logo.Infof("START SDK %s ID %s PID %d Env: %s\n", typ, ipcID, cmd.Process.Pid, info.Env)
	logo.Infoln(cmd.Args)

	ch := make(chan []byte, 3)
	cancel := fnNotify(ctx, ipcID, ch, logo.Infoln)

	w := &Worker{
		pid:  cmd.Process.Pid,
		cmd:  cmd,
		info: info,
		trans: &transit{
			chNotify: ch,
			cancel:   cancel,
		},
	}
	d.workers[typ] = append(d.workers[typ], w)
}

func runProc(ctxt context.Context, bin, env string, args []string) *exec.Cmd {
	cmd := exec.CommandContext(ctxt, bin, args...)
	rEnv := ""
	if len(env) != 0 {
		runtime := "LD_LIBRARY_PATH"
		rEnv = runtime + "=" + env
		logo.Infoln("Env String: ", rEnv)

		// remove os environ ld
		old := os.Getenv(runtime)
		os.Unsetenv(runtime)
		cmd.Env = os.Environ()
		cmd.Env = append(cmd.Env, rEnv)
		os.Setenv(runtime, old)
	}

	//debug
	cmd.Stdout = os.Stdout
	cmd.Stderr = os.Stderr
	cmd.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGTERM}

	if err := cmd.Start(); err == nil {
		return cmd
	}
	return nil
}