package tidbtools

import (
	"strconv"

	importintong "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/import_into/v1beta2"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
)

// DedicatedImport mirrors `DedicatedImport` in the FE types.
type DedicatedImport struct {
	ID                         string           `json:"id"`
	Name                       string           `json:"name,omitempty"`
	SourceType                 ImportSourceType `json:"source_type" enums:"LOCAL_FILE,S3,GCS"`
	FileType                   ImportFileType   `json:"file_type" enums:"CSV,SQL,PARQUET"`
	TargetDatabase             string           `json:"target_database"`
	TargetTable                string           `json:"target_table"`
	Filter                     string           `json:"filter,omitempty"`
	ConnMethod                 string           `json:"conn_method,omitempty"`
	Status                     ImportStatus     `json:"status" enums:"pending,preparing,running,succeeded,failed,canceled"`
	ErrorMsg                   string           `json:"error_msg,omitempty"`
	TotalRows                  int64            `json:"total_rows"`
	ImportedRows               int64            `json:"imported_rows"`
	SizeBytes                  int64            `json:"size_bytes"`
	TotalTablesCount           int32            `json:"total_tables_count,omitempty"`
	CompletedTables            int32            `json:"completed_tables,omitempty"`
	PendingTables              int32            `json:"pending_tables,omitempty"`
	CompletedPercent           int32            `json:"completed_percent,omitempty"`
	PostImportCompletedPercent int32            `json:"post_import_completed_percent,omitempty"`
	CreatedAt                  int64            `json:"created_at"`
	FinishedAt                 int64            `json:"finished_at,omitempty"`
}

// ListImportsResp is the response body for `GET .../imports` (Dedicated).
type ListImportsResp struct {
	Items []*DedicatedImport `json:"items"`
	Total int64              `json:"total"`
}

// PremiumImportIntoTask mirrors `PremiumImportIntoTask` in FE types.
type PremiumImportIntoTask struct {
	ID                         string       `json:"id"`
	Name                       string       `json:"name,omitempty"`
	SQLStatement               string       `json:"sql_statement"`
	Status                     ImportStatus `json:"status" enums:"pending,preparing,running,succeeded,failed,canceled"`
	ErrorMsg                   string       `json:"error_msg,omitempty"`
	SourceURI                  string       `json:"source_uri"`
	ConnMethod                 string       `json:"conn_method,omitempty"`
	CreatedAt                  int64        `json:"created_at"`
	FinishedAt                 int64        `json:"finished_at,omitempty"`
	ProgressPercent            int32        `json:"progress_percent"`
	TotalTablesCount           int32        `json:"total_tables_count,omitempty"`
	CompletedTables            int32        `json:"completed_tables,omitempty"`
	PendingTables              int32        `json:"pending_tables,omitempty"`
	CompletedPercent           int32        `json:"completed_percent,omitempty"`
	PostImportCompletedPercent int32        `json:"post_import_completed_percent,omitempty"`
}

// ListImportIntoTasksResp is the response body for `GET .../import-into-tasks`.
type ListImportIntoTasksResp struct {
	Items []*PremiumImportIntoTask `json:"items"`
	Total int64                    `json:"total"`
}

// ToListImportsResp converts a proto ListImportResp to the FE-facing
// ListImportsResp.
func ToListImportsResp(pb *dataflow.ListImportResp) *ListImportsResp {
	out := &ListImportsResp{Items: make([]*DedicatedImport, 0)}
	if pb == nil {
		return out
	}
	out.Total = pb.Total
	for _, imp := range pb.Imports {
		if imp == nil {
			continue
		}
		var createdAt int64
		if imp.CreatedAt != nil {
			createdAt = imp.CreatedAt.AsTime().Unix()
		}
		var targetDB, targetTable, filter string
		if cr := imp.ImportCreateReq; cr != nil {
			filter = cr.GetFilter()
			if len(cr.TargetTableInfos) > 0 && cr.TargetTableInfos[0] != nil {
				if t := cr.TargetTableInfos[0].GetTargetTable(); t != nil {
					targetDB = t.Schema
					targetTable = t.Table
				}
			}
		}
		out.Items = append(out.Items, &DedicatedImport{
			ID:                         strconv.FormatUint(imp.Id, 10),
			Name:                       imp.Name,
			SourceType:                 importSourceType(dataflow.ImportDataSource(imp.DataSourceType)),
			FileType:                   importDataFormat(dataflow.ImportDataFormat(imp.DataFormat)),
			TargetDatabase:             targetDB,
			TargetTable:                targetTable,
			Filter:                     filter,
			Status:                     importStatus(imp.Status),
			ErrorMsg:                   imp.Message,
			SizeBytes:                  int64(imp.TotalSize),
			TotalTablesCount:           int32(imp.TotalTablesCount),
			CompletedTables:            int32(imp.CompletedTables),
			PendingTables:              int32(imp.PendingTables),
			CompletedPercent:           int32(imp.CompletedPercent),
			PostImportCompletedPercent: int32(imp.PostImportCompletedPercent),
			CreatedAt:                  createdAt,
		})
	}
	return out
}

