package task

import (
	"fmt"
	"log"
	"strconv"
	"sync"
	"time"

	"github.com/tidbcloud/observability/data-collector/pkg"
	"github.com/tidbcloud/observability/data-collector/pkg/collector/sinks"
	"github.com/tidbcloud/observability/data-collector/pkg/collector/sources"
)

type Task struct {
	Source sources.Source
	Sink   sinks.Sink
}

func NewTask(source sources.Source, sink sinks.Sink) Task {
	return Task{
		Source: source,
		Sink:   sink,
	}
}

type Excutor struct {
	tasks []Task
}

func NewExcutor(tasks []Task) Excutor {
	return Excutor{
		tasks: tasks,
	}
}
func (e *Excutor) Excute(delay int) error {
	log.Println("task starts")
	var wg sync.WaitGroup
	now := time.Now().Unix() - (int64(delay) * 60)
	for _, task := range e.tasks {
		wg.Add(1)
		go func(t Task) {
			if err := e.excute(&wg, t, now); err != nil {
				log.Println(fmt.Errorf("excute task failed: %w", err))
			}

		}(task)
	}
	wg.Wait()
	log.Println("all task finished")
	return nil
}

func (e *Excutor) excute(wg *sync.WaitGroup, task Task, timestemp int64) error {
	defer wg.Done()
	//get checkpoint
	dbconf := task.Source.GetConfig()
	checkpoint, exits, err := task.Sink.DownloadFile(fmt.Sprintf("/%s/%s/checkpoint", dbconf.GetName(), dbconf.GetVersion()))
	if err != nil {
		log.Printf("download checkpoint failed %v", err)
		failedToScrape(dbconf.GetName(), dbconf.GetVersion())
		return err
	}
	var checkpointTime int64
	if !exits {
		log.Println("not found checkpoint, get db data from the beginning")
		checkpointTime = 0
	} else {
		log.Printf("found checkpoint %s\n", string(checkpoint))
		checkpointTime, _ = strconv.ParseInt(string(checkpoint), 10, 64)
	}
	//run source
	results, err := task.Source.Run(timestemp, checkpointTime)
	if err != nil {
		log.Printf("run source failed %v", err)
		failedToScrape(dbconf.GetName(), dbconf.GetVersion())
		return err
	}
	//upload files
	for _, result := range results {
		err = task.Sink.WriteToFile(result.Path, result.Body)
		if err != nil {
			log.Printf("upload file failed %v", err)
			failedToUpload(dbconf.GetName(), dbconf.GetVersion())
			return err
		}
	}
	//upload checkpoint
	err = task.Sink.WriteToFile(fmt.Sprintf("/%s/%s/checkpoint", dbconf.GetName(), dbconf.GetVersion()), []byte(strconv.FormatInt(timestemp, 10)))
	if err != nil {
		log.Printf("upload checkpoint failed %v", err)
		failedToUpload(dbconf.GetName(), dbconf.GetVersion())
		return err
	}
	runsucc(dbconf.GetName(), dbconf.GetVersion())
	return nil
}

func failedToUpload(name, version string) {
	pkg.AgentFailedToUploadDataMetric.WithLabelValues(name, version).Inc()
}

func failedToScrape(name, version string) {
	pkg.AgentFailedToScrapeDataMetric.WithLabelValues(name, version).Inc()
}

func runsucc(name, version string) {
	pkg.AgentTaskSuccess.WithLabelValues(name, version).Inc()
}