package pimport

import (
	"context"
	"encoding/json"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

	"github.com/gogo/protobuf/types"
	"github.com/pingcap/dbaas/central/db/entity"
	pimport "github.com/pingcap/dbaas/central/model/import"
	"github.com/pingcap/dbaas/central/service"
	"github.com/pingcap/errcode"
	"github.com/pingcap/errors"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

type importService struct {
	central.UnimplementedImportServiceServer
	service.OperationsService
	importModel pimport.Model
}

// NewService constructs a importService
func NewService(s service.OperationsService, model pimport.Model) central.ImportServiceServer {
	return &importService{OperationsService: s, importModel: model}
}

func (s *importService) userImport(ctx context.Context) pimport.UserContext {
	user := s.CurrentUser(ctx)
	return s.importModel.User(ctx, user)
}

func (s *importService) Log(ctx context.Context) *zap.Logger {
	return log.AppendCtxFields(ctx, s.userImport(ctx).Log())
}

func restoreSpecialChars(ctx context.Context, str string) string {
	logger := log.WithContextS(ctx)

	// https://docs.pingcap.com/tidb/stable/migrate-from-csv-using-tidb-lightning
	replaced := make([]rune, 0, len(str))
	s := []byte(str)
	r := 0
	for r < len(s) {
		if s[r] != '\\' {
			c, size := utf8.DecodeRune(s[r:])
			r += size
			replaced = append(replaced, c)
			continue
		}
		r += 1
		if r >= len(s) {
			logger.Error("Escape sequence at end of string.")
			return ""
		}
		switch s[r] {
		default:
			logger.Error("Expected valid escape code after \\, but got %q.", s[r])
			return ""
		case ' ', '\t':
			logger.Error("invalid escape: '\\%c'", s[r])
			return ""
		case 'b':
			replaced = append(replaced, rune(0x0008))
			r += 1
		case 't':
			replaced = append(replaced, rune(0x0009))
			r += 1
		case 'n':
			replaced = append(replaced, rune(0x000A))
			r += 1
		case 'f':
			replaced = append(replaced, rune(0x000C))
			r += 1
		case 'r':
			replaced = append(replaced, rune(0x000D))
			r += 1
		case '"':
			replaced = append(replaced, rune(0x0022))
			r += 1
		case '\\':
			replaced = append(replaced, rune(0x005C))
			r += 1
		case 'u':
			// At this point, we know we have a Unicode escape of the form
			// `uXXXX` at [r, r+5). (Because the lexer guarantees this
			// for us.)
			escaped := asciiEscapeToUnicode(ctx, s[r+1:r+5])
			replaced = append(replaced, escaped)
			r += 5
		case 'U':
			// At this point, we know we have a Unicode escape of the form
			// `uXXXX` at [r, r+9). (Because the lexer guarantees this
			// for us.)
			escaped := asciiEscapeToUnicode(ctx, s[r+1:r+9])
			replaced = append(replaced, escaped)
			r += 9
		}
	}
	return string(replaced)
}

func asciiEscapeToUnicode(ctx context.Context, bs []byte) rune {
	logger := log.WithContextS(ctx)

	s := string(bs)
	hex, err := strconv.ParseUint(strings.ToLower(s), 16, 32)
	if err != nil {
		logger.Error("Could not parse '%s' as a hexadecimal number, but the lexer claims it's OK: %s", s, err)
	}
	if !utf8.ValidRune(rune(hex)) {
		logger.Panic("Escaped character '\\u%s' is not valid UTF-8.", s)
	}
	return rune(hex)
}

func buildImport(ctx context.Context, req *central.ImportCreateReq) (*entity.Import, error) {
	imp := &entity.Import{
		ClusterID:    entity.ID(req.ClusterId),
		SourceURL:    req.SourceUrl,
		BucketRegion: req.Region,
	}
	imp.DataFormat = entity.DataFormat(req.DataFormat)
	imp.DataSourceType = entity.DataSource(req.DataSourceType)
	imp.TargetUsername = req.TargetUsername
	imp.TargetPassword = req.TargetPassword
	imp.Filter = req.Filter
	imp.AwsRoleArn = req.AwsRoleArn

	var csvConfig string
	if req.CsvFormat != nil {
		req.CsvFormat.Delimiter = restoreSpecialChars(ctx, req.CsvFormat.Delimiter)
		req.CsvFormat.Separator = restoreSpecialChars(ctx, req.CsvFormat.Separator)
		req.CsvFormat.Null = restoreSpecialChars(ctx, req.CsvFormat.Null)

		csvConfigBytes, err := json.Marshal(req.CsvFormat)
		if err != nil {
			log.Error(ctx, "failed to marshal csv config", zap.Error(err))
			return nil, err
		}

		csvConfig = string(csvConfigBytes)
	}
	if csvConfig == "" {
		csvConfig = "{}"
	}

	imp.CustomCSVFormat = csvConfig
	imp.OptionalArgs = &entity.OptionalArgs{
		ObjectNamePattern:  req.ObjectNamePattern,
		TargetTablePattern: req.TargetTablePattern,
	}

	return imp, nil
}

func wrapErr(err error, msg string) error {
	if _, ok := err.(errcode.ErrorCode); ok {
		return err
	}

	return errcode.NewInternalErr(errors.New(msg))
}

func (s *importService) Create(ctx context.Context, req *central.ImportCreateReq) (*central.ClusterImportId, error) {
	orgID := entity.ID(req.OrgId)
	projectID := entity.ID(req.ProjectId)
	userCtx := s.userImport(ctx)
	err := userCtx.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	logger := s.Log(ctx)
	pass := req.TargetPassword
	req.TargetPassword = ""
	logger.Info("create import", zap.Reflect("request", req))
	req.TargetPassword = pass

	err = validateReq(req)
	if err != nil {
		return nil, errcode.NewInvalidInputErr(err)
	}

	imp, err := buildImport(ctx, req)
	if err != nil {
		logger.Error("failed to build import", zap.Reflect("req", req), zap.Reflect("import", imp), zap.Error(err))
		return nil, wrapErr(err, "failed to create import")
	}

	id, err := userCtx.CreateImport(ctx, orgID, projectID, imp)
	if err != nil {
		logger.Error("failed to create import", zap.Reflect("req", req), zap.Reflect("import", imp), zap.Error(err))
		return nil, err
	}

	resp := &central.ClusterImportId{
		ClusterId: req.ClusterId,
		Id:        id,
	}

	logger.Info("create import success", zap.Reflect("response", resp))
	return resp, nil
}

func (s *importService) Get(ctx context.Context, in *central.ClusterImportId) (*central.ImportGetResp, error) {
	orgID := entity.ID(in.OrgId)
	projectID := entity.ID(in.ProjectId)
	userCtx := s.userImport(ctx)
	err := userCtx.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	logger := s.Log(ctx)
	logger.Info("get import", zap.Reflect("request", in))

	imp, err := userCtx.GetImport(in)
	if err != nil {
		logger.Error("failed to get import", zap.Uint64("cluster id", in.ClusterId), zap.Uint64("import id", in.Id), zap.Error(err))
		if errors.Cause(err) == gorm.ErrRecordNotFound {
			return nil, errcode.NewNotFoundErr(err)
		}
		return nil, errcode.NewInternalErr(errors.Errorf("failed to get import"))
	}

	resp := &central.ImportGetResp{
		ClusterId:        in.ClusterId,
		SourceUrl:        imp.SourceURL,
		CompletedTables:  imp.CompletedTables,
		PendingTables:    imp.PendingTables,
		CompletedPercent: imp.CompletedPercent,
		DataFormat:       uint32(imp.DataFormat),
		DataSourceType:   uint32(imp.DataSourceType),
	}

	createdAt, err := types.TimestampProto(imp.CreatedAt)
	if err != nil {
		logger.Error("failed to parse import's CreatedAt field", zap.Uint64("id", uint64(imp.ID)), zap.Error(err))
	} else {
		resp.CreatedAt = createdAt
	}

	if imp.TotalSize != nil {
		resp.TotalSize = *imp.TotalSize
	}
	if imp.TotalFiles != nil {
		resp.TotalFiles = *imp.TotalFiles
	}

	switch imp.Status {
	case entity.ImportStatusPreparing:
		resp.Status = central.ImportGetResp_PREPARING
	case entity.ImportStatusImporting:
		resp.Status = central.ImportGetResp_IMPORTING
	case entity.ImportStatusCompleted:
		resp.CompletedPercent = 100
		resp.Status = central.ImportGetResp_COMPLETED
	case entity.ImportStatusFailed:
		resp.Status = central.ImportGetResp_FAILED
	case entity.ImportStatusCanceled:
		resp.Status = central.ImportGetResp_CANCELED
	case entity.ImportStatusCanceling:
		resp.Status = central.ImportGetResp_CANCELING
	default:
		logger.Error("unknown status", zap.Stringer("status", imp.Status))
		return nil, errcode.NewInternalErr(errors.New("failed to get import"))
	}

	if imp.StartedAt != nil {
		end := time.Now()
		if imp.FinishedAt != nil {
			end = *imp.FinishedAt
		}

		resp.ElapsedTimeSeconds = uint32(end.Sub(*imp.StartedAt).Seconds())
	}

	resp.Message = PrettyMessage(imp.Message)

	for _, t := range imp.CurrentTables.Tables {
		respT := &central.CurrentTable{
			Name:             t.Name,
			Size:             t.Size,
			CompletedPercent: t.CompletedPercent,
		}
		resp.CurrentTables = append(resp.CurrentTables, respT)
	}

	logger.Info("get import success", zap.Reflect("response", resp))
	return resp, nil
}

func (s *importService) Cancel(ctx context.Context, in *central.ClusterImportId) (*types.Empty, error) {
	orgID := entity.ID(in.OrgId)
	projectID := entity.ID(in.ProjectId)
	userCtx := s.userImport(ctx)
	err := userCtx.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	logger := s.Log(ctx)
	logger.Info("cancel import", zap.Reflect("request", in))

	err = userCtx.CancelImport(ctx, in, false /* waitCanceled */)
	if err != nil {
		logger.Error("failed to cancel import", zap.Reflect("request", in), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Errorf("failed to cancel import"))
	}

	logger.Info("cancel import success")
	return &types.Empty{}, nil
}

func (s *importService) Exit(ctx context.Context, in *central.ClusterImportId) (*types.Empty, error) {
	orgID := entity.ID(in.OrgId)
	projectID := entity.ID(in.ProjectId)
	userCtx := s.userImport(ctx)
	err := userCtx.CheckOrgIDAndProjectID(orgID, projectID)
	if err != nil {
		return nil, err
	}

	logger := s.Log(ctx)
	logger.Info("exit import", zap.Reflect("request", in))

	err = userCtx.ExitImport(in)
	if err != nil {
		logger.Error("failed to exit import", zap.Reflect("request", in), zap.Error(err))
		return nil, errcode.NewInternalErr(errors.Errorf("failed to exit import"))
	}

	logger.Info("exit import success")
	return &types.Empty{}, nil
}

// the error from lightning contains the stack, so we trim it.
func PrettyMessage(msg string) string {
	idx := strings.Index(msg, "github.com")
	if idx != -1 {
		msg = msg[:idx]
	}

	return strings.TrimSpace(msg)
}