package k8s

import (
	"errors"
	"sync"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/rest"

	"github.com/grafana/grafana-app-sdk/metrics"
	"github.com/grafana/grafana-app-sdk/resource"
)

var _ resource.ClientGenerator = &ClientRegistry{}

// NewClientRegistry returns a new ClientRegistry which will make Client structs using the provided rest.Config
func NewClientRegistry(kubeCconfig rest.Config, clientConfig ClientConfig) *ClientRegistry {
	kubeCconfig.NegotiatedSerializer = &GenericNegotiatedSerializer{}
	kubeCconfig.UserAgent = rest.DefaultKubernetesUserAgent()

	// Apply stream error handling if enabled
	if clientConfig.EnableStreamErrorHandling {
		WrapWithStreamErrorHandling(&kubeCconfig)
	}

	return &ClientRegistry{
		clients:      make(map[schema.GroupVersionKind]rest.Interface),
		crClients:    make(map[schema.GroupVersion]rest.Interface),
		cfg:          kubeCconfig,
		clientConfig: clientConfig,
		requestDurations: prometheus.NewHistogramVec(prometheus.HistogramOpts{
			Namespace:                       clientConfig.MetricsConfig.Namespace,
			Subsystem:                       "kubernetes_client",
			Name:                            "request_duration_seconds",
			Help:                            "Time (in seconds) spent serving HTTP requests.",
			Buckets:                         metrics.LatencyBuckets,
			NativeHistogramBucketFactor:     clientConfig.MetricsConfig.NativeHistogramBucketFactor,
			NativeHistogramMaxBucketNumber:  clientConfig.MetricsConfig.NativeHistogramMaxBucketNumber,
			NativeHistogramMinResetDuration: time.Hour,
		}, []string{"status_code", "verb", "kind", "subresource"}),
		totalRequests: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:      "requests_total",
			Subsystem: "kubernetes_client",
			Namespace: clientConfig.MetricsConfig.Namespace,
			Help:      "Total number of kubernetes requests",
		}, []string{"status_code", "verb", "kind", "subresource"}),
		watchEventsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:      "watch_events_total",
			Subsystem: "kubernetes_client",
			Namespace: clientConfig.MetricsConfig.Namespace,
			Help:      "Total number of watch events received by type",
		}, []string{"event_type", "kind"}),
		watchErrorsTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:      "watch_errors_total",
			Subsystem: "kubernetes_client",
			Namespace: clientConfig.MetricsConfig.Namespace,
			Help:      "Total number of watch event parsing/translation errors",
		}, []string{"error_type", "kind"}),
	}
}

// ClientRegistry implements resource.ClientGenerator, and keeps a cache of kubernetes clients based on
// GroupVersion (the largest unit a kubernetes rest.RESTClient can work with).
type ClientRegistry struct {
	clients          map[schema.GroupVersionKind]rest.Interface
	crClients        map[schema.GroupVersion]rest.Interface
	discoveryClient  *DiscoveryClient
	cfg              rest.Config
	clientConfig     ClientConfig
	mutex            sync.Mutex
	requestDurations *prometheus.HistogramVec
	totalRequests    *prometheus.CounterVec
	watchEventsTotal *prometheus.CounterVec
	watchErrorsTotal *prometheus.CounterVec
}

// ClientFor returns a Client with the underlying rest.Interface being a cached one for the Schema's GroupVersion.
// If no such client is cached, it creates a new one with the stored config.
func (c *ClientRegistry) ClientFor(sch resource.Kind) (resource.Client, error) {
	codec := sch.Codec(resource.KindEncodingJSON)
	if codec == nil {
		return nil, errors.New("no codec for KindEncodingJSON")
	}
	client, err := c.getClientFor(sch)
	if err != nil {
		return nil, err
	}
	return &Client{
		client: &groupVersionClient{
			client:           client,
			version:          sch.Version(),
			config:           c.clientConfig,
			requestDurations: c.requestDurations,
			totalRequests:    c.totalRequests,
			watchEventsTotal: c.watchEventsTotal,
			watchErrorsTotal: c.watchErrorsTotal,
		},
		schema: sch,
		codec:  codec,
		config: c.clientConfig,
	}, nil
}

// GetCustomRouteClient returns a Client with the underlying rest.Interface being a cached one for the provided GroupVersion.
// If no such client is cached, it creates a new one with the stored config. This method is used for generating
// clients that are not tied to a specific schema.
func (c *ClientRegistry) GetCustomRouteClient(gv schema.GroupVersion, defaultNamespace string) (resource.CustomRouteClient, error) {
	client, err := c.getCustomRouteClient(gv)
	if err != nil {
		return nil, err
	}
	return &CustomRouteClient{
		defaultNamespace: defaultNamespace,
		groupVersionClient: &groupVersionClient{
			client:           client,
			version:          gv.Version,
			config:           c.clientConfig,
			requestDurations: c.requestDurations,
			totalRequests:    c.totalRequests,
			watchEventsTotal: c.watchEventsTotal,
			watchErrorsTotal: c.watchErrorsTotal,
		},
	}, nil
}

// PrometheusCollectors returns the prometheus metric collectors used by all clients generated by this ClientRegistry to allow for registration
func (c *ClientRegistry) PrometheusCollectors() []prometheus.Collector {
	return []prometheus.Collector{
		c.totalRequests, c.requestDurations, c.watchEventsTotal, c.watchErrorsTotal,
	}
}

func (c *ClientRegistry) getCustomRouteClient(gv schema.GroupVersion) (rest.Interface, error) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	if c, ok := c.crClients[gv]; ok {
		return c, nil
	}

	ccfg := c.cfg
	ccfg.GroupVersion = &gv
	restClient, err := rest.RESTClientFor(&ccfg)
	if err != nil {
		return nil, err
	}
	c.crClients[gv] = restClient
	return restClient, nil
}

// DiscoveryClient returns a DiscoveryClient configured with the registry's kubeconfig and
// KubeConfigProvider, so per-group routing (if any) is preserved. The DiscoveryClient is
// cached on first call and reused, so its internal per-group client cache is shared across
// callers.
func (c *ClientRegistry) DiscoveryClient() (resource.DiscoveryClient, error) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	if c.discoveryClient == nil {
		c.discoveryClient = NewDiscoveryClient(c.cfg, c.clientConfig.KubeConfigProvider)
	}
	return c.discoveryClient, nil
}

func (c *ClientRegistry) getClientFor(sch resource.Kind) (rest.Interface, error) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	gvk := schema.GroupVersionKind{
		Group:   sch.Group(),
		Version: sch.Version(),
		Kind:    sch.Kind(),
	}

	if c, ok := c.clients[gvk]; ok {
		return c, nil
	}

	ccfg := c.cfg
	ccfg.GroupVersion = &schema.GroupVersion{
		Group:   gvk.Group,
		Version: gvk.Version,
	}
	if c.clientConfig.NegotiatedSerializerProvider != nil {
		ccfg.NegotiatedSerializer = c.clientConfig.NegotiatedSerializerProvider(sch)
	} else {
		ccfg.NegotiatedSerializer = &GenericNegotiatedSerializer{}
	}
	if c.clientConfig.KubeConfigProvider != nil {
		ccfg = c.clientConfig.KubeConfigProvider(sch, ccfg)
	}
	client, err := rest.RESTClientFor(&ccfg)
	if err != nil {
		return nil, err
	}
	c.clients[gvk] = client
	return client, nil
}
