package tikv

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
	"time"

	"github.com/go-logr/logr"
)

var _ Api = &api{}

const (
	getRestoreTaskPath              = "/api/v1/restore_keyspace/%d?cluster_id=%d&keyspace=%s"
	createRestorePath               = "/api/v1/restore_keyspace/%d?cluster_id=%d&keyspace=%s&source_keyspace=%s&point_in_time=%s"
	createOnlineSnapshotRestorePath = "/api/v1/restore_keyspace/%d?cluster_id=%d&keyspace=%s&source_keyspace=%s&restore_mode=%s"
	cancelLoadDataTasksPath         = "/load_data?cluster_id=%d&task_id_prefix=%s"
)

// Branch keyspace restore modes, see --branch-restore-mode on the manager.
const (
	RestoreModePitr           = "pitr"
	RestoreModeOnlineSnapshot = "online_snapshot"
)

type Api interface {
	GetRestoreProgress(
		ctx context.Context,
		restoreId uint64,
		clusterId uint64,
		keyspaceName string,
	) (*RestoreProgress, error)
	CreateRestore(ctx context.Context,
		restoreId uint64,
		clusterId uint64,
		keyspaceName string,
		parentKeyspaceName string,
		timepoint time.Time,
	) error
	CreateOnlineSnapshotRestore(ctx context.Context,
		restoreId uint64,
		clusterId uint64,
		keyspaceName string,
		parentKeyspaceName string,
	) error
	CancelLoadDataTasks(
		ctx context.Context,
		clusterId uint64,
		keyspaceId uint32,
		importTaskId int64,
	) error
}

// RestoreProgress is the progress of restore task, for more detail, see:
//
//nolint:lll // https://github.com/tidbcloud/cloud-storage-engine/blob/2ac6c1929d0cd3dcad99034616c37f28cdb5369d/cmd/tikv-worker/src/native_br.rs#L530
type RestoreProgress struct {
	Status       string   `json:"status"` // Pending, Init, Running, Succeed, Error
	Error        string   `json:"error"`
	Duration     int64    `json:"duration"`
	Id           uint64   `json:"id"`
	Keyspace     string   `json:"keyspace"`
	RestoreType  string   `json:"restore_type"`
	RestoreTs    uint64   `json:"restore_ts"` // actual snapshot TS, reported on success
	RestoreBytes uint64   `json:"restore_bytes"`
	Progress     Progress `json:"progress"`
}

type Progress struct {
	Step     string `json:"step"`
	Progress int32  `json:"progress"`
}

type api struct {
	client      *http.Client
	serviceAddr string
}

func NewApi(serviceAddr string, tlsConfig *tls.Config) Api {
	var (
		scheme = "http://"
		client = http.DefaultClient
	)
	if tlsConfig != nil {
		scheme = "https://"
		client.Transport = &http.Transport{
			TLSClientConfig:   tlsConfig,
			ForceAttemptHTTP2: true,
		}
	}

	return &api{
		client:      client,
		serviceAddr: scheme + serviceAddr,
	}
}

func (a *api) newRequest(ctx context.Context, method string, path string) (*http.Request, error) {
	req, err := http.NewRequestWithContext(ctx, method, a.serviceAddr+path, nil)
	if err != nil {
		return nil, fmt.Errorf("create http request error: %w", err)
	}

	logger, err := logr.FromContext(ctx)
	if err == nil {
		logger.Info("create http request", "url", req.URL.String())
	}
	return req, nil
}

func (a *api) do(req *http.Request, statusChecker func(statusCode int, body []byte) error) ([]byte, error) {
	resp, err := a.client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("http request error: %w", err)
	}
	defer resp.Body.Close() //nolint:errcheck // ignore error
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("read body error: %w", err)
	}

	logger, err := logr.FromContext(req.Context())
	if err == nil {
		logger.Info("get http response", "url", req.URL.String(), "code", resp.StatusCode, "body", string(body))
	}
	err = statusChecker(resp.StatusCode, body)
	if err != nil {
		return nil, err
	}
	return body, nil
}

