package br

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"strings"

	v1alpha1br "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/clustergateway"
	dp "github.com/tidbcloud/cluster-service-ng/pkg/dataplane"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const FlushWALActionName = "flush-wal"
const StrictFlushWALActionName = "flush-wal-before-pause"

type flushWALAction struct {
	sharedPoolRepo    repo.SharedPoolRepo
	dataplaneRepo     repo.DataplaneRepo
	clustergatewayCli clustergateway.Client
	dataplaneManager  dp.Manager
	disableFlushWAL   bool
}

type flushWALPrerequisiteError struct {
	err      error
	kind     FlushWALPrerequisiteFailureKind
	terminal bool
}

func (e *flushWALPrerequisiteError) Error() string { return e.err.Error() }
func (e *flushWALPrerequisiteError) Unwrap() error { return e.err }

type FlushWALPrerequisiteFailureKind string

const (
	FlushWALPrerequisiteDependencyUnavailable FlushWALPrerequisiteFailureKind = "dependency-unavailable"
	FlushWALPrerequisiteDisabled              FlushWALPrerequisiteFailureKind = "flush-wal-disabled"
	FlushWALPrerequisiteSharedPoolNotFound    FlushWALPrerequisiteFailureKind = "shared-pool-not-found"
	FlushWALPrerequisiteDataplaneNotFound     FlushWALPrerequisiteFailureKind = "dataplane-not-found"
	FlushWALPrerequisiteGatewayURLMissing     FlushWALPrerequisiteFailureKind = "gateway-url-missing"
	FlushWALPrerequisiteTiBRNotFound          FlushWALPrerequisiteFailureKind = "tibr-not-found"
	FlushWALPrerequisiteTiBRNotReady          FlushWALPrerequisiteFailureKind = "tibr-not-ready"
)

func newFlushWALPrerequisiteError(kind FlushWALPrerequisiteFailureKind, terminal bool, format string, args ...any) error {
	return &flushWALPrerequisiteError{err: fmt.Errorf(format, args...), kind: kind, terminal: terminal}
}

func IsTerminalFlushWALPrerequisiteError(err error) bool {
	var prerequisiteErr *flushWALPrerequisiteError
	return errors.As(err, &prerequisiteErr) && prerequisiteErr.terminal
}

func FlushWALPrerequisiteFailureKindOf(err error) (FlushWALPrerequisiteFailureKind, bool) {
	var prerequisiteErr *flushWALPrerequisiteError
	if !errors.As(err, &prerequisiteErr) {
		return "", false
	}
	return prerequisiteErr.kind, true
}

// ValidateFlushWALPrerequisites is intentionally read-only. It proves that
// FlushWAL can be attempted, not that the endpoint is supported or will succeed.
// preloadedClient may be nil; supplying it avoids another Dataplane client health check.
func ValidateFlushWALPrerequisites(
	ctx context.Context,
	sharedPoolID string,
	sharedPoolRepo repo.SharedPoolRepo,
	dataplaneRepo repo.DataplaneRepo,
	dataplaneManager dp.Manager,
	preloadedClient client.Client,
	disableFlushWAL bool,
) (*entity.SharedPool, error) {
	if disableFlushWAL {
		return nil, newFlushWALPrerequisiteError(FlushWALPrerequisiteDisabled, true, "FlushWAL is disabled")
	}
	target, err := sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		kind := FlushWALPrerequisiteDependencyUnavailable
		if status.Code(err) == codes.NotFound {
			kind = FlushWALPrerequisiteSharedPoolNotFound
		}
		return nil, newFlushWALPrerequisiteError(kind, status.Code(err) == codes.NotFound, "load shared pool: %v", err)
	}
	provider, err := dataplaneRepo.LoadByName(ctx, target.GetDataplaneName())
	if err != nil {
		kind := FlushWALPrerequisiteDependencyUnavailable
		if status.Code(err) == codes.NotFound {
			kind = FlushWALPrerequisiteDataplaneNotFound
		}
		return nil, newFlushWALPrerequisiteError(kind, status.Code(err) == codes.NotFound, "load dataplane: %v", err)
	}
	if provider.GetGatewayUrl() == "" {
		return nil, newFlushWALPrerequisiteError(FlushWALPrerequisiteGatewayURLMissing, true, "dataplane gateway URL is empty")
	}
	target.Provider = provider
	target.GatewayBaseUrl = provider.GetGatewayUrl()
	if !strings.HasPrefix(target.GatewayBaseUrl, "http") {
		target.GatewayBaseUrl = "http://" + target.GatewayBaseUrl
	}
	if preloadedClient == nil {
		preloadedClient, err = dataplaneManager.GetClient(target.GetDataplaneName())
		if err != nil {
			return nil, newFlushWALPrerequisiteError(FlushWALPrerequisiteDependencyUnavailable, false, "get dataplane client: %v", err)
		}
	}
	tibr := &v1alpha1br.TiBR{}
	if err := preloadedClient.Get(ctx, types.NamespacedName{Name: tioperator.DefaultTiBRName, Namespace: target.GetDataPlaneNamespace()}, tibr); err != nil {
		kind := FlushWALPrerequisiteDependencyUnavailable
		if apierrors.IsNotFound(err) {
			kind = FlushWALPrerequisiteTiBRNotFound
		}
		return nil, newFlushWALPrerequisiteError(kind, apierrors.IsNotFound(err), "get default TiBR: %v", err)
	}
	if !tioperator.IsTiBRSyncedReady(tibr) {
		return nil, newFlushWALPrerequisiteError(FlushWALPrerequisiteTiBRNotReady, false, "default TiBR is not Ready and Synced")
	}
	return target, nil
}

