package app

import (
	"time"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap/tipb/go-tipb"
)

type SendBuffer struct {
	cpuTimeRecordChan chan *tipb.CPUTimeRecord
	sqlMetaChan       chan *tipb.SQLMeta
	planMetaChan      chan *tipb.PlanMeta
}

// SendBufferConfig specifies the max counts of data in memory, exceeding which the data will be dropped
type SendBufferConfig struct {
	// MaxCPUTimeRecordCount specifies the maximum number of CPUTimeRecord in buffered channel.
	// We should notice that each CPUTimeRecord could contain multiple <timestamp, cpuTime> tuples.
	MaxCPUTimeRecordCount uint32
	MaxSQLMetaCount       uint32
	MaxPlanMetaCount      uint32
}

func NewSendBuffer(config SendBufferConfig) *SendBuffer {
	return &SendBuffer{
		cpuTimeRecordChan: make(chan *tipb.CPUTimeRecord, config.MaxCPUTimeRecordCount),
		sqlMetaChan:       make(chan *tipb.SQLMeta, config.MaxSQLMetaCount),
		planMetaChan:      make(chan *tipb.PlanMeta, config.MaxPlanMetaCount),
	}
}

func (p *SendBuffer) WriteOneCPUTimeRecordOrDrop(record *tipb.CPUTimeRecord) {
	select {
	case p.cpuTimeRecordChan <- record:
	default:
	}
}
func (p *SendBuffer) WriteOneSQLMetaOrDrop(meta *tipb.SQLMeta) {
	select {
	case p.sqlMetaChan <- meta:
	default:
	}
}
func (p *SendBuffer) WriteOnePlanMetaOrDrop(meta *tipb.PlanMeta) {
	select {
	case p.planMetaChan <- meta:
	default:
	}
}

func (p *SendBuffer) readSomeCPUTimeRecords(maxCount int) []*tipb.CPUTimeRecord {
	ret := make([]*tipb.CPUTimeRecord, 0, maxCount)
	receivedCount := 0

	// wait for at least one record received
	record := <-p.cpuTimeRecordChan
	ret = append(ret, record)
	log.Debug("readCPUTimeRecords append", zap.Int("timestamp list size", len(record.RecordListTimestampSec)))
	receivedCount += len(record.RecordListTimestampSec)
	if receivedCount > maxCount {
		return ret
	}

	// read from channel, break if:
	// 1. read maxCount numbers of records (in timestamp count)
	// 2. timeout
	timeoutRead := time.After(100 * time.Millisecond)
	for {
		select {
		case record := <-p.cpuTimeRecordChan:
			ret = append(ret, record)
			log.Debug("readCPUTimeRecords append", zap.Int("timestamp list size", len(record.RecordListTimestampSec)))
			receivedCount += len(record.RecordListTimestampSec)
			if receivedCount > maxCount {
				return ret
			}
		case <-timeoutRead:
			log.Debug("readCPUTimeRecords timeout")
			return ret
		}
	}
}

func (p *SendBuffer) ReadOneCPUTimeRecord() *tipb.CPUTimeRecord {
	return <-p.cpuTimeRecordChan
}

// similar to readSomeCPUTimeRecords
func (p *SendBuffer) readSomeSQLMeta(maxCount int) []*tipb.SQLMeta {
	ret := make([]*tipb.SQLMeta, 0, maxCount)

	meta := <-p.sqlMetaChan
	ret = append(ret, meta)
	log.Debug("ReadSomeSQLMeta append")
	if len(ret) > maxCount {
		return ret
	}

	timeoutRead := time.After(100 * time.Millisecond)
	for {
		select {
		case meta := <-p.sqlMetaChan:
			ret = append(ret, meta)
			log.Debug("ReadSomeSQLMeta append")
			if len(ret) > maxCount {
				return ret
			}
		case <-timeoutRead:
			log.Debug("ReadSomeSQLMeta timeout")
			return ret
		}
	}
}

func (p *SendBuffer) ReadOneSQLMeta() *tipb.SQLMeta {
	return <-p.sqlMetaChan
}

// similar to readSomeCPUTimeRecords
func (p *SendBuffer) readSomePlanMeta(maxCount int) []*tipb.PlanMeta {
	ret := make([]*tipb.PlanMeta, 0, maxCount)

	meta := <-p.planMetaChan
	ret = append(ret, meta)
	log.Debug("readSomePlanMeta append")
	if len(ret) > maxCount {
		return ret
	}

	timeoutRead := time.After(100 * time.Millisecond)
	for {
		select {
		case meta := <-p.planMetaChan:
			ret = append(ret, meta)
			log.Debug("readSomePlanMeta append")
			if len(ret) > maxCount {
				return ret
			}
		case <-timeoutRead:
			log.Debug("readSomePlanMeta timeout")
			return ret
		}
	}
}

func (p *SendBuffer) ReadOnePlanMeta() *tipb.PlanMeta {
	return <-p.planMetaChan
}

type Sender struct {
	buf   *SendBuffer
	store Store
}

func NewSender(buf *SendBuffer, store Store) *Sender {
	return &Sender{
		buf:   buf,
		store: store,
	}
}

func (s *Sender) Start() {
	go func() {
		for {
			s.sendNextCPUTimeRecordBatch()
		}
	}()
	go func() {
		for {
			s.sendNextSQLMetaBatch()
		}
	}()
	go func() {
		for {
			s.sendNextPlanMetaBatch()
		}
	}()
}

func (s *Sender) sendNextCPUTimeRecordBatch() {
	batch := s.buf.readSomeCPUTimeRecords(1000)
	log.Debug("WriteCPUTimeRecordBatch", zap.Int("batch size", len(batch)))
	if err := s.store.WriteCPUTimeRecordBatch(batch); err != nil {
		log.Error("WriteCPUTimeRecordBatch failed", zap.Error(err))
	}
}

func (s *Sender) sendNextSQLMetaBatch() {
	batch := s.buf.readSomeSQLMeta(50)
	log.Debug("WriteSQLMetaBatch", zap.Int("batch size", len(batch)))
	if err := s.store.WriteSQLMetaBatch(batch); err != nil {
		log.Error("WriteSQLMetaBatch failed", zap.Error(err))
	}
}

func (s *Sender) sendNextPlanMetaBatch() {
	batch := s.buf.readSomePlanMeta(50)
	log.Debug("WritePlanMetaBatch", zap.Int("batch size", len(batch)))
	if err := s.store.WritePlanMetaBatch(batch); err != nil {
		log.Error("WritePlanMetaBatch failed", zap.Error(err))
	}
}