package processor

import (
	"encoding/csv"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"time"

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

// readDir is the directory for raw data
// writeDir is the directory for joined data
func (p *DataProcessor) JoinCleanedData(readDir string, writeDir string, dataSource string, version string) error {
	var records = make(map[string]map[string]string)
	if dataSource != common.OSSInsight {
		return nil
	}
	patternA := fmt.Sprintf("%s/%s/%s/*_explorer_questions.csv", readDir, dataSource, version)
	patternB := fmt.Sprintf("%s/%s/%s/*_explorer_question_feedbacks.csv", readDir, dataSource, version)
	err := readFileToMapOssInsightV2(patternA, patternB, records)
	if err != nil {
		log.Printf("read file to map failed, err: %s", err.Error())
		return err
	}

	err = writeMapToCsvFileV2(writeDir, dataSource, version, records)
	if err != nil {
		log.Printf("write map to file failed, err: %s", err.Error())
	}
	defer func() {
		err = DeleteFilesFunc(patternA)
		if err != nil {
			log.Printf("delete cleaned files %s failed, error: %v", patternA, err)
		}
		err = DeleteFilesFunc(patternB)
		if err != nil {
			log.Printf("delete cleaned files %s failed, error: %v", patternB, err)
		}
	}()
	return err
}

func readFileToMapOssInsightV2(patternA string, patternB string, mp map[string]map[string]string) error {
	mpA := make(map[string]map[string]string)
	mpB := make(map[string]map[string]string)

	files, _ := filepath.Glob(patternA)
	log.Printf("read file %s, len: %v", patternA, len(files))
	for _, file := range files {
		err := readFileToMapV2Func(file, mpA, "id")
		if err != nil {
			return err
		}
	}
	// read explorer_question_feedbacks files
	files, _ = filepath.Glob(patternB)
	log.Printf("read file %s, len: %v", patternB, len(files))
	for _, file := range files {
		err := readFileToMapV2Func(file, mpB, "question_id")
		if err != nil {
			return err
		}
	}

	// merge two maps according to id==question_id
	for id, v := range mpA {
		mp[id] = make(map[string]string)
		for k1, v1 := range v {
			mp[id][k1] = v1 // Read each field from the question file
		}
		for k2, v2 := range mpB[id] {
			if k2 == "user_id" || k2 == "created_at" {
				continue
			}
			mp[id][k2] = v2 // Read each field from the feedback file except user_id/created_at
		}
		resultType := getResultType(mp[id]["result"]) // Parse result_type and add it to the map
		mp[id]["result_type"] = string(resultType)
	}
	return nil
}

// readFileToMap read fields of the row in the file into the map
// selectedID is an id selected to connect two tables, such as session_id or question_id
func readFileToMapV2(path string, mp map[string]map[string]string, selectedID string) error {
	file, err := os.Open(path)
	if err != nil {
		return err
	}
	defer file.Close()

	r := csv.NewReader(file)
	// read head
	head, err := r.Read()
	if err != nil {
		return err
	}
	var idIndex int
	for i, h := range head {
		if h == selectedID { // find the selectedID index
			idIndex = i
		}
	}
	// start read
	for {
		row, err := r.Read()
		if err != nil && err != io.EOF {
			return err
		}
		if err == io.EOF {
			break
		}
		id := row[idIndex] // get id
		if _, ok := mp[id]; !ok {
			mp[id] = make(map[string]string)
		}
		for i, h := range head {
			if h == selectedID {
				continue
			}
			mp[id][h] = row[i]
		}
	}
	return nil
}

var readFileToMapV2Func = func(path string, mp map[string]map[string]string, selectedID string) error {
	return readFileToMapV2(path, mp, selectedID)
}

// writeMapToCsvFile selects the required fields content from the map according to the corresponding title,
// and then writes them to the file
func writeMapToCsvFileV2(dir string, dataSource string, version string, mp map[string]map[string]string) (err error) {
	// Required fields
	title := []string{"question_id", "user_id", "title", "status", "query_sql",
		"result", "result_type", "feedback_type", "feedback_content", "created_at"}

	// Extract these fields from the map
	for k, v := range mp {
		// v is the content corresponding to a certain id
		row := make([]string, len(title))
		// row[0] saves the id such as session_id or question_id
		row[0] = k
		for i := 1; i < len(title); i++ {
			if a, ok := v[title[i]]; ok {
				row[i] = a
			}
		}
		// Parse the time and write the row to the corresponding file based on the time
		timeStr := v["created_at"]
		t, _ := time.ParseInLocation("2006-01-02 15:04:05", timeStr, time.UTC)
		date := t.Format("20060102")

		outPath := fmt.Sprintf("%s/%s/%s/", dir, dataSource, version)
		fn := fmt.Sprintf("%s/%s/%s/%s.csv", dir, dataSource, version, date)

		var w *csv.Writer
		var file *os.File
		defer file.Close()
		if _, err = os.Stat(fn); os.IsNotExist(err) { // If the file does not exist, create it first and write the title
			err = os.MkdirAll(outPath, 0700)
			if err != nil {
				return
			}
			file, err = os.Create(fn)
			if err != nil {
				return
			}
			w = csv.NewWriter(file)
			err = WriteCsvFileFunc(w, title)
			if err != nil {
				return
			}
		} else {
			file, err = os.OpenFile(fn, os.O_WRONLY|os.O_APPEND|os.O_CREATE, os.ModePerm)
			if err != nil {
				return
			}
			w = csv.NewWriter(file)
		}

		err = WriteCsvFileFunc(w, row)
		if err != nil {
			return
		}
	}
	log.Printf("write to file done, length %v", len(mp))

	return
}

func getResultType(result string) (resultType common.ResultType) {
	if result == "null" {
		return common.Null
	}
	if len(result) == 0 {
		return common.Empty
	}
	r := common.OssInsightResult{}
	err := json.Unmarshal([]byte(result), &r)
	if err != nil {
		return
	}
	if len(r.Rows) == 0 {
		return common.ResultEmpty
	}
	return common.Normal
}