package dao

import (
	"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"
	"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
}

func New(batchSize int, flashTime time.Duration, dbAddress string, cacheFileDir string) *Dao {
	rtn := &Dao{
		batchSize:    batchSize,
		size:         0,
		channel:      make(chan *client.Point, ChannelSize),
		flashTime:    flashTime,
		dbAddress:    dbAddress,
		cacheFileDir: cacheFileDir,
	}

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

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

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

	return rtn
}

//平滑退出Dao，会flash缓存
//TODO 未实现
func (d *Dao) Close() {

}

func (d *Dao) flashFileCache() {
	for {
		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)
			}
		}
		time.Sleep(d.flashTime)
	}
}

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

	d.channel <- point
}

//list满或者超时，则数据入库
func (d *Dao) run() {
	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)
		}
	}
}

func (d *Dao) writeDb(pointList []*client.Point) error {
	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 {
	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) {
	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
	}
}
