package controllercmd

import (
	"context"
	"fmt"
	"k8s.io/utils/clock"
	"os"
	"strings"
	"sync"
	"time"

	configv1 "github.com/openshift/api/config/v1"
	operatorv1alpha1 "github.com/openshift/api/operator/v1alpha1"
	"github.com/openshift/library-go/pkg/authorization/hardcodedauthorizer"
	"github.com/openshift/library-go/pkg/config/client"
	"github.com/openshift/library-go/pkg/config/clusterstatus"
	"github.com/openshift/library-go/pkg/config/configdefaults"
	leaderelectionconverter "github.com/openshift/library-go/pkg/config/leaderelection"
	"github.com/openshift/library-go/pkg/config/serving"
	"github.com/openshift/library-go/pkg/controller/fileobserver"
	"github.com/openshift/library-go/pkg/operator/events"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/version"
	"k8s.io/apiserver/pkg/authorization/union"
	genericapiserver "k8s.io/apiserver/pkg/server"
	"k8s.io/apiserver/pkg/server/healthz"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/leaderelection"
	"k8s.io/client-go/tools/record"
	"k8s.io/component-base/metrics"
	"k8s.io/component-base/metrics/legacyregistry"
	"k8s.io/klog/v2"
)

// StartFunc is the function to call on leader election start
type StartFunc func(context.Context, *ControllerContext) error

type ControllerContext struct {
	ComponentConfig *unstructured.Unstructured

	// Clock is a potentially fake clock that must be used to run controllers.
	Clock clock.Clock

	// KubeConfig provides the REST config with no content type (it will default to JSON).
	// Use this config for CR resources.
	KubeConfig *rest.Config

	// ProtoKubeConfig provides the REST config with "application/vnd.kubernetes.protobuf,application/json" content type.
	// Note that this config might not be safe for CR resources, instead it should be used for other resources.
	ProtoKubeConfig *rest.Config

	// EventRecorder is used to record events in controllers.
	EventRecorder events.Recorder

	// Server is the GenericAPIServer serving healthz checks and debug info
	Server *genericapiserver.GenericAPIServer

	// Namespace where the operator runs. Either specified on the command line or autodetected.
	OperatorNamespace string
}

// defaultObserverInterval specifies the default interval that file observer will do rehash the files it watches and react to any changes
// in those files.
var defaultObserverInterval = 5 * time.Second

// ControllerBuilder allows the construction of an controller in optional pieces.
type ControllerBuilder struct {
	kubeAPIServerConfigFile *string
	clientOverrides         *client.ClientConnectionOverrides
	leaderElection          *configv1.LeaderElection
	fileObserver            fileobserver.Observer
	fileObserverReactorFn   func(file string, action fileobserver.ActionType) error
	eventRecorderOptions    record.CorrelatorOptions
	componentOwnerReference *corev1.ObjectReference
	clock                   clock.Clock

	startFunc          StartFunc
	componentName      string
	componentNamespace string
	instanceIdentity   string
	observerInterval   time.Duration

	servingInfo          *configv1.HTTPServingInfo
	authenticationConfig *operatorv1alpha1.DelegatedAuthentication
	authorizationConfig  *operatorv1alpha1.DelegatedAuthorization
	healthChecks         []healthz.HealthChecker

	versionInfo *version.Info

	// nonZeroExitFn takes a function that exit the process with non-zero code.
	// This stub exists for unit test where we can check if the graceful termination work properly.
	// Default function will klog.Warning(args) and os.Exit(1).
	nonZeroExitFn func(args ...interface{})

	// Keep track if we defaulted leader election, used to make sure we don't stomp on the users intent for leader election
	// We use this flag to determine at runtime if we can alter leader election for SNO configurations
	userExplicitlySetLeaderElectionValues bool

	// different deployment strategies will require sensing topologies in disjoint manners
	topologyDetector TopologyDetector

	// Allow enabling HTTP2
	enableHTTP2 bool
}

type TopologyDetector interface {
	DetectTopology(ctx context.Context, restClient *rest.Config) (configv1.TopologyMode, error)
}

type infrastructureStatusTopologyDetector struct{}

func (i infrastructureStatusTopologyDetector) DetectTopology(ctx context.Context, restClient *rest.Config) (configv1.TopologyMode, error) {
	infraStatus, err := clusterstatus.GetClusterInfraStatus(ctx, restClient)
	if err != nil {
		return "", err
	}
	if infraStatus == nil {
		return "", nil
	}
	return infraStatus.ControlPlaneTopology, nil
}

var _ TopologyDetector = (*infrastructureStatusTopologyDetector)(nil)

// NewController returns a builder struct for constructing the command you want to run
func NewController(componentName string, startFunc StartFunc, clock clock.Clock) *ControllerBuilder {
	return &ControllerBuilder{
		startFunc:        startFunc,
		componentName:    componentName,
		clock:            clock,
		observerInterval: defaultObserverInterval,
		nonZeroExitFn: func(args ...interface{}) {
			klog.Warning(args...)
			os.Exit(1)
		},
		topologyDetector: infrastructureStatusTopologyDetector{},
	}
}

// WithRestartOnChange will enable a file observer controller loop that observes changes into specified files. If a change to a file is detected,
// the specified channel will be closed (allowing to graceful shutdown for other channels).
func (b *ControllerBuilder) WithRestartOnChange(stopCh chan<- struct{}, startingFileContent map[string][]byte, files ...string) *ControllerBuilder {
	if len(files) == 0 {
		return b
	}
	if b.fileObserver == nil {
		observer, err := fileobserver.NewObserver(b.observerInterval)
		if err != nil {
			panic(err)
		}
		b.fileObserver = observer
	}
	var once sync.Once

	b.fileObserverReactorFn = func(filename string, action fileobserver.ActionType) error {
		once.Do(func() {
			klog.Warning(fmt.Sprintf("Restart triggered because of %s", action.String(filename)))
			close(stopCh)
		})
		return nil
	}

	b.fileObserver.AddReactor(b.fileObserverReactorFn, startingFileContent, files...)
	return b
}

func (b *ControllerBuilder) WithComponentNamespace(ns string) *ControllerBuilder {
	b.componentNamespace = ns
	return b
}

// WithLeaderElection adds leader election options
func (b *ControllerBuilder) WithLeaderElection(leaderElection configv1.LeaderElection, defaultNamespace, defaultName string) *ControllerBuilder {
	if leaderElection.Disable {
		return b
	}

	// Set flag that SNO leader election configs can be used since user provided no timing configs
	b.userExplicitlySetLeaderElectionValues = leaderElection.LeaseDuration.Duration != 0 ||
		leaderElection.RenewDeadline.Duration != 0 ||
		leaderElection.RetryPeriod.Duration != 0

	defaulted := leaderelectionconverter.LeaderElectionDefaulting(leaderElection, defaultNamespace, defaultName)
	b.leaderElection = &defaulted
	return b
}

func (b *ControllerBuilder) WithTopologyDetector(topologyDetector TopologyDetector) *ControllerBuilder {
	b.topologyDetector = topologyDetector
	return b
}

// WithVersion accepts a getting that provide binary version information that is used to report build_info information to prometheus
func (b *ControllerBuilder) WithVersion(info version.Info) *ControllerBuilder {
	b.versionInfo = &info
	return b
}

// WithServer adds a server that provides metrics and healthz
func (b *ControllerBuilder) WithServer(servingInfo configv1.HTTPServingInfo, authenticationConfig operatorv1alpha1.DelegatedAuthentication, authorizationConfig operatorv1alpha1.DelegatedAuthorization) *ControllerBuilder {
	b.servingInfo = servingInfo.DeepCopy()
	configdefaults.SetRecommendedHTTPServingInfoDefaults(b.servingInfo)
	b.authenticationConfig = &authenticationConfig
	b.authorizationConfig = &authorizationConfig
	return b
}

// WithHTTP2 indicates that http2 should be enabled
func (b *ControllerBuilder) WithHTTP2() *ControllerBuilder {
	b.enableHTTP2 = true
	return b
}

// WithHealthChecks adds a list of healthchecks to the server
func (b *ControllerBuilder) WithHealthChecks(healthChecks ...healthz.HealthChecker) *ControllerBuilder {
	b.healthChecks = append(b.healthChecks, healthChecks...)
	return b
}

// WithKubeConfigFile sets an optional kubeconfig file. inclusterconfig will be used if filename is empty
func (b *ControllerBuilder) WithKubeConfigFile(kubeConfigFilename string, defaults *client.ClientConnectionOverrides) *ControllerBuilder {
	b.kubeAPIServerConfigFile = &kubeConfigFilename
	b.clientOverrides = defaults
	return b
}

// WithInstanceIdentity sets the instance identity to use if you need something special. The default is just a UID which is
// usually fine for a pod.
func (b *ControllerBuilder) WithInstanceIdentity(identity string) *ControllerBuilder {
	b.instanceIdentity = identity
	return b
}

// WithEventRecorderOptions allows to override the default Kubernetes event recorder correlator options.
// This is needed if the binary is sending a lot of events.
// Using events.DefaultOperatorEventRecorderOptions here makes a good default for normal operator binary.
func (b *ControllerBuilder) WithEventRecorderOptions(options record.CorrelatorOptions) *ControllerBuilder {
	b.eventRecorderOptions = options
	return b
}

// WithComponentOwnerReference overrides controller reference resolution for event recording
func (b *ControllerBuilder) WithComponentOwnerReference(reference *corev1.ObjectReference) *ControllerBuilder {
	b.componentOwnerReference = reference
	return b
}

// Run starts your controller for you.  It uses leader election if you asked, otherwise it directly calls you
func (b *ControllerBuilder) Run(ctx context.Context, config *unstructured.Unstructured) error {
	clientConfig, err := b.getClientConfig()
	if err != nil {
		return err
	}

	if b.fileObserver != nil {
		go b.fileObserver.Run(ctx.Done())
	}

	kubeClient := kubernetes.NewForConfigOrDie(clientConfig)
	namespace, err := b.getComponentNamespace()
	if err != nil {
		klog.Warningf("unable to identify the current namespace for events: %v", err)
	}
	controllerRef := b.componentOwnerReference

	if controllerRef == nil {
		controllerRef, err = events.GetControllerReferenceForCurrentPod(ctx, kubeClient, namespace, nil)
		if err != nil {
			klog.Warningf("unable to get owner reference (falling back to namespace): %v", err)
		}
	}
	eventRecorder := events.NewKubeRecorderWithOptions(kubeClient.CoreV1().Events(namespace), b.eventRecorderOptions, b.componentName, controllerRef, b.clock)

	utilruntime.PanicHandlers = append(utilruntime.PanicHandlers, func(c context.Context, r interface{}) {
		eventRecorder.Warningf(fmt.Sprintf("%sPanic", strings.Title(b.componentName)), "Panic observed: %v", r)
	})

	// if there is file observer defined for this command, add event into default reaction function.
	if b.fileObserverReactorFn != nil {
		originalFileObserverReactorFn := b.fileObserverReactorFn
		b.fileObserverReactorFn = func(file string, action fileobserver.ActionType) error {
			eventRecorder.Warningf("OperatorRestart", "Restarted because of %s", action.String(file))
			return originalFileObserverReactorFn(file, action)
		}
	}

	// report the binary version metrics to prometheus
	if b.versionInfo != nil {
		buildInfo := metrics.NewGaugeVec(
			&metrics.GaugeOpts{
				Name: strings.Replace(namespace, "-", "_", -1) + "_build_info",
				Help: "A metric with a constant '1' value labeled by major, minor, git version, git commit, git tree state, build date, Go version, " +
					"and compiler from which " + b.componentName + " was built, and platform on which it is running.",
				StabilityLevel: metrics.ALPHA,
			},
			[]string{"major", "minor", "gitVersion", "gitCommit", "gitTreeState", "buildDate", "goVersion", "compiler", "platform"},
		)
		legacyregistry.MustRegister(buildInfo)
		buildInfo.WithLabelValues(b.versionInfo.Major, b.versionInfo.Minor, b.versionInfo.GitVersion, b.versionInfo.GitCommit, b.versionInfo.GitTreeState, b.versionInfo.BuildDate, b.versionInfo.GoVersion,
			b.versionInfo.Compiler, b.versionInfo.Platform).Set(1)
		klog.Infof("%s version %s-%s", b.componentName, b.versionInfo.GitVersion, b.versionInfo.GitCommit)
	}

	kubeConfig := ""
	if b.kubeAPIServerConfigFile != nil {
		kubeConfig = *b.kubeAPIServerConfigFile
	}

	var server *genericapiserver.GenericAPIServer
	if b.servingInfo != nil {
		serverConfig, err := serving.ToServerConfig(ctx, *b.servingInfo, *b.authenticationConfig, *b.authorizationConfig, kubeConfig, kubeClient, b.leaderElection, b.enableHTTP2, b.versionInfo)
		if err != nil {
			return err
		}
		serverConfig.Authorization.Authorizer = union.New(
			// prefix the authorizer with the permissions for metrics scraping which are well known.
			// openshift RBAC policy will always allow this user to read metrics.
			hardcodedauthorizer.NewHardCodedMetricsAuthorizer(),
			serverConfig.Authorization.Authorizer,
		)
		serverConfig.HealthzChecks = append(serverConfig.HealthzChecks, b.healthChecks...)

		server, err = serverConfig.Complete(nil).New(b.componentName, genericapiserver.NewEmptyDelegate())
		if err != nil {
			return err
		}

		go func() {
			if err := server.PrepareRun().Run(ctx.Done()); err != nil {
				klog.Fatal(err)
			}
			klog.Info("server exited")
		}()
	}

	protoConfig := rest.CopyConfig(clientConfig)
	protoConfig.AcceptContentTypes = "application/vnd.kubernetes.protobuf,application/json"
	protoConfig.ContentType = "application/vnd.kubernetes.protobuf"

	controllerContext := &ControllerContext{
		ComponentConfig:   config,
		Clock:             b.clock,
		KubeConfig:        clientConfig,
		ProtoKubeConfig:   protoConfig,
		EventRecorder:     eventRecorder,
		Server:            server,
		OperatorNamespace: namespace,
	}

	if b.leaderElection == nil {
		if err := b.startFunc(ctx, controllerContext); err != nil {
			return err
		}
		return nil
	}

	if !b.userExplicitlySetLeaderElectionValues {
		topology, err := b.topologyDetector.DetectTopology(ctx, clientConfig)
		if err != nil || topology == "" {
			eventRecorder.Warningf("ControlPlaneTopology", "unable to get control plane topology, using HA cluster values for leader election: %v", err)
			klog.Warningf("unable to get control plane topology, using HA cluster values for leader election: %v", err)
		} else {
			snoLeaderElection := topologyLeaderElection(topology, *b.leaderElection)
			b.leaderElection = &snoLeaderElection
		}
	}

	// ensure blocking TCP connections don't block the leader election
	leaderConfig := rest.CopyConfig(protoConfig)
	leaderConfig.Timeout = b.leaderElection.RenewDeadline.Duration

	leaderElection, err := leaderelectionconverter.ToLeaderElectionWithLease(leaderConfig, *b.leaderElection, b.componentName, b.instanceIdentity)
	if err != nil {
		return err
	}

	// 10s is the graceful termination time we give the controllers to finish their workers.
	// when this time pass, we exit with non-zero code, killing all controller workers.
	// NOTE: The pod must set the termination graceful time.
	leaderElection.Callbacks.OnStartedLeading = b.getOnStartedLeadingFunc(controllerContext, 10*time.Second)

	leaderelection.RunOrDie(ctx, leaderElection)
	return nil
}

func (b ControllerBuilder) getOnStartedLeadingFunc(controllerContext *ControllerContext, gracefulTerminationDuration time.Duration) func(ctx context.Context) {
	return func(ctx context.Context) {
		stoppedCh := make(chan struct{})
		go func() {
			defer close(stoppedCh)
			if err := b.startFunc(ctx, controllerContext); err != nil {
				b.nonZeroExitFn(fmt.Sprintf("graceful termination failed, controllers failed with error: %v", err))
			}
		}()

		select {
		case <-ctx.Done(): // context closed means the process likely received signal to terminate
			controllerContext.EventRecorder.Shutdown()
		case <-stoppedCh:
			// if context was not cancelled (it is not "done"), but the startFunc terminated, it means it terminated prematurely
			// when this happen, it means the controllers terminated without error.
			if ctx.Err() == nil {
				b.nonZeroExitFn("graceful termination failed, controllers terminated prematurely")
			}
		}

		select {
		case <-time.After(gracefulTerminationDuration): // when context was closed above, give controllers extra time to terminate gracefully
			b.nonZeroExitFn(fmt.Sprintf("graceful termination failed, some controllers failed to shutdown in %s", gracefulTerminationDuration))
		case <-stoppedCh: // stoppedCh here means the controllers finished termination and we exit 0
		}
	}
}

func (b *ControllerBuilder) getComponentNamespace() (string, error) {
	if len(b.componentNamespace) > 0 {
		return b.componentNamespace, nil
	}
	nsBytes, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace")
	if err != nil {
		return "openshift-config-managed", err
	}
	return string(nsBytes), nil
}

func (b *ControllerBuilder) getClientConfig() (*rest.Config, error) {
	kubeconfig := ""
	if b.kubeAPIServerConfigFile != nil {
		kubeconfig = *b.kubeAPIServerConfigFile
	}

	return client.GetKubeConfigOrInClusterConfig(kubeconfig, b.clientOverrides)
}

func topologyLeaderElection(topology configv1.TopologyMode, original configv1.LeaderElection) configv1.LeaderElection {
	// if we can't determine the infra topology, return original
	if topology == "" {
		return original
	}

	// If we are running in a SingleReplicaTopologyMode and leader election is not disabled, configure leader election for SNO Topology
	if topology == configv1.SingleReplicaTopologyMode && !original.Disable {
		klog.Info("detected SingleReplicaTopologyMode, the original leader election has been altered for the default SingleReplicaTopology")
		return leaderelectionconverter.LeaderElectionSNOConfig(original)
	}
	return original
}