package main

import (
	"flag"
	"fmt"
	"log"
	"net/http"
	"time"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/tidbcloud/observability/data-collector/pkg"
	"github.com/tidbcloud/observability/data-collector/pkg/collector/config"
	"github.com/tidbcloud/observability/data-collector/pkg/collector/sinks"
	"github.com/tidbcloud/observability/data-collector/pkg/collector/sources"
	"github.com/tidbcloud/observability/data-collector/pkg/collector/task"
)

func main() {
	var configpath string
	flag.StringVar(&configpath, "config", "./config.yaml", "path to config files")
	flag.Parse()
	//init config
	conf := config.NewConfig(configpath)
	//init sink we only use one sink now
	sink := sinks.NewS3Sink(config.NewS3Sink(conf.Sinks[0]))
	//init task
	var tasks []task.Task
	for _, source := range conf.Sources {
		switch source["type"].(string) {
		case "DB":
			source := sources.NewDBSource(config.NewDBSource(source))
			tasks = append(tasks, task.NewTask(&source, &sink))
		}
	}
	//init metrics
	pkg.RegisterMetric()
	http.Handle("/metrics", promhttp.Handler())
	server := &http.Server{
		Addr:              ":8081",
		ReadHeaderTimeout: 3 * time.Second,
	}
	go func() {
		err := server.ListenAndServe()
		if err != nil {
			panic(fmt.Errorf("start metrics server failed: %w", err))
		}
	}()
	//run
	run(conf, tasks)
}

func run(conf config.Config, tasks []task.Task) {
	t := time.NewTimer(time.Duration(conf.Global.Interval) * time.Minute)
	//init excutor
	excutor := task.NewExcutor(tasks)
	for {
		<-t.C
		if err := excutor.Excute(conf.Global.Delay); err != nil {
			log.Println(fmt.Errorf("excute task failed: %w", err))
		}
		t.Reset(time.Duration(conf.Global.Interval) * time.Minute)
	}
}