var (
	ErrRestoreTaskNotExist     = fmt.Errorf("restore task not exist")
	ErrRestoreConcurrencyLimit = fmt.Errorf("restore reach concurrency limit")
)

func createRestoreStatusChecker(statusCode int, body []byte) error {
	if statusCode == http.StatusInternalServerError && strings.Contains(string(body), "reach concurrency limit") {
		return ErrRestoreConcurrencyLimit
	}
	if statusCode >= http.StatusBadRequest {
		return fmt.Errorf("create restore task error, status: %d, body: %s", statusCode, body)
	}
	return nil
}

func (a *api) GetRestoreProgress(
	ctx context.Context,
	restoreId uint64,
	clusterId uint64,
	keyspaceName string,
) (*RestoreProgress, error) {
	req, err := a.newRequest(
		ctx,
		http.MethodGet,
		fmt.Sprintf(getRestoreTaskPath, restoreId, clusterId, keyspaceName),
	)
	if err != nil {
		return nil, err
	}

	resp, err := a.do(req, func(statusCode int, body []byte) error {
		if statusCode == http.StatusNotFound {
			return ErrRestoreTaskNotExist
		}
		if statusCode >= http.StatusBadRequest {
			return fmt.Errorf("get restore task status error, status: %d, body: %s", statusCode, body)
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	var progress RestoreProgress
	err = json.Unmarshal(resp, &progress)
	if err != nil {
		return nil, fmt.Errorf("json unmarshal error: %w", err)
	}
	return &progress, nil
}

func (a *api) CreateRestore(ctx context.Context,
	restoreId uint64,
	clusterId uint64,
	keyspaceName string,
	parentKeyspaceName string,
	timepoint time.Time,
) error {
	t := timepoint.Format("2006-01-02 15:04:05.000")
	escapedTime := strings.ReplaceAll(url.QueryEscape(t), "+", "%20")
	req, err := a.newRequest(ctx,
		http.MethodPut,
		fmt.Sprintf(
			createRestorePath,
			restoreId,
			clusterId,
			keyspaceName,
			parentKeyspaceName,
			escapedTime,
		),
	)
	if err != nil {
		return err
	}

	_, err = a.do(req, createRestoreStatusChecker)
	return err
}

// CreateOnlineSnapshotRestore creates an online snapshot restore task: the
// snapshot TS is chosen by the tikv worker from PD at task start, so no
// point_in_time (nor backup_id/backup_name) may be sent.
func (a *api) CreateOnlineSnapshotRestore(ctx context.Context,
	restoreId uint64,
	clusterId uint64,
	keyspaceName string,
	parentKeyspaceName string,
) error {
	req, err := a.newRequest(ctx,
		http.MethodPut,
		fmt.Sprintf(
			createOnlineSnapshotRestorePath,
			restoreId,
			clusterId,
			keyspaceName,
			parentKeyspaceName,
			RestoreModeOnlineSnapshot,
		),
	)
	if err != nil {
		return err
	}

	_, err = a.do(req, createRestoreStatusChecker)
	return err
}

func (a *api) CancelLoadDataTasks(ctx context.Context, clusterId uint64, keyspaceId uint32, importTaskId int64) error {
	req, err := a.newRequest(ctx,
		http.MethodDelete,
		fmt.Sprintf(
			cancelLoadDataTasksPath,
			clusterId,
			fmt.Sprintf("%d-%d", keyspaceId,
				importTaskId),
		),
	)
	if err != nil {
		return err
	}

	_, err = a.do(req, func(statusCode int, body []byte) error {
		if statusCode >= http.StatusBadRequest {
			return fmt.Errorf("cancel loadData tasks error, status: %d, body: %s", statusCode, body)
		}
		return nil
	})
	return err
}