package producer

import (
	"strconv"
	"sync"
	"time"

	"github.com/coreos/etcd/clientv3"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/wormhole/pkg/etcdutil"
	"github.com/pingcap/wormhole/pkg/whpb"
	"golang.org/x/net/context"
)

const (
	defaultFlushSavepointInterval    = time.Hour * 1
	defaultFlushLatestOffsetInterval = time.Second * 5
	defaultRetention                 = 60 * 60 * 24 * 7 // 7 days
)

// PositionManager manages processed position from upstream data sources like MySQL/TiDB/MongoDB
type PositionManager interface {
	MarkAsDone(partition *whpb.Partition) error
	// Remember to call Close when PositionManger no longer used.
	Close()
}

// NewPositionManager returns a PositionManager.
// Currently only support etcd as storage.
func NewPositionManager(client *etcdutil.Client, opt ...PositionManagerOption) PositionManager {
	return newEtcdPositionManager(client, opt...)
}

type options struct {
	// retention period for savepoint in remote backend(unit in seconds).
	// If set zero, means we will not save offset to remote backend.
	retention int64

	flushSavepointInterval    time.Duration
	flushLatestOffsetInterval time.Duration

	// enableSavepoint is mainly for incremental syncing; fully dump doesn't need it.
	enableSavepoint bool
}

var defaultOptions = options{
	retention:                 defaultRetention,
	flushSavepointInterval:    defaultFlushSavepointInterval,
	flushLatestOffsetInterval: defaultFlushLatestOffsetInterval,
	enableSavepoint:           false,
}

// A PositionManagerOption sets options such as retention, flush interval, etc.
type PositionManagerOption func(*options)

// EnableSavepoint enables savepoint in PositionManager.
func EnableSavepoint() PositionManagerOption {
	return func(o *options) {
		o.enableSavepoint = true
	}
}

// WithRetention sets the retention time(in second) to saved offsets.
func WithRetention(retention int64) PositionManagerOption {
	return func(o *options) {
		o.retention = retention
	}
}

// WithFlushSavepointInterval sets interval of flushing savepoint in PositionManager.
func WithFlushSavepointInterval(interval time.Duration) PositionManagerOption {
	return func(o *options) {
		o.flushSavepointInterval = interval
	}
}

// WithFlushLatestOffsetInterval sets interval of flushing latest offset in PositionManager.
func WithFlushLatestOffsetInterval(interval time.Duration) PositionManagerOption {
	return func(o *options) {
		o.flushLatestOffsetInterval = interval
	}
}

type etcdPositionManager struct {
	opts options

	client *etcdutil.Client

	// partitions that has been saved to remote backend recently.
	queue          chan *whpb.Partition
	lastSavepoints struct {
		data map[string]map[int32]*whpb.Partition
		mu   sync.RWMutex
	}

	quit chan struct{}
	wg   sync.WaitGroup
}

func newEtcdPositionManager(client *etcdutil.Client, opt ...PositionManagerOption) PositionManager {
	opts := defaultOptions
	for _, o := range opt {
		o(&opts)
	}

	m := &etcdPositionManager{
		opts:   opts,
		client: client,
		queue:  make(chan *whpb.Partition, 1024),
		quit:   make(chan struct{}),
	}
	m.lastSavepoints.data = make(map[string]map[int32]*whpb.Partition)
	go m.mainLoop()
	return m
}

// mark partition as processed.
func (m *etcdPositionManager) MarkAsDone(partition *whpb.Partition) (err error) {
	m.queue <- partition
	return nil
}

func (m *etcdPositionManager) Close() {
	m.quit <- struct{}{}
	m.wg.Wait()
}

func (m *etcdPositionManager) mainLoop() {
	m.wg.Add(1)
	defer m.wg.Done()

	savepointTicker := time.NewTicker(m.opts.flushSavepointInterval)
	defer savepointTicker.Stop()

	latestOffsetTicker := time.NewTicker(m.opts.flushLatestOffsetInterval)
	defer latestOffsetTicker.Stop()

	for {
		select {
		case p := <-m.queue:
			m.save(p)

		case <-m.quit:
			log.Info("[producer] etcd position manager is going to exit")
			if err := m.flushSavepoints(); err != nil {
				log.Errorf("[producer] flush all partitions savepoints err %s", errors.ErrorStack(err))
			}
			if err := m.flushAllLatestOffsets(); err != nil {
				log.Errorf("[producer] flush all latest offsets err %s", errors.ErrorStack(err))
			}
			log.Info("[producer] etcd position manager exited")
			return

		case <-savepointTicker.C:
			if err := m.flushSavepoints(); err != nil {
				log.Errorf("[producer] flush all partitions savepoints err %s", errors.ErrorStack(err))
			}

		case <-latestOffsetTicker.C:
			if err := m.flushAllLatestOffsets(); err != nil {
				log.Errorf("[producer] flush all latest offsets err %s", errors.ErrorStack(err))
			}
		}
	}
}

// flush partition offset to remote backend.
func (m *etcdPositionManager) save(p *whpb.Partition) {
	m.lastSavepoints.mu.Lock()
	if _, ok := m.lastSavepoints.data[p.GetTopicName()]; !ok {
		m.lastSavepoints.data[p.GetTopicName()] = make(map[int32]*whpb.Partition)
	}
	m.lastSavepoints.data[p.GetTopicName()][p.GetPartitionId()] = p
	m.lastSavepoints.mu.Unlock()
}

// flushSavepoints flushes savepoints of all the topics.
func (m *etcdPositionManager) flushSavepoints() (err error) {
	if !m.opts.enableSavepoint {
		// disable saving.
		return nil
	}

	for _, partitions := range m.lastSavepoints.data {
		data := make([]*whpb.Partition, 0, len(partitions))
		for _, partition := range partitions {
			data = append(data, partition)
		}
		err = m.flushSavepoint(data)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// flushSavepoint flushes savepoint of single topic.
// We use unix timestamp(int64) as savepoint id.
func (m *etcdPositionManager) flushSavepoint(partitions []*whpb.Partition) (err error) {
	if len(partitions) == 0 {
		return nil
	}
	resp, err := m.client.Client.Grant(context.TODO(), m.opts.retention)
	if err != nil {
		return errors.Trace(err)
	}
	leaseOpt := clientv3.WithLease(resp.ID)

	data := &whpb.Partitions{
		Partitions: partitions,
	}
	topic := partitions[0].GetTopicName()
	savepointID := strconv.FormatInt(time.Now().Unix(), 10)
	err = m.client.SaveProducerPartitionSavepoint(topic, savepointID, data, leaseOpt)
	return errors.Trace(err)
}

// flushAllLatestOffsets flushes the latest offsets of all the topics.
func (m *etcdPositionManager) flushAllLatestOffsets() (err error) {
	for _, partitions := range m.lastSavepoints.data {
		data := make([]*whpb.Partition, 0, len(partitions))
		for _, partition := range partitions {
			data = append(data, partition)
		}
		err = m.flushLatestOffsets(data)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// flushLatestOffsets flushes the latest offsets of single single topic.
func (m *etcdPositionManager) flushLatestOffsets(partitions []*whpb.Partition) (err error) {
	if len(partitions) == 0 {
		return nil
	}
	data := &whpb.Partitions{
		Partitions: partitions,
	}
	topic := partitions[0].GetTopicName()
	err = m.client.SaveProducerPartition(topic, data)
	return errors.Trace(err)
}