package dao

import (
	"context"
	"encoding/json"
	"git.quantgroup.cn/DevOps/enoch/pkg/global"
	"git.quantgroup.cn/DevOps/enoch/pkg/glog"
	"github.com/influxdata/influxdb/client/v2"
	"io/ioutil"
	"os"
	"path"
	"strings"
	"sync"
	"time"
)

const (
	ChannelSize       = 1024
	filePrefixWriting = "writing"
	filePrefixCache   = "cache"
)

type Dao struct {
	batchSize    int
	size         int
	channel      chan *client.Point
	flashTime    time.Duration
	dbAddress    string
	cacheFileDir string
	isClose      bool //平滑退出标记
	ctx          context.Context
	ctxCancel    context.CancelFunc
	wg           *sync.WaitGroup
}

func New(batchSize int, flashTime time.Duration, dbAddress string, cacheFileDir string) *Dao {
	ctx, cancel := context.WithCancel(context.Background())
	rtn := &Dao{
		batchSize:    batchSize,
		size:         0,
		channel:      make(chan *client.Point, ChannelSize),
		flashTime:    flashTime,
		dbAddress:    dbAddress,
		cacheFileDir: cacheFileDir,
		isClose:      false,
		ctx:          ctx,
		ctxCancel:    cancel,
		wg:           new(sync.WaitGroup),
	}

	if stat, err := os.Stat(cacheFileDir); err != nil || !stat.IsDir() {
		glog.Error("cacheFileDir:", err)
	}

	//数据入库
	go rtn.run()

	//清空文件缓存
	go rtn.flashFileCache()

	return rtn
}

//平滑退出Dao，会flash缓存
func (d *Dao) Close() {
	d.isClose = true
	d.ctxCancel()
	d.wg.Wait()
}

func (d *Dao) flashFileCache() {
	d.wg.Add(1)
	timer := time.NewTimer(0)
	for {
		select {
		case <-timer.C:
			fileList, err := ioutil.ReadDir(d.cacheFileDir)
			if err != nil || len(fileList) == 0 {
				continue
			}
			for _, file := range fileList {
				sl := strings.Split(file.Name(), ":")
				if len(sl) == 0 || sl[0] != filePrefixCache {
					continue
				}

				//读取文件
				filePath := path.Join(d.cacheFileDir, file.Name())
				data, err := ioutil.ReadFile(filePath)
				if err != nil {
					glog.Error("can not read file:", filePath, err)
					continue
				}

				cachePointList := make([]CachePoint, 0)
				if err := json.Unmarshal(data, &cachePointList); err != nil {
					glog.Error("can not unmarshal file:", filePath)
					continue
				}

				pointList := make([]*client.Point, 0)
				for _, cachePoint := range cachePointList {
					point, err := client.NewPoint(cachePoint.Name, cachePoint.Tags, cachePoint.Fields, cachePoint.Time)
					if err != nil {
						continue
					}
					pointList = append(pointList, point)
				}

				if err := d.writeDb(pointList); err != nil {
					glog.Warn("flash file cache: can not write db")
					continue
				} else {
					glog.Info("文件缓存写入db成功：", filePath)
				}

				if err := os.Remove(filePath); err != nil {
					glog.Error("删除文件失败：", filePath)
				}
			}
			timer.Reset(d.flashTime)
		case <-d.ctx.Done():
			glog.Info("flash file cache 平滑退出")
			timer.Stop()
			d.wg.Done()
			return
		}
	}
}

func (d *Dao) MsgProcess(point *client.Point) {
	if point == nil {
		return
	}

	if d.isClose {
		return
	}

	d.channel <- point
}

//list满或者超时，则数据入库
func (d *Dao) run() {
	d.wg.Add(1)
	defer func() {
		if err := recover(); err != nil {
			glog.Error(err)
		}
	}()

	pointList := make([]*client.Point, 0, d.batchSize)
	timer := time.NewTimer(d.flashTime)
	for {
		select {
		case point := <-d.channel:
			if len(pointList) == d.batchSize {
				go d.batchWrite(pointList)
				pointList = make([]*client.Point, 0, d.batchSize)
				timer.Reset(d.flashTime)
			}
			pointList = append(pointList, point)
		case <-timer.C:
			//保存文件
			go d.batchWrite(pointList)
			pointList = make([]*client.Point, 0, d.batchSize)
			timer.Reset(d.flashTime)
		case <-d.ctx.Done():
			go d.batchWrite(pointList)
			glog.Info("存入influx平滑退出")
			timer.Stop()
			d.wg.Done()
			return
		}
	}
}

func (d *Dao) writeDb(pointList []*client.Point) error {
	defer func() {
		if err := recover(); err != nil {
			glog.Error("pointList panic!", err)
			return
		}
	}()

	config := client.HTTPConfig{
		Addr:    d.dbAddress,
		Timeout: time.Second * 10,
	}

	//建立连接
	connect, err := client.NewHTTPClient(config)
	defer func() { _ = connect.Close() }()
	if err != nil {
		return err
	}

	points, err := client.NewBatchPoints(client.BatchPointsConfig{
		Database: global.InfluxDbName,
	})
	if err != nil {
		return err
	}

	points.AddPoints(pointList)
	err = connect.Write(points)
	if err != nil {
		return err
	}
	return nil
}

func (d *Dao) writeFile(pointList []*client.Point) error {
	defer func() {
		if err := recover(); err != nil {
			glog.Error("writeFile panic!", err)
			return
		}
	}()

	cachePointList := make([]CachePoint, 0)
	for _, point := range pointList {
		cachePointList = append(cachePointList, NewPoint(point))
	}

	data, err := json.Marshal(cachePointList)
	if err != nil {
		return err
	}

	fileName := time.Now().Format("2006-01-02T15:04:05.999999999")
	writingFileName := path.Join(d.cacheFileDir, filePrefixWriting+":"+fileName)
	if err := ioutil.WriteFile(writingFileName, data, 0644); err != nil {
		return err
	}

	cacheFileName := path.Join(d.cacheFileDir, filePrefixCache+":"+fileName)
	if err := os.Rename(writingFileName, cacheFileName); err != nil {
		return err
	}

	return nil
}

func (d *Dao) batchWrite(pointList []*client.Point) {
	defer func() {
		if err := recover(); err != nil {
			glog.Error("batch write panic!", err)
			return
		}
	}()

	d.wg.Add(1)
	defer d.wg.Done()

	if len(pointList) == 0 {
		return
	}

	if err := d.writeDb(pointList); err != nil {
		glog.Warn("写入influxDB失败:", err)
	} else {
		glog.Infof("成功写入influxDB%d条", len(pointList))
		return
	}

	if err := d.writeFile(pointList); err != nil {
		glog.Error("写入文件缓存失败，数据丢失:", err)
	} else {
		glog.Infof("成功写入文件缓存%d条", len(pointList))
		return
	}
}