type strictFlushWALAction struct {
	sharedPoolRepo    repo.SharedPoolRepo
	dataplaneRepo     repo.DataplaneRepo
	clustergatewayCli clustergateway.Client
	dataplaneManager  dp.Manager
	disableFlushWAL   bool
}

func NewStrictFlushWALAction(sharedPoolRepo repo.SharedPoolRepo, dataplaneRepo repo.DataplaneRepo, gateway clustergateway.Client, manager dp.Manager, disabled bool) actions.Action {
	return &strictFlushWALAction{sharedPoolRepo: sharedPoolRepo, dataplaneRepo: dataplaneRepo, clustergatewayCli: gateway, dataplaneManager: manager, disableFlushWAL: disabled}
}

func (a *strictFlushWALAction) Name() string { return StrictFlushWALActionName }

func (a *strictFlushWALAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	ctx, cancel := context.WithTimeout(ctx, actions.GuardedWriteTimeout)
	defer cancel()
	sharedPoolID, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		return actions.NewTerminalError(err)
	}
	sharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolID, false)
	if err != nil {
		return err
	}
	var remoteCli client.Client
	if sharedPool.PauseResumeState == entity.PauseResumeStatePausing {
		remoteCli, err = a.dataplaneManager.GetClient(sharedPool.GetDataplaneName())
		if err != nil {
			return err
		}
		operatorStatus, err := tioperator.InspectClusterSuspendCompute(ctx, remoteCli, sharedPoolID)
		if err != nil {
			return err
		}
		if operatorStatus.SpecDesired && !sharedPool.SuspendComputeDesired {
			return actions.NewTerminalError(fmt.Errorf("operator suspend spec is true while suspend_compute_desired is false"))
		}
		if operatorStatus.SpecDesired && operatorStatus.Observed {
			return nil
		}
	}
	target, err := ValidateFlushWALPrerequisites(ctx, sharedPoolID, a.sharedPoolRepo, a.dataplaneRepo, a.dataplaneManager, remoteCli, a.disableFlushWAL)
	if err != nil {
		if IsTerminalFlushWALPrerequisiteError(err) {
			return actions.NewTerminalError(err)
		}
		return err
	}
	if err := a.clustergatewayCli.FlushWAL(ctx, target, tioperator.DefaultTiBRName); err != nil {
		var httpErr *clustergateway.FlushWALHTTPError
		if errors.Is(err, clustergateway.ErrFlushWALUnsupported) ||
			(errors.As(err, &httpErr) && httpErr.StatusCode >= 400 && httpErr.StatusCode < 500 && httpErr.StatusCode != http.StatusRequestTimeout && httpErr.StatusCode != http.StatusTooManyRequests) {
			return actions.NewTerminalError(err)
		}
		return err
	}
	return nil
}

