package taskqueue

import (
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus"

	"go.uber.org/zap"
)

var (
	DownstreamTimestampMetrics = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Namespace: "metrics_integration_pipeline",
			Name:      "downstream_timestamp",
		}, []string{},
	)
)

func init() {
	prometheus.MustRegister(DownstreamTimestampMetrics)
}

func (tqe *taskQueueExtension) GetAckTimestamp() time.Time {
	tqe.timerMu.Lock()
	defer tqe.timerMu.Unlock()

	return tqe.ackedTs
}

func (tqe *taskQueueExtension) GetNewestTimestamp() time.Time {
	return time.Now().Truncate(tqe.interval)
}

func (tqe *taskQueueExtension) GetCurrentTimestampAndProgress() time.Time {
	tqe.timerMu.Lock()
	defer tqe.timerMu.Unlock()

	currentTs := tqe.ackedTs
	//tqe.currentTs = tqe.currentTs.Add(tqe.interval)

	DownstreamTimestampMetrics.WithLabelValues().Set(float64(tqe.ackedTs.Unix()))

	err := tqe.persist(tqe.ackedTs)
	if err != nil {
		tqe.logger.Warn("persist timestamp failed", zap.Time("current_timestamp", tqe.ackedTs), zap.Error(err))
	}

	return currentTs
}

func (tqe *taskQueueExtension) AckTimestamp(ts time.Time) bool {
	tqe.timerMu.Lock()
	defer tqe.timerMu.Unlock()

	if ts.Truncate(tqe.interval) != ts {
		return false
	}

	if tqe.ackedTs.After(ts) {
		tqe.logger.Info("ack a staled ts", zap.Time("ts_to_ack", ts), zap.Time("acked_ts", tqe.ackedTs))
	}

	// cumulative ack
	if tqe.lastTs.Before(ts) || tqe.lastTs.Equal(ts) {
		tqe.logger.Info("timestamp acked", zap.Time("acked_ts", ts))
		//tqe.currentTs = ts.Add(tqe.interval)
		tqe.ackedTs = ts

		DownstreamTimestampMetrics.WithLabelValues().Set(float64(tqe.ackedTs.Unix()))

		err := tqe.persist(tqe.ackedTs)
		if err != nil {
			tqe.logger.Warn("persist timestamp failed", zap.Time("ack_timestamp", tqe.ackedTs), zap.Error(err))
		}

		return true
	} else {
		return false
	}
}

func (tqe *taskQueueExtension) SetLastTimestamp(ts time.Time) {
	tqe.timerMu.Lock()
	defer tqe.timerMu.Unlock()

	tqe.lastTs = ts
}

func (tqe *taskQueueExtension) SetAckTimestamp(ts time.Time) {
	tqe.timerMu.Lock()
	defer tqe.timerMu.Unlock()

	tqe.lastTs = ts

	DownstreamTimestampMetrics.WithLabelValues().Set(float64(tqe.ackedTs.Unix()))

	err := tqe.persist(tqe.ackedTs)
	if err != nil {
		tqe.logger.Warn("persist timestamp failed", zap.Time("timestamp_to_ack", tqe.ackedTs), zap.Error(err))
	}
}

func (tqe *taskQueueExtension) persist(ts time.Time) error {
	unixTs := ts.Unix()
	return tqe.persistCli.Set(DOWNSTREAM_TIMESTAMP_KEY, strconv.FormatInt(unixTs, 10))
}