func importStatus(s dataflow.ImportGetResp_Status) ImportStatus {
	switch s {
	case dataflow.ImportGetResp_PREPARING:
		return ImportStatusPreparing
	case dataflow.ImportGetResp_IMPORTING:
		return ImportStatusRunning
	case dataflow.ImportGetResp_COMPLETED:
		return ImportStatusSucceeded
	case dataflow.ImportGetResp_FAILED:
		return ImportStatusFailed
	case dataflow.ImportGetResp_CANCELING, dataflow.ImportGetResp_CANCELED:
		return ImportStatusCanceled
	default:
		return ""
	}
}

func importSourceType(t dataflow.ImportDataSource) ImportSourceType {
	switch t {
	case dataflow.ImportDataSource_S3_SOURCE:
		return ImportSourceTypeS3
	case dataflow.ImportDataSource_GCS_SOURCE:
		return ImportSourceTypeGCS
	case dataflow.ImportDataSource_LOCAL_SOURCE:
		return ImportSourceTypeLocalFile
	default:
		return ""
	}
}

func importDataFormat(f dataflow.ImportDataFormat) ImportFileType {
	switch f {
	case dataflow.ImportDataFormat_SQL:
		return ImportFileTypeSQL
	case dataflow.ImportDataFormat_CSV:
		return ImportFileTypeCSV
	case dataflow.ImportDataFormat_PARQUET:
		return ImportFileTypeParquet
	default:
		return ""
	}
}

// GetImportIntoTaskResp is the response body for `GET .../import-into-tasks/{task_id}`.
type GetImportIntoTaskResp = PremiumImportIntoTask

// ToListImportIntoTasksResp converts an ImportIntoTasksResponse to the FE-facing DTO.
func ToListImportIntoTasksResp(pb *importintong.ListImportIntoTasksResponse) *ListImportIntoTasksResp {
	out := &ListImportIntoTasksResp{Items: make([]*PremiumImportIntoTask, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.TotalSize)
	for _, t := range pb.Imports {
		if t == nil {
			continue
		}
		out.Items = append(out.Items, toImportIntoTask(t))
	}
	return out
}

// ToGetImportIntoTaskResp converts an ImportIntoTask to the FE-facing detail.
func ToGetImportIntoTaskResp(pb *importintong.ImportIntoTask) *GetImportIntoTaskResp {
	if pb == nil {
		return &GetImportIntoTaskResp{}
	}
	return toImportIntoTask(pb)
}

func toImportIntoTask(t *importintong.ImportIntoTask) *PremiumImportIntoTask {
	if t == nil {
		return &PremiumImportIntoTask{}
	}
	out := &PremiumImportIntoTask{
		ID:              t.ImportId,
		Status:          importIntoStateToStatus(t.State),
		ProgressPercent: t.CompletePercent,
	}
	// Message is a general output field — only expose as ErrorMsg when the
	// task has actually failed, to avoid showing success messages in red.
	if out.Status == ImportStatusFailed {
		out.ErrorMsg = t.Message
	}
	if t.Name != nil {
		out.Name = *t.Name
	}
	if t.CreateTime != nil {
		out.CreatedAt = t.CreateTime.AsTime().Unix()
	}
	if t.CompleteTime != nil {
		out.FinishedAt = t.CompleteTime.AsTime().Unix()
	}
	if cd := t.CreationDetails; cd != nil {
		if src := cd.GetSource(); src != nil {
			out.SourceURI = src.Uri
		}
	}
	// Count subtasks for table-level progress.
	var completed, pending int32
	for _, st := range t.Subtasks {
		if st == nil {
			continue
		}
		switch st.Status {
		case importintong.ImportSubTaskStatus_ENUM_FINISHED:
			completed++
		case importintong.ImportSubTaskStatus_ENUM_PENDING, importintong.ImportSubTaskStatus_ENUM_RUNNING:
			pending++
		}
	}
	out.TotalTablesCount = int32(len(t.Subtasks))
	out.CompletedTables = completed
	out.PendingTables = pending
	return out
}

func importIntoStateToStatus(s importintong.ImportState_Enum) ImportStatus {
	switch s {
	case importintong.ImportState_ENUM_PREPARING:
		return ImportStatusPreparing
	case importintong.ImportState_ENUM_IMPORTING, importintong.ImportState_ENUM_SCALING_UP:
		return ImportStatusRunning
	case importintong.ImportState_ENUM_COMPLETED:
		return ImportStatusSucceeded
	case importintong.ImportState_ENUM_FAILED:
		return ImportStatusFailed
	case importintong.ImportState_ENUM_CANCELING, importintong.ImportState_ENUM_CANCELED:
		return ImportStatusCanceled
	default:
		return ImportStatusPending
	}
}