package schedule

import (
	"context"
	"strconv"
	"time"

	"github.com/tidbcloud/metering/pkg/common/constants"
	"github.com/tidbcloud/metering/pkg/common/metrics"
	"github.com/tidbcloud/metering/pkg/errors"

	log "github.com/sirupsen/logrus"
	"github.com/tidbcloud/metering/pkg/facade"
)

type Task struct {
	Name          string
	start         int64
	interval      time.Duration
	immediately   bool
	nextTimestamp func(ctx context.Context, checkpoint int64, start int64) int64
	ready         func(ctx context.Context, timestamp int64) bool
	execute       func(ctx context.Context, timestamp int64) error
	ticker        *time.Ticker
}

func (t Task) AsyncTicker() {
	t.ticker = time.NewTicker(t.interval)

	go func() {
		if t.immediately {
			err := t.Execute()
			if err != nil {
				t.handleError(err)
			}
		}
		for range t.ticker.C {
			err := t.Execute()
			if err != nil {
				t.handleError(err)
			}
		}
	}()
}

func (t Task) handleError(err error) {
	if err != nil {
		log.Errorf("task %s failed, %s", t.Name, err.Error())
		failedType := errors.UnrecognizedError
		if a, ok := err.(errors.Error); ok {
			failedType = a.Code()
		}
		labels := map[string]string{
			constants.MetricsTaskName:   t.Name,
			constants.MetricsFailedType: strconv.FormatInt(int64(failedType), 10),
		}
		metrics.MetricsInstanceFailedCount.With(labels).Inc()
	}
}

func (t Task) Execute() error {
	ctx := context.Background()
	checkpoint, err := facade.GetCheckpointReaderWriter(ctx).GetCheckpoint(ctx, t.Name)
	if err != nil {
		return err
	}

	err = t.loop(ctx, checkpoint)
	if err != nil {
		return err
	}
	return nil
}

func (t Task) loop(ctx context.Context, checkpoint int64) error {
	savedCheckpoint := checkpoint
	for nextTS := t.nextTimestamp(ctx, savedCheckpoint, t.start); t.ready(ctx, nextTS); nextTS = t.nextTimestamp(ctx, savedCheckpoint, t.start) {
		err := t.run(ctx, nextTS)
		if err != nil {
			log.Errorf("loop execute task failed, %s", err.Error())
			return err
		}
		savedCheckpoint = nextTS
	}

	log.Infof("loop execute task %s from %d to %d", t.Name, checkpoint, savedCheckpoint)

	return nil
}

func (t Task) run(ctx context.Context, next int64) error {
	log.Infof("start to execute next task, name = %s, timestamp = %d", t.Name, next)
	err := t.execute(ctx, next)
	if err != nil {
		log.Errorf("execute task failed, %s", err.Error())
		return err
	}

	log.Infof("execute task succeed, start to upload data, name = %s, timestamp = %d,", t.Name, next)
	err = facade.GetScheduleTaskDataSync(ctx).Upload(ctx, t.Name, next)
	if err != nil {
		log.Errorf("Upload data failed, %s", err.Error())
		return err
	}

	log.Infof("upload data succeed, start to save checkpoint, name = %s, timestamp = %d,", t.Name, next)
	err = facade.GetCheckpointReaderWriter(ctx).SaveCheckpoint(ctx, t.Name, next)
	if err != nil {
		log.Errorf("SaveCheckpoint failed, %s", err.Error())
		return err
	}

	lag := time.Now().Unix() - next
	metricslabels := map[string]string{
		constants.MetricsTaskName: t.Name,
	}
	metrics.MetricsReportLag.With(metricslabels).Set(float64(lag))

	return nil
}