func NewFlushWALAction(
	sharedPoolRepo repo.SharedPoolRepo,
	dataplaneRepo repo.DataplaneRepo,
	clustergatewayCli clustergateway.Client,
	dataplaneManager dp.Manager,
	disableFlushWAL bool,
) actions.Action {
	return &flushWALAction{
		sharedPoolRepo:    sharedPoolRepo,
		dataplaneRepo:     dataplaneRepo,
		clustergatewayCli: clustergatewayCli,
		dataplaneManager:  dataplaneManager,
		disableFlushWAL:   disableFlushWAL,
	}
}

func (a *flushWALAction) Name() string {
	return FlushWALActionName
}

func (a *flushWALAction) Invoke(args actions.ActionArgs) error {
	ctx := utils.GenerateContext(args)
	tilog.Info(ctx, fmt.Sprintf("start %s action...", a.Name()))

	if a.disableFlushWAL {
		tilog.Info(ctx, "flush wal is disabled, skip flush wal action")
		return nil
	}

	sharedPoolId, err := args.GetString(workflow.SharedPoolIdKey)
	if err != nil {
		tilog.Error(ctx, "failed to get shared pool id", zap.Error(err))
		return fmt.Errorf("failed to get shared pool id: %w", err)
	}

	targetSharedPool, err := a.sharedPoolRepo.Load(ctx, sharedPoolId, true)
	if err != nil {
		return fmt.Errorf("failed to load sharedPool: %w", err)
	}

	dataplane, err := a.dataplaneRepo.LoadByName(ctx, targetSharedPool.GetDataplaneName())
	if err != nil {
		if status.Code(err) == codes.NotFound {
			tilog.Warn(ctx, "dataplane not found, skip flush wal", zap.String("dataplaneName", targetSharedPool.GetDataplaneName()), zap.Error(err))
			return nil
		}
		return fmt.Errorf("failed to load dataplane: %w", err)
	}

	targetSharedPool.Provider = dataplane
	if dataplane.GetGatewayUrl() == "" {
		tilog.Warn(ctx, "gateway url is empty, skip flush wal", zap.String("dataplaneName", targetSharedPool.GetDataplaneName()))
		return nil
	}
	targetSharedPool.GatewayBaseUrl = dataplane.GetGatewayUrl()
	if !strings.HasPrefix(targetSharedPool.GatewayBaseUrl, "http") {
		targetSharedPool.GatewayBaseUrl = fmt.Sprintf("http://%s", targetSharedPool.GatewayBaseUrl)
	}

	remoteCli, err := a.dataplaneManager.GetClient(targetSharedPool.GetDataplaneName())
	if err != nil {
		tilog.Error(ctx, "failed to get remote client for shared pool", zap.String("dataplaneName", targetSharedPool.GetDataplaneName()), zap.Error(err))
		return fmt.Errorf("failed to get remote client for shared pool: %w", err)
	}

	tibrCR := &v1alpha1br.TiBR{}
	tibrName := tioperator.DefaultTiBRName
	namespace := targetSharedPool.GetDataPlaneNamespace()
	err = remoteCli.Get(ctx, types.NamespacedName{
		Name:      tibrName,
		Namespace: namespace,
	}, tibrCR)
	if err != nil {
		if apierrors.IsNotFound(err) {
			tilog.Info(ctx, "tibr not found, skip flush wal", zap.String("name", tibrName), zap.String("namespace", namespace))
			return nil
		}
		tilog.Error(ctx, "failed to get tibr", zap.String("name", tibrName), zap.String("namespace", namespace), zap.Error(err))
		return fmt.Errorf("failed to get tibr: %w", err)
	}

	if !tioperator.IsTiBRSyncedReady(tibrCR) {
		tilog.Info(ctx, "tibr is not ready and synced, skip flush wal", zap.String("name", tibrName), zap.String("namespace", namespace))
		return nil
	}

	err = a.clustergatewayCli.FlushWAL(ctx, targetSharedPool, tioperator.DefaultTiBRName)
	if err != nil {
		FlushWALFailureTotal.With(prometheus.Labels{
			LabelSharedPoolId: sharedPoolId,
		}).Inc()
		tilog.Error(ctx, "failed to flush wal, continue dag", zap.Error(err), zap.String("shared_pool_id", sharedPoolId))
		return nil
	}

	tilog.Info(ctx, "successfully flushed wal", zap.String("shared_pool_id", sharedPoolId))
	return nil
}