package processor

import (
	"log"

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

type DataTask struct {
	p DataProcessor
}

func NewTask(region string) DataTask {
	p := NewDataProcessor(region)
	return DataTask{
		p: *p,
	}
}

func (t *DataTask) CleanAndProcessData(bucket string, dataSources []string, version string) {
	for _, ds := range dataSources {
		// clean data
		log.Printf("start clean %s data", ds)
		err := t.p.CleanData(bucket, common.CleanedDataPath, ds, version)
		if err != nil {
			log.Printf("clean %s data failed, error: %v", ds, err)
			failedToCleanData(ds, version)
		} else {
			succeedToCleanData(ds, version)
		}
		log.Printf("end clean %s data", ds)
		// process cleaned data
		log.Printf("start process %s data", ds)
		err = t.p.ProcessCleanedData(common.CleanedDataPath, common.ProcessedDataPath, ds, version)
		if err != nil {
			log.Printf("process %s data failed, error: %v", ds, err)
			failedToProcessData(ds, version)
		} else {
			succeedToProcessData(ds, version)
		}
		log.Printf("end process %s data", ds)
		// upload processed data
		log.Printf("start upload %s processed data", ds)
		err = t.p.UploadProcessedData(bucket, common.ProcessedDataPath, ds, version)
		if err != nil {
			log.Printf("upload %s processed data failed, error: %v", ds, err)
			failedToUploadData(ds, version)
		} else {
			succeedToUploadData(ds, version)
		}
		log.Printf("end upload %s processed data", ds)
	}
}

func (t *DataTask) MergeData(bucket string, evaBucket string, dataSources []string, version string) {
	for _, ds := range dataSources {
		//merge processed data
		log.Printf("start merge %s processed data", ds)
		err := t.p.MergeProcessedData(bucket, common.MergedDataPath, ds, version)
		if err != nil {
			log.Printf("merge %s processed data failed, error: %v", ds, err)
			failedToProcessData(ds, version)
		} else {
			succeedToProcessData(ds, version)
		}
		log.Printf("end merge %s processed data", ds)
		//upload merged data
		log.Printf("start upload %s merged data", ds)
		err = t.p.UploadMergedData(evaBucket, common.MergedDataPath, ds, version)
		if err != nil {
			log.Printf("upload %s merged data failed, error: %v", ds, err)
			failedToUploadData(ds, version)
		} else {
			succeedToUploadData(ds, version)
		}
		log.Printf("end upload %s merged data", ds)
	}
}

func (t *DataTask) JoinData(bucket string, evaBucket string, dataSources []string, version string) {
	for _, ds := range dataSources {
		//merge processed data
		log.Printf("start download %s row data", ds)
		err := t.p.CleanData(bucket, common.RawDataPath, ds, version)
		if err != nil {
			log.Printf("download %s row data failed, error: %v", ds, err)
			failedToProcessData(ds, version)
		} else {
			succeedToProcessData(ds, version)
		}
		log.Printf("end download %s row data", ds)

		log.Printf("start join %s row data", ds)
		err = t.p.JoinCleanedData(common.RawDataPath, common.JoinedDataPath, ds, version)
		if err != nil {
			log.Printf("join %s row data failed, error: %v", ds, err)
			failedToProcessData(ds, version)
		} else {
			succeedToProcessData(ds, version)
		}
		log.Printf("end join %s row data", ds)

		//upload joined data
		log.Printf("start upload %s joined data", ds)
		err = t.p.UploadJoinedData(evaBucket, common.JoinedDataPath, ds, version)
		if err != nil {
			log.Printf("upload %s joined data failed, error: %v", ds, err)
			failedToUploadData(ds, version)
		} else {
			succeedToUploadData(ds, version)
		}
		log.Printf("end upload %s joined data", ds)
	}
}

func failedToCleanData(dataSource string, version string) {
	pkg.ProcessorFailedToCleanDataMetric.WithLabelValues(dataSource, version).Inc()
}

func succeedToCleanData(dataSource string, version string) {
	pkg.ProcessorSucceedToCleanDataMetric.WithLabelValues(dataSource, version).Inc()
}

func failedToProcessData(dataSource string, version string) {
	pkg.ProcessorFailedToProcessDataMetric.WithLabelValues(dataSource, version).Inc()
}

func succeedToProcessData(dataSource string, version string) {
	pkg.ProcessorSucceedToProcessDataMetric.WithLabelValues(dataSource, version).Inc()
}

func failedToUploadData(dataSource string, version string) {
	pkg.ProcessorFailedToUploadDataMetric.WithLabelValues(dataSource, version).Inc()
}

func succeedToUploadData(dataSource string, version string) {
	pkg.ProcessorSucceedToUploadDataMetric.WithLabelValues(dataSource, version).Inc()
}