package processor

import (
	"encoding/csv"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

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

	"github.com/aws/aws-sdk-go-v2/service/s3"
)

var GetCheckPointFunc = func(fn string) (*common.CheckPoint, error) {
	cp, err := GetCheckPoint(fn)
	return cp, err
}

var writeDataFunc = func(object *s3.GetObjectOutput, outPath string, name string) error {
	return writeData(object, outPath, name)
}

func (p *DataProcessor) CleanData(bucket string, dir string, dataSource string, version string) error {
	outPath := fmt.Sprintf("%s/%s/%s", dir, dataSource, version)
	fn := fmt.Sprintf("%s/checkpoint.json", outPath)
	cp, err := GetCheckPointFunc(fn) // download all if no checkpoint
	timeNow := time.Now()
	ts := timeNow.Unix()
	if err != nil {
		if os.IsNotExist(err) {
			err = p.DownloadAndWriteData(bucket, dir, dataSource, version, "20") // start with 20
			if err != nil {
				return err
			}
			SaveCheckPoint(fn, ts)
			return nil
		} else {
			log.Printf("get checkpoint failed, err: %s\n", err.Error())
			return err
		}
	}
	// get start time point
	cpTs := cp.Timestamp
	for {
		if cpTs+1800 > ts {
			break
		}
		tm := time.Unix(cpTs, 0)
		date := tm.Format("20060102-15")
		minute := tm.Minute()
		err = p.DownloadAndWriteData(bucket, dir, dataSource, version, fmt.Sprintf("%s-%v", date, minute/30))
		cpTs += 1800
		SaveCheckPoint(fn, cpTs) // update checkpoint
	}
	return nil
}

func (p *DataProcessor) DownloadAndWriteData(bucket string, dir string, dataSource string, version string, dateTime string) error {
	outPath := fmt.Sprintf("%s/%s/%s", dir, dataSource, version)
	pathPrefix := fmt.Sprintf("raw-data/%s/%s/", dataSource, version)
	objects, err := p.cli.DownloadFiles(bucket, pathPrefix+dateTime)
	if err != nil {
		log.Printf("download data failed, err: %s\n", err.Error())
		return err
	}
	for k, o := range objects {
		err = writeDataFunc(o, outPath, strings.TrimPrefix(k, pathPrefix))
		if err != nil {
			log.Printf("clean data failed, err: %s\n", err.Error())
			return err
		}
	}

	return nil
}

func writeData(object *s3.GetObjectOutput, outPath string, name string) error {
	r := csv.NewReader(object.Body)
	fileName := fmt.Sprintf("%s/%s", outPath, name)
	if _, err := os.Stat(fileName); os.IsNotExist(err) {
		err = os.MkdirAll(outPath, 0700)
		if err != nil {
			return err
		}
	} else {
		return err
	}
	f, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE, os.ModePerm)
	if err != nil {
		return err
	}
	defer f.Close()
	w := csv.NewWriter(f)
	content, _ := r.ReadAll()
	for _, row := range content {
		err = WriteCsvFileFunc(w, row)
		if err != nil {
			return err
		}
	}
	stat, err := os.Stat(fileName)
	if err != nil {
		return err
	}
	log.Printf("write file %s done, size: %v", fileName, stat.Size())
	return nil
}