package processor

import (
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"time"

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

func (p *DataProcessor) UploadMergedData(bucket string, dir string, dataSource string, version string) (err error) {
	pattern := fmt.Sprintf("%s/%s/%s/*.csv", dir, dataSource, version)
	files, err := filepath.Glob(pattern)
	if err != nil {
		return
	}
	log.Printf("start upload merged data %s, len: %v", pattern, len(files))
	for _, f := range files {
		fi, _ := os.Stat(f)
		key := fmt.Sprintf("dataset/%s/%s/%s", dataSource, version, fi.Name())
		log.Printf("upload to %s/%s", bucket, key)
		if err = p.cli.UploadFile(f, bucket, key); err != nil {
			return
		}
		log.Printf("upload to %s", key)
		os.Remove(f)
	}
	return
}

func (p *DataProcessor) UploadProcessedData(bucket string, dir string, dataSource string, version string) error {
	pattern := fmt.Sprintf("%s/%s/%s/*.csv", dir, dataSource, version)
	files, _ := filepath.Glob(pattern)
	log.Printf("start upload data %s, len: %v", pattern, len(files))
	for _, f := range files {
		fi, err := os.Stat(f)
		if err != nil {
			return err
		}
		now := time.Now()
		date := now.Format("20060102")
		clusterID := fi.Name()[0:strings.Index(fi.Name(), ".csv")]
		if _, ok := common.GetClusterIDMap(dataSource)[date][clusterID]; !ok {
			common.GetClusterIDMap(dataSource)[date] = make(map[string]bool)
		}
		common.GetClusterIDMap(dataSource)[date][clusterID] = true
		key := fmt.Sprintf("processed-data/%s/%s/%s/%s-%d.csv", dataSource, version, clusterID, date, time.Now().Unix())
		if err = p.cli.UploadFile(f, bucket, key); err != nil {
			return err
		}
		log.Printf("upload to %s", key)
		os.Remove(f)
	}
	return nil
}

func (p *DataProcessor) UploadJoinedData(bucket string, dir string, dataSource string, version string) error {
	pattern := fmt.Sprintf("%s/%s/%s/*.csv", dir, dataSource, version)
	files, _ := filepath.Glob(pattern)
	log.Printf("start upload data %s, len: %v", pattern, len(files))
	for _, f := range files {
		fi, err := os.Stat(f)
		if err != nil {
			return err
		}
		date := fi.Name()[0:strings.Index(fi.Name(), ".csv")]
		key := fmt.Sprintf("dataset/joined-data/%s/%s/%s-%d.csv", dataSource, version, date, time.Now().Unix())
		if err = p.cli.UploadFile(f, bucket, key); err != nil {
			return err
		}
		log.Printf("upload to %s", key)
		os.Remove(f)
	}
	return nil
}

func (p *DataProcessor) UploadConstructedDB(bucket string, dir string, dataSource string, version string) error {
	pattern := fmt.Sprintf("%s/%s/%s/*", dir, dataSource, version) // id
	dirs, _ := filepath.Glob(pattern)
	log.Printf("start upload data %s, len: %v", pattern, len(dirs))
	for _, d := range dirs {
		fi, _ := os.Stat(d)
		id := fi.Name()
		if id == "checkpoint.json" {
			continue
		}
		path := fmt.Sprintf("%s/%s/%s/%s/*", dir, dataSource, version, id)
		file, _ := filepath.Glob(path)
		for _, f := range file {
			fi, _ = os.Stat(f)
			key := fmt.Sprintf("dataset/constructed-db/%s/%s/%s/%s", dataSource, version, id, fi.Name())
			if err := p.cli.UploadFile(f, bucket, key); err != nil {
				return err
			}
		}
		os.RemoveAll(d)
	}
	return nil
}