package dm

import (
	"bytes"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"strconv"

	"github.com/golang/protobuf/jsonpb"
	tiflowdmpb "github.com/pingcap/tiflow/dm/pb"
	"github.com/pingcap/tiflow/engine/framework/model"
	frameModel "github.com/pingcap/tiflow/engine/framework/model"
	"github.com/pingcap/tiflow/engine/jobmaster/dm"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"go.uber.org/zap"
)

const Component = "component"

func FnvHash(raw string) string {
	hash := fnv.New32()
	// fnv.New32().Write() never return error
	_, _ = hash.Write([]byte(raw))
	return fmt.Sprintf("%x", hash.Sum32())
}

func FormatBucketName(job *entity.DFJob) string {
	const (
		labelLengthLimit = 63
		hashSize         = 8
		resultLimit      = labelLengthLimit - hashSize - 1
	)
	result := fmt.Sprintf("dm-%s-%d", job.Region, job.ProjectID.Uint64())
	if len(result) > resultLimit {
		result = result[:resultLimit]
	}

	return fmt.Sprintf("%s-%s", result, FnvHash(result))
}

func generateDataflowSubDir(jobID uint64) string {
	return "dataflow-" + strconv.FormatUint(jobID, 10)
}

func generateLoaderDir(job *entity.DFJob, connProfileID uint64) string {
	scheme := "s3://"
	if job == nil {
		return scheme + "invalid"
	}
	if job.Provider == cloudv1.ProviderTypeGKE {
		scheme = "gs://"
	}
	jobID := job.ID.Uint64()
	if jobID == 0 || connProfileID == 0 {
		// shouldn't happen
		return scheme + "invalid"
	}
	dataflowSubDir := "/" + generateDataflowSubDir(jobID)
	sourceSubDir := "/source-" + strconv.FormatUint(connProfileID, 10)
	return scheme + FormatBucketName(job) + dataflowSubDir + sourceSubDir + "/"
}

type taskStatus struct {
	finishedUnit model.WorkerType
	dumpStatus   *entity.DMDumpStatus
	loadStatus   *entity.DMLoadStatus
	syncStatus   *tiflowdmpb.SyncStatus
}

// detail2Status transfers job status from engine to job information in dm model
func detail2Status(logger *zap.Logger, jobStatus *dm.JobStatus, idStr string) (*taskStatus, error) {
	transferredStatus := &taskStatus{
		finishedUnit: frameModel.WorkerDMSync, // unit keeps the last finished unit, equals to Sync if there is no finished unit
	}
	if jobStatus.FinishedUnitStatus != nil {
		for _, finishedStatus := range jobStatus.FinishedUnitStatus[idStr] {
			switch finishedStatus.Unit {
			case frameModel.WorkerDMDump:
				statusBytes, _ := finishedStatus.Status.MarshalJSON()
				dumpStatus := &entity.DMDumpStatus{}
				if err := json.Unmarshal(statusBytes, dumpStatus); err != nil {
					logger.Error("failed to unmarshal dump status", zap.Error(err))
					return nil, err
				}
				dumpStatus.Duration = int64(finishedStatus.Duration.Seconds())
				transferredStatus.dumpStatus = dumpStatus
				transferredStatus.finishedUnit = frameModel.WorkerDMDump
			case frameModel.WorkerDMLoad:
				statusBytes, _ := finishedStatus.Status.MarshalJSON()
				loadStatus := &entity.DMLoadStatus{}
				if err := json.Unmarshal(statusBytes, loadStatus); err != nil {
					logger.Error("failed to unmarshal load status", zap.Error(err))
					return nil, err
				}
				loadStatus.Duration = int64(finishedStatus.Duration.Seconds())
				transferredStatus.loadStatus = loadStatus
				transferredStatus.finishedUnit = frameModel.WorkerDMLoad // won't happen a status in all mode have both dump and load unit
			}
		}
	}
	if jobStatus.TaskStatus != nil {
		// transfer sync unit status
		if taskStatus, ok := jobStatus.TaskStatus[idStr]; ok && taskStatus.Status != nil && taskStatus.Status.Unit == frameModel.WorkerDMSync {
			statusBytes, _ := taskStatus.Status.Status.MarshalJSON()
			syncStatus := &tiflowdmpb.SyncStatus{}
			if err := jsonpb.Unmarshal(bytes.NewReader(statusBytes), syncStatus); err != nil {
				logger.Error("failed to unmarshal sync status", zap.Error(err))
				return nil, err
			}
			transferredStatus.syncStatus = syncStatus
		}
	}
	return transferredStatus, nil
}

func getDMJobName(dmJobID uint64) string {
	return fmt.Sprintf("dm-job-%d", dmJobID)
}

func convertUint64ToID(ids []uint64) []entity.ID {
	result := make([]entity.ID, 0, len(ids))
	for _, id := range ids {
		result = append(result, entity.ID(id))
	}
	return result
}