package branch

import (
	stderr "errors"
	"fmt"
	"strconv"
	"time"

	"github.com/pingcap/errors"

	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tikv"
)

func (r *reconcileContext) needRestoreKeyspace() bool {
	// If AnnoKeyBranchRestoreId is not set, a restore is needed.
	if _, ok := r.branch.Annotations[types.AnnoKeyBranchRestoreId]; !ok {
		return true
	}

	// If AnnoKeyBranchErrorAndRecreate is missing, val will be an empty string.
	// strconv.ParseBool("") returns (false, error), so 'need' will be false.
	val := r.branch.Annotations[types.AnnoKeyBranchErrorAndRecreate]
	need, _ := strconv.ParseBool(val)
	return need
}

func (r *reconcileContext) createRestoreKeyspace() error {
	var parentTimestamp time.Time
	if !r.onlineSnapshotRestore() {
		var err error
		parentTimestamp, err = r.getParentTimestamp()
		if err != nil {
			return err
		}
	}

	parentKeyspace, err := r.getParentKeyspace()
	if err != nil {
		return err
	}

	restoreId, err := r.createRestoreTaskFromParent(parentKeyspace, parentTimestamp)
	if err != nil {
		if stderr.Is(err, tikv.ErrRestoreConcurrencyLimit) {
			return &utils.RequeueAfterError{
				Cause:        err,
				RequeueAfter: time.Second * 5, //nolint:mnd // mask
			}
		}
		return errors.Wrap(err, "failed to create restore task")
	}

	r.branch.Annotations[types.AnnoKeyBranchRestoreId] = strconv.FormatUint(restoreId, 10)
	r.branch.Annotations[types.AnnoKeyBranchErrorAndRecreate] = "false"
	return errors.Annotate(r.InfraClient.Update(r.ctx, r.branch), "failed to update AnnoKeyBranchRestoreId")
}

// tikv-api GET
// 200 但 payload 为 Error，创建一个新的 restore
// 404 restore 已经被清理了（cse 重启或者 cse 自动清理机制），创建一个新的 restore
// 500 和其他 400 错误，持续重试并人工介入
func (r *reconcileContext) checkRestoreKeyspace() (bool, error) {
	val, ok := r.branch.Annotations[types.AnnoKeyBranchRestoreId]
	if !ok {
		return false, errors.New("failed to get annotation: " + types.AnnoKeyBranchRestoreId)
	}
	restoreId, err := strconv.ParseUint(val, 10, 64)
	if err != nil {
		return false, errors.New("invalid AnnoKeyBranchRestoreId annotation value")
	}
	clusterId, err := r.PdClient.GetClusterID(r.ctx)
	if err != nil {
		return false, err
	}

	r.Logger.V(1).Info("check restore",
		"restoreId", restoreId,
		"clusterId", clusterId,
		"keyspace", r.keyspaceToken,
	)
	task, err := r.TikvApi.GetRestoreProgress(r.ctx, restoreId, clusterId, r.keyspaceToken)
	if err != nil {
		if stderr.Is(err, tikv.ErrRestoreTaskNotExist) {
			r.Logger.Info(fmt.Sprintf("check restore task %d not exist", restoreId))
			// preserve restoreId and set AnnoKeyBranchErrorAndRecreate to true to make needRestoreKeyspace() pass
			r.branch.Annotations[types.AnnoKeyBranchErrorAndRecreate] = "true"
			return false, errors.Annotate(
				r.InfraClient.Update(r.ctx, r.branch),
				"failed to update AnnoKeyBranchErrorAndRecreate",
			)
		}
		return false, fmt.Errorf("check restore task %d failed: %w", restoreId, err)
	}

	if task.Status == "Succeed" {
		r.Info("[branch lifecycle] branch keyspace restore finished",
			"branch", r.branch.Name,
			"restore_id", restoreId,
			"restore_mode", r.BranchRestoreMode,
			"duration", task.Duration, // reported by tikv worker, unit unverified
			"restore_ts", task.RestoreTs, // actual snapshot TS
			"restore_bytes", task.RestoreBytes,
		)
		return true, nil
	}
	if task.Status == "Error" {
		r.Logger.Info(fmt.Sprintf("restore(%d) keyspace state is error: %s", restoreId, task.Error))
		// preserve restoreId and set AnnoKeyBranchErrorAndRecreate to true to make needRestoreKeyspace() pass
		r.branch.Annotations[types.AnnoKeyBranchErrorAndRecreate] = "true"
		return false, errors.Annotate(
			r.InfraClient.Update(r.ctx, r.branch),
			"failed to update AnnoKeyBranchErrorAndRecreate",
		)
	}
	r.Logger.Info(fmt.Sprintf(
		"restore(%d) keyspace state is %s, step: %s, progress: %d",
		restoreId,
		task.Status,
		task.Progress.Step,
		task.Progress.Progress,
	))
	return false, nil
}

func (r *reconcileContext) createRestoreTaskFromParent(
	parentKeyspace string,
	parentTimestamp time.Time,
) (uint64, error) {
	var restoreId uint64
	label, ok := r.branch.Annotations[types.AnnoKeyBranchRestoreId]
	if !ok {
		now := time.Now().UTC()
		// restoreId should be unique across all concurrent requests. Starting from MaxUint64 to be different
		// from what dataflow service uses.
		// 0x00 belongs to branch
		// 0xFFFFFFFFFFF has 17592186044415 slots for random choice
		restoreId = uint64(0x00FFF00000000000 | (now.UnixNano() & 0xFFFFFFFFFFF)) //nolint:mnd // mask
	} else {
		var err error
		restoreId, err = strconv.ParseUint(label, 10, 64)
		if err != nil {
			return 0, errors.New("invalid AnnoKeyBranchRestoreId annotation value")
		}
	}

	// tikv-api expects the physical tidb cluster id, not the logical serverless cluster id
	clusterId, err := r.PdClient.GetClusterID(r.ctx)
	if err != nil {
		return 0, err
	}
	r.Logger.V(1).Info(
		"create restore",
		"restoreId", restoreId,
		"clusterId", clusterId,
		"keyspace", r.keyspaceToken,
		"parentKeyspace", parentKeyspace,
		"point_in_time", parentTimestamp.String(),
		"restore_mode", r.BranchRestoreMode,
	)
	var createErr error
	if r.onlineSnapshotRestore() {
		createErr = r.TikvApi.CreateOnlineSnapshotRestore(r.ctx, restoreId, clusterId, r.keyspaceToken, parentKeyspace)
	} else {
		createErr = r.TikvApi.CreateRestore(r.ctx, restoreId, clusterId, r.keyspaceToken, parentKeyspace, parentTimestamp)
	}
	if createErr != nil {
		if stderr.Is(createErr, tikv.ErrRestoreConcurrencyLimit) {
			return 0, createErr
		}
		return 0, fmt.Errorf(
			"restore(%d) mode %s keyspace %s from %s failed: %w",
			restoreId, r.BranchRestoreMode, r.keyspaceToken, parentKeyspace, createErr,
		)
	}
	return restoreId, nil
}

// onlineSnapshotRestore reports whether this branch should be restored via
// online snapshot: only in online_snapshot mode and only when no explicit
// parent timestamp was requested. An explicit timestamp asks for a restore
// to that historical point, which an online snapshot cannot express, so it
// always goes through PiTR regardless of the mode.
func (r *reconcileContext) onlineSnapshotRestore() bool {
	if r.BranchRestoreMode != tikv.RestoreModeOnlineSnapshot {
		return false
	}
	_, hasTimestamp := r.branch.Annotations[ParentTimestampAnnKey]
	return !hasTimestamp
}

func (r *reconcileContext) getParentTimestamp() (parentTimestamp time.Time, err error) {
	parentTimestampStr, exist := r.branch.Annotations[ParentTimestampAnnKey]
	if exist {
		parentTimestamp, err = time.Parse(time.RFC3339, parentTimestampStr)
		if err != nil {
			return time.Time{}, errors.Wrap(err, "failed to parse parent timestamp")
		}
		now := time.Now().UTC()
		// sleep 1 seconds to prevent "future time error" from tikv-api
		if parentTimestamp.After(now.Add(-time.Second)) {
			time.Sleep(time.Second)
		}
		return parentTimestamp, nil
	}

	// No explicit timestamp (the branch creator omitted it): fall back to the
	// CR creation time as the PiTR point, ~= what upstream used to record as
	// "now" at API time. online_snapshot mode never reaches this branch.
	ts := r.branch.CreationTimestamp.Time.UTC()
	// sleep 1 second to prevent "future time error" from tikv-api, same as
	// the annotation path above
	if ts.After(time.Now().UTC().Add(-time.Second)) {
		time.Sleep(time.Second)
	}
	return ts, nil
}

// tikv-api DELETE
// 200 OK: 成功, "Restore task deleted"
// 209 CONFLICT: 任务非终态, "Restore task is not in final state"。可能是哪里有 bug，不应该发起 delete，需要人工介入
// 404 NOT FOUND: Restore task not found: <restore_id>。可以忽略，可能是 tikv-api 重启过，或者之前失败的请求实际上成功了
// 400 BAD REQUEST: 参数校验错误：可能是代码 bug，需要人工介入
// 500 INTERNAL SREVER ERROR，其他内部错误：可以重试，或者告警并人工介入
// func (r *reconcileContext) deleteRestoreKeyspace(clusterId, restoreId uint64) error {
// 	url := fmt.Sprintf(
// 		"http://%s/api/v1/restore_keyspace/%d?cluster_id=%d&keyspace=%s",
// 		tikvApiEndpoint, restoreId, clusterId, r.keyspaceToken,
// 	)
// 	r.Logger.V(1).Info("delete restore url: " + url)
// 	req, err := http.NewRequestWithContext(r.ctx, "DELETE", url, nil)
// 	if err != nil {
// 		return err
// 	}
// 	res, err := r.httpClient.Do(req)
// 	if err != nil {
// 		return err
// 	}
// 	if res.StatusCode == http.StatusOK || res.StatusCode == http.StatusNotFound {
// 		return nil
// 	}

// 	return fmt.Errorf("delete restore task %d failed: %s", restoreId, res.Status)
// }