/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package kube

import (
	"fmt"
	"sync"
	"time"

	"github.com/apache/dubbo-kubernetes/pkg/admin/cache"
	"github.com/apache/dubbo-kubernetes/pkg/admin/cache/selector"
	"github.com/apache/dubbo-kubernetes/pkg/admin/constant"
	"github.com/apache/dubbo-kubernetes/pkg/core/kubeclient/client"
	"github.com/apache/dubbo-kubernetes/pkg/core/logger"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/informers"
	appsv1Listers "k8s.io/client-go/listers/apps/v1"
	corev1Listers "k8s.io/client-go/listers/core/v1"
	kubeToolsCache "k8s.io/client-go/tools/cache"
)

var KubernetesCacheInstance *KubernetesCache

func NewKubernetesCache(kc *client.KubeClient, clusterScoped bool) *KubernetesCache {
	return &KubernetesCache{
		client:               kc,
		clusterScoped:        clusterScoped,
		namespaceCacheLister: make(map[string]*cacheLister),
		namespaceStopChan:    make(map[string]chan struct{}),
	}
}

type KubernetesCache struct {
	lock                 sync.RWMutex
	client               *client.KubeClient
	clusterScoped        bool
	refreshDuration      time.Duration
	clusterCacheLister   *cacheLister
	clusterStopChan      chan struct{}
	namespaceCacheLister map[string]*cacheLister
	namespaceStopChan    map[string]chan struct{}
}

type cacheLister struct {
	configMapLister   corev1Listers.ConfigMapLister
	daemonSetLister   appsv1Listers.DaemonSetLister
	deploymentLister  appsv1Listers.DeploymentLister
	endpointLister    corev1Listers.EndpointsLister
	podLister         corev1Listers.PodLister
	replicaSetLister  appsv1Listers.ReplicaSetLister
	serviceLister     corev1Listers.ServiceLister
	statefulSetLister appsv1Listers.StatefulSetLister

	cachesSynced []kubeToolsCache.InformerSynced
}

func (c *KubernetesCache) GetApplications(namespace string) ([]*cache.ApplicationModel, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	applicationSet := make(map[string]struct{})
	deployments, err := c.getCacheLister(namespace).deploymentLister.List(labels.Everything())
	if err != nil {
		return nil, err
	}

	res := make([]*cache.ApplicationModel, 0)
	for _, deployment := range deployments {
		if _, ok := deployment.Labels[constant.ApplicationLabel]; ok {
			if _, exist := applicationSet[constant.ApplicationLabel]; !exist {
				applicationSet[deployment.Labels[constant.ApplicationLabel]] = struct{}{} // mark as exist
				res = append(res, &cache.ApplicationModel{
					Name: deployment.Name,
				})
			}
		}
	}

	return res, nil
}

func (c *KubernetesCache) GetWorkloads(namespace string) ([]*cache.WorkloadModel, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	res := make([]*cache.WorkloadModel, 0)

	deployments, err := c.getCacheLister(namespace).deploymentLister.List(labels.Everything())
	if err != nil {
		return nil, err
	}
	for _, deployment := range deployments {
		if _, ok := deployment.Labels[constant.ApplicationLabel]; ok {
			res = append(res, &cache.WorkloadModel{
				Application: &cache.ApplicationModel{
					Name: deployment.Labels[constant.ApplicationLabel],
				},
				Name:   deployment.Name,
				Type:   constant.DeploymentType,
				Image:  deployment.Spec.Template.Spec.Containers[0].Image,
				Labels: deployment.Labels,
			})
		}
	}

	statefulSets, err := c.getCacheLister(namespace).statefulSetLister.List(labels.Everything())
	if err != nil {
		return nil, err
	}
	for _, statefulSet := range statefulSets {
		if _, ok := statefulSet.Labels[constant.ApplicationLabel]; ok {
			res = append(res, &cache.WorkloadModel{
				Application: &cache.ApplicationModel{
					Name: statefulSet.Labels[constant.ApplicationLabel],
				},
				Name:   statefulSet.Name,
				Type:   constant.StatefulSetType,
				Image:  statefulSet.Spec.Template.Spec.Containers[0].Image,
				Labels: statefulSet.Labels,
			})
		}
	}

	daemonSets, err := c.getCacheLister(namespace).daemonSetLister.List(labels.Everything())
	if err != nil {
		return nil, err
	}
	for _, daemonSet := range daemonSets {
		if _, ok := daemonSet.Labels[constant.ApplicationLabel]; ok {
			res = append(res, &cache.WorkloadModel{
				Application: &cache.ApplicationModel{
					Name: daemonSet.Labels[constant.ApplicationLabel],
				},
				Name:   daemonSet.Name,
				Type:   constant.DaemonSetType,
				Image:  daemonSet.Spec.Template.Spec.Containers[0].Image,
				Labels: daemonSet.Labels,
			})
		}
	}

	return res, nil
}

func (c *KubernetesCache) GetWorkloadsWithSelector(namespace string, selector selector.Selector) ([]*cache.WorkloadModel, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	res := make([]*cache.WorkloadModel, 0)

	deployments, err := c.getCacheLister(namespace).deploymentLister.List(selector.AsLabelsSelector())
	if err != nil {
		return nil, err
	}
	for _, deployment := range deployments {
		res = append(res, &cache.WorkloadModel{
			Application: &cache.ApplicationModel{
				Name: deployment.Labels[constant.ApplicationLabel],
			},
			Name:   deployment.Name,
			Type:   constant.DeploymentType,
			Image:  deployment.Spec.Template.Spec.Containers[0].Image,
			Labels: deployment.Labels,
		})
	}

	statefulSets, err := c.getCacheLister(namespace).statefulSetLister.List(selector.AsLabelsSelector())
	if err != nil {
		return nil, err
	}
	for _, statefulSet := range statefulSets {
		res = append(res, &cache.WorkloadModel{
			Application: &cache.ApplicationModel{
				Name: statefulSet.Labels[constant.ApplicationLabel],
			},
			Name:   statefulSet.Name,
			Type:   constant.StatefulSetType,
			Image:  statefulSet.Spec.Template.Spec.Containers[0].Image,
			Labels: statefulSet.Labels,
		})
	}

	daemonSets, err := c.getCacheLister(namespace).daemonSetLister.List(selector.AsLabelsSelector())
	if err != nil {
		return nil, err
	}
	for _, daemonSet := range daemonSets {
		res = append(res, &cache.WorkloadModel{
			Application: &cache.ApplicationModel{
				Name: daemonSet.Labels[constant.ApplicationLabel],
			},
			Name:   daemonSet.Name,
			Type:   constant.DaemonSetType,
			Image:  daemonSet.Spec.Template.Spec.Containers[0].Image,
			Labels: daemonSet.Labels,
		})
	}

	return res, nil
}

func (c *KubernetesCache) GetInstances(namespace string) ([]*cache.InstanceModel, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	res := make([]*cache.InstanceModel, 0)

	pods, err := c.getCacheLister(namespace).podLister.List(labels.Everything())
	if err != nil {
		return nil, err
	}
	for _, pod := range pods {
		if _, ok := pod.Labels[constant.ApplicationLabel]; ok {
			res = append(res, &cache.InstanceModel{
				Application: &cache.ApplicationModel{
					Name: pod.Labels[constant.ApplicationLabel],
				},
				Workload: &cache.WorkloadModel{
					// TODO: implement me
				},
				Name:   pod.Name,
				Ip:     pod.Status.PodIP,
				Status: string(pod.Status.Phase),
				Node:   pod.Spec.NodeName,
				Labels: pod.Labels,
			})
		}
	}

	return res, nil
}

func (c *KubernetesCache) GetInstancesWithSelector(namespace string, selector selector.Selector) ([]*cache.InstanceModel, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	res := make([]*cache.InstanceModel, 0)

	pods, err := c.getCacheLister(namespace).podLister.List(selector.AsLabelsSelector())
	if err != nil {
		return nil, err
	}
	for _, pod := range pods {
		res = append(res, &cache.InstanceModel{
			Application: &cache.ApplicationModel{
				Name: pod.Labels[constant.ApplicationLabel],
			},
			Workload: &cache.WorkloadModel{
				// TODO: implement me
			},
			Name:   pod.Name,
			Ip:     pod.Status.PodIP,
			Status: string(pod.Status.Phase),
			Node:   pod.Spec.NodeName,
			Labels: pod.Labels,
		})
	}

	return res, nil
}

func (c *KubernetesCache) GetServices(namespace string) ([]*cache.ServiceModel, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	res := make([]*cache.ServiceModel, 0)
	services, err := c.getCacheLister(namespace).serviceLister.List(labels.Everything())
	if err != nil {
		return nil, err
	}
	for _, service := range services {
		if _, ok := service.Labels[constant.ApplicationLabel]; ok {
			res = append(res, &cache.ServiceModel{
				Application: &cache.ApplicationModel{
					Name: service.Labels[constant.ApplicationLabel],
				},
				Category:   constant.ProviderSide,
				Name:       service.Name,
				Labels:     service.Labels,
				ServiceKey: service.Labels[constant.ServiceKeyLabel],
				Group:      service.Labels[constant.GroupLabel],
				Version:    service.Labels[constant.VersionLabel],
			})
		}
	}

	return res, nil
}

func (c *KubernetesCache) GetServicesWithSelector(namespace string, selector selector.Selector) ([]*cache.ServiceModel, error) {
	c.lock.RLock()
	defer c.lock.RUnlock()

	res := make([]*cache.ServiceModel, 0)
	services, err := c.getCacheLister(namespace).serviceLister.List(selector.AsLabelsSelector())
	if err != nil {
		return nil, err
	}
	for _, service := range services {
		res = append(res, &cache.ServiceModel{
			Application: &cache.ApplicationModel{
				Name: service.Labels[constant.ApplicationLabel],
			},
			Category:   constant.ProviderSide,
			Name:       service.Name,
			Labels:     service.Labels,
			ServiceKey: service.Labels[constant.ServiceKeyLabel],
			Group:      service.Labels[constant.GroupLabel],
			Version:    service.Labels[constant.VersionLabel],
		})
	}

	return res, nil
}

// getCacheLister returns the cache lister for the given namespace if the cache is namespace scoped, otherwise it returns the cluster cache lister.
func (c *KubernetesCache) getCacheLister(namespace string) *cacheLister {
	if c.clusterScoped {
		return c.clusterCacheLister
	}
	return c.namespaceCacheLister[namespace]
}

func (c *KubernetesCache) createInformer(namespace string) informers.SharedInformerFactory {
	var informer informers.SharedInformerFactory
	if c.clusterScoped {
		informer = informers.NewSharedInformerFactoryWithOptions(c.client, c.refreshDuration)
	} else {
		informer = informers.NewSharedInformerFactoryWithOptions(c.client, c.refreshDuration, informers.WithNamespace(namespace))
	}

	lister := &cacheLister{
		deploymentLister:  informer.Apps().V1().Deployments().Lister(),
		statefulSetLister: informer.Apps().V1().StatefulSets().Lister(),
		daemonSetLister:   informer.Apps().V1().DaemonSets().Lister(),
		serviceLister:     informer.Core().V1().Services().Lister(),
		endpointLister:    informer.Core().V1().Endpoints().Lister(),
		podLister:         informer.Core().V1().Pods().Lister(),
		replicaSetLister:  informer.Apps().V1().ReplicaSets().Lister(),
		configMapLister:   informer.Core().V1().ConfigMaps().Lister(),
	}
	lister.cachesSynced = append(lister.cachesSynced,
		informer.Apps().V1().Deployments().Informer().HasSynced,
		informer.Apps().V1().StatefulSets().Informer().HasSynced,
		informer.Apps().V1().DaemonSets().Informer().HasSynced,
		informer.Core().V1().Services().Informer().HasSynced,
		informer.Core().V1().Endpoints().Informer().HasSynced,
		informer.Core().V1().Pods().Informer().HasSynced,
		informer.Apps().V1().ReplicaSets().Informer().HasSynced,
		informer.Core().V1().ConfigMaps().Informer().HasSynced,
	)

	if c.clusterScoped {
		c.clusterCacheLister = lister
	} else {
		c.namespaceCacheLister[namespace] = lister
	}

	return informer
}

// startInformers starts informers to sync data from kubernetes.
func (c *KubernetesCache) startInformers(namespaces ...string) error {
	logger.Infof("[dubbo-cp cache] Starting informers")
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.clusterScoped {
		if err := c.startInformer(""); err != nil {
			return err
		}
	} else {
		for _, namespace := range namespaces {
			if err := c.startInformer(namespace); err != nil {
				return err
			}
		}
	}
	return nil
}

// startInformer starts the informer for the given namespace.
func (c *KubernetesCache) startInformer(namespace string) error {
	informer := c.createInformer(namespace)
	var scope string
	stop := make(chan struct{})
	if c.clusterScoped {
		scope = "cluster-scoped"
		c.clusterStopChan = stop
	} else {
		scope = fmt.Sprintf("namespace-scoped for namespace: %s", namespace)
		c.namespaceStopChan[namespace] = stop
	}
	logger.Debugf("[dubbo-cp cache] Starting %s informer", scope)
	go informer.Start(stop)

	logger.Infof("[dubbo-cp cache] Waiting for %s informer caches to sync", scope)
	if !kubeToolsCache.WaitForCacheSync(stop, c.getCacheLister(namespace).cachesSynced...) {
		logger.Errorf("[dubbo-cp cache] Failed to sync %s informer caches", scope)
		return fmt.Errorf("failed to sync %s informer caches", scope)
	}
	logger.Infof("[dubbo-cp cache] Synced %s informer caches", scope)

	return nil
}

func (c *KubernetesCache) stopInformers() {
	logger.Infof("[dubbo-cp cache] Stopping informers")
	c.lock.Lock()
	defer c.lock.Unlock()
	if c.clusterScoped {
		c.stopInformer("")
	} else {
		for namespace := range c.namespaceStopChan {
			logger.Debugf("[dubbo-cp cache] Stopping informer for namespace: %s", namespace)
			c.stopInformer(namespace)
		}
	}
}

func (c *KubernetesCache) stopInformer(namespace string) {
	if c.clusterScoped {
		close(c.clusterStopChan)
	} else {
		if ch, ok := c.namespaceStopChan[namespace]; ok {
			close(ch)
			delete(c.namespaceStopChan, namespace)
			delete(c.namespaceCacheLister, namespace)
		}
	}
}