package apiserver

import (
	"bytes"
	"database/sql"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/tep/db/entity"
	"github.com/pingcap/tep/pkg/controller"
	"github.com/pingcap/tep/pkg/local"
	"github.com/pingcap/tep/pkg/metrics/client"
	"github.com/pingcap/tep/pkg/metrics/provider"
	"github.com/pingcap/tep/pkg/operator"
	"github.com/pingcap/tep/pkg/s3client"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tep/pkg/utils/alert"
	"github.com/pingcap/tep/pkg/utils/cronutil"
	"github.com/pingcap/tep/pkg/utils/es"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/remotecommand"
	// mysql driver
	_ "github.com/go-sql-driver/mysql"
	"github.com/jinzhu/gorm"
	"github.com/pingcap/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
	commoncfg "github.com/prometheus/common/config"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
)

var (
	EnablePortForward  bool
	Path2ConfigMapName map[string]string
)

const (
	AuditPlugin           = "audit"
	WhitelistPlugin       = "whitelist"
	TEPWebhook            = "pingcap.tep.webhook"
	UserSettingsPath      = "user-settings"
	DefaultSettingsPath   = "default-settings"
	userSettingsName      = "user-platform-settings"
	defaultSettingsName   = "default-platform-settings"
	UTCOffsetConfigMapKey = "utcOffset"
)

func init() {
	Path2ConfigMapName = map[string]string{
		UserSettingsPath:    userSettingsName,
		DefaultSettingsPath: defaultSettingsName,
	}
}

type Store struct {
	RestConfig                      *rest.Config
	DB                              *gorm.DB
	Operator                        operator.Interface
	KubeClient                      *kubernetes.Clientset
	TidbClusterClientWithoutTimeout *versioned.Clientset
	TidbClusterClient               *versioned.Clientset
	S3Client                        *s3client.S3Client
	Conf                            es.ESConf
	Forwarder                       *local.PortForwarder
	controller                      *controller.Controller
	K8SMetricsProvider              *provider.PrometheusProvider
	PodExec                         *remotecommand.Executor
	StatusMonitor                   *ClusterStatusMonitor
	ServerCIDR                      string
	AlertmanagerNamespace           string
	AlertmanagerServiceName         string
	AlertmanagerSecretName          string
	Namespace                       string
	MetadataBackupImage             string
	DSN                             string
	K8sPromNamespace                string
}

func NewStore(dsn string, cfg *rest.Config, conf es.ESConf, cidr string, k8sPromURL string, alertmanagerNamespace string, alertmanagerServiceName string, alertmanagerSecretName string, tepWebHookPath string, metadataBackupImage string, alertRetention time.Duration, k8sPromNamespace string, kubeClientTimeout time.Duration) (*Store, error) {
	db, err := gorm.Open("mysql", dsn)
	if err != nil {
		return nil, err
	}
	db.SingularTable(true)

	cfgWithoutTimeout := *cfg
	cfg.Timeout = kubeClientTimeout
	kubeClient, err := kubernetes.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	clusterClient, err := versioned.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	clusterClientWithoutTimeout, err := versioned.NewForConfig(&cfgWithoutTimeout)
	if err != nil {
		return nil, err
	}
	log.Info("TidbClusterClient's timeout", zap.Duration("timeout", cfg.Timeout))
	log.Info("TidbClusterClientWithoutTimeout's timeout", zap.Duration("timeout", cfgWithoutTimeout.Timeout))

	var portForwarder *local.PortForwarder = nil
	if EnablePortForward {
		portForwarder, err = local.NewPortForwarder(cfg)
		if err != nil {
			return nil, err
		}
	}

	operatorClient := operator.NewOperatorClient(clusterClient)

	ctrl := controller.New(kubeClient, clusterClient)

	ctrl.Forwarder = portForwarder

	k8sProvider, err := NewMetricsProvider(k8sPromURL)
	if err != nil {
		return nil, err
	}

	go func() {
		d := time.Duration(30) * time.Second
		t := time.NewTicker(d)
		defer t.Stop()

		for {
			u, err := url.Parse(tepWebHookPath)
			if err != nil {
				sleepWithErr("parse url failed", err)
				continue
			}

			cfg, err := ctrl.GetAlertmanagerConfig(alertmanagerSecretName, alertmanagerNamespace)
			if err != nil {
				sleepWithErr("get alertmanager config failed", err)
				continue
			}

			hookURL := &alert.URL{u}

			receviers := make([]*alert.Receiver, 0)
			existed := false
			stream.FromArray(cfg.Receivers).Filter(func(r *alert.Receiver) bool {
				if r.Name == TEPWebhook {
					return true
				}

				receviers = append(receviers, r)
				return false
			}).Each(func(r *alert.Receiver) {
				if r.WebhookConfigs != nil && len(r.WebhookConfigs) > 0 {
					if r.WebhookConfigs[0].URL.String() == hookURL.String() {
						receviers = append(receviers, r)
						existed = true
					}
				}
			})

			if existed {
				break
			}

			log.Info("create alertmanager webhook")
			cfg.Receivers = receviers

			route := alert.Route{Receiver: TEPWebhook, Continue: true}
			cfg.Route.Routes = append(cfg.Route.Routes, &route)

			webconfig := &alert.WebhookConfig{
				URL: hookURL,
				HTTPConfig: &commoncfg.HTTPClientConfig{
					TLSConfig: commoncfg.TLSConfig{
						InsecureSkipVerify: true,
					},
				},
				NotifierConfig: alert.NotifierConfig{
					VSendResolved: true,
				},
			}

			receiver := alert.Receiver{
				Name: TEPWebhook,
				WebhookConfigs: []*alert.WebhookConfig{
					webconfig,
				},
			}

			cfg.Receivers = append(cfg.Receivers, &receiver)

			err = ctrl.AlertmanagerReload(cfg, alertmanagerSecretName, alertmanagerNamespace)
			if err != nil {
				sleepWithErr("reload config failed", err)
			} else {
				log.Info("create alertmanager webhook successful")
			}
		}
	}()

	go func() {
		d := time.Duration(1) * time.Hour
		t := time.NewTicker(d)
		defer t.Stop()
		for {
			<-t.C

			alert.DeleteExpireAlerts(time.Now().Add(-alertRetention).Unix(), db.DB())
		}
	}()

	store := &Store{
		DB:                              db,
		Operator:                        operatorClient,
		controller:                      ctrl,
		Conf:                            conf,
		Forwarder:                       portForwarder,
		ServerCIDR:                      cidr,
		K8SMetricsProvider:              k8sProvider,
		AlertmanagerNamespace:           alertmanagerNamespace,
		AlertmanagerServiceName:         alertmanagerServiceName,
		AlertmanagerSecretName:          alertmanagerSecretName,
		Namespace:                       os.Getenv("NAMESPACE"),
		MetadataBackupImage:             metadataBackupImage,
		DSN:                             dsn,
		KubeClient:                      kubeClient,
		TidbClusterClient:               clusterClient,
		TidbClusterClientWithoutTimeout: clusterClientWithoutTimeout,
		RestConfig:                      cfg,
		K8sPromNamespace:                k8sPromNamespace,
	}

	store.StatusMonitor = &ClusterStatusMonitor{
		store: store,
	}
	store.StatusMonitor.Start(make(chan struct{}))

	return store, nil
}

func sleepWithErr(errMsg string, err error) {
	log.Error(errMsg, zap.Error(err))
	time.Sleep(time.Second * 30)
}

func NewMetricsProvider(baseURLString string) (*provider.PrometheusProvider, error) {
	baseURL, err := url.Parse(baseURLString)
	if err != nil {
		log.Error("failed to parse the url", zap.String("url", baseURLString))
		return nil, errors.NewBadRequest(fmt.Sprintf("failed to parse the url: %s", baseURLString))
	}

	c := client.NewClient(&http.Client{}, baseURL)

	return provider.NewPrometheusProvider(c), nil
}

func (s *Store) Controller() *controller.Controller {
	return s.controller
}

func (s *Store) GetTiDBConnection(tc *v1alpha1.TidbCluster) (*sql.DB, error) {
	dbConn, err := s.controller.GetDBConn(tc)
	if err != nil {
		log.Error("failed to get tidb connection", zap.String("TidbCluster", tc.GetNamespace()), zap.Error(err))
		return nil, errors.NewInternalError(err)
	}
	return dbConn.GetDB(), nil
}

func (s *Store) AddEvent(event *utils.Event) error {
	return utils.AddEvent(s.DB, event)
}

//do not need to restart the tidb, just send flush command
func (s *Store) FlushPlugin(db *sql.DB, pluginName string) error {
	_, err := db.Exec(fmt.Sprintf("flush tidb plugins %s;", pluginName))
	if err != nil {
		log.Error("flush plugin failed", zap.String("plugin", pluginName), zap.Error(err))
		return errors.NewInternalError(err)
	}
	return nil
}

func (s *Store) recordOperateEvent(userID string, content []byte, opType entity.OperationType) error {
	operateEvent := entity.OperateEvent{
		ManagementType: string(entity.InstanceManagement),
		User:           userID,
		OperationType:  string(opType),
		OperationTime:  time.Now().Unix(),
	}
	if len(content) != 0 {
		operateEvent.Content = string(content)
	}
	return s.DB.Create(&operateEvent).Error
}

func (s *Store) GetAlertmanagerURL() string {
	return fmt.Sprintf("%s.%s.svc:9093", s.AlertmanagerServiceName, s.AlertmanagerNamespace)
}

// ExecToPodThroughAPI uninterractively exec to the pod with the command specified.
// :param string command: list of the str which specify the command.
// :param string pod_name: Pod name
// :param string namespace: namespace of the Pod.
// :param io.Reader stdin: Standerd Input if necessary, otherwise `nil`
// :return: string: Output of the command. (STDOUT)
//          string: Errors. (STDERR)
//           error: If any error has occurred otherwise `nil`
func (s *Store) ExecToPodThroughAPI(command, containerName, podName, namespace string, stdin io.Reader) (string, string, error) {
	req := s.KubeClient.CoreV1().RESTClient().Post().
		Resource("pods").
		Name(podName).
		Namespace(namespace).
		SubResource("exec")
	scheme := runtime.NewScheme()
	if err := corev1.AddToScheme(scheme); err != nil {
		return "", "", fmt.Errorf("error adding to scheme: %v", err)
	}

	parameterCodec := runtime.NewParameterCodec(scheme)
	req.VersionedParams(&corev1.PodExecOptions{
		Command:   strings.Fields(command),
		Container: containerName,
		Stdin:     stdin != nil,
		Stdout:    true,
		Stderr:    true,
		TTY:       false,
	}, parameterCodec)

	fmt.Println("Request URL:", req.URL().String())

	exec, err := remotecommand.NewSPDYExecutor(s.RestConfig, "POST", req.URL())
	if err != nil {
		return "", "", fmt.Errorf("error while creating Executor: %v", err)
	}

	var stdout, stderr bytes.Buffer
	err = exec.Stream(remotecommand.StreamOptions{
		Stdin:  stdin,
		Stdout: &stdout,
		Stderr: &stderr,
		Tty:    false,
	})
	if err != nil {
		return "", "", fmt.Errorf("error in Stream: %v", err)
	}

	return stdout.String(), stderr.String(), nil
}

func (s *Store) GetUTCOffsetFromConfigMap(cm *corev1.ConfigMap, key string) (int, error) {
	utcOffsetStr, exist := cm.Data[key]
	if !exist {
		return 0, fmt.Errorf("key[%s] not found in ConfigMap[%s]", key, cm.GetName())
	}

	utcOffset, err := strconv.Atoi(utcOffsetStr)
	if err != nil {
		return 0, fmt.Errorf("invalid UTC offset[%s], error: %v", utcOffsetStr, err)
	}

	valid, err := s.IsUTCOffsetValid(utcOffset)
	if !valid {
		return 0, err
	}

	return utcOffset, nil
}

func (s *Store) IsUTCOffsetValid(utcOffset int) (bool, error) {
	// https://en.wikipedia.org/wiki/List_of_UTC_time_offsets
	if utcOffset < -12 || utcOffset > 14 {
		return false, fmt.Errorf("invalid UTC offset[%d], it should be an integer between -12 and 14", utcOffset)
	}
	return true, nil
}

// if opposite is true , the schedule should add the UTCOffset, otherwise,
// the schedule should minu the UTCOffset
func (s *Store) ParseScheduleViaUTCOffset(schedule string, opposite bool) (string, error) {
	var UTCOffset int
	var err error
	var cm *corev1.ConfigMap
	if cm, err = s.Controller().GetConfigMap(s.Namespace, Path2ConfigMapName[UserSettingsPath]); err != nil {
		return "", err
	}
	if UTCOffset, err = s.GetUTCOffsetFromConfigMap(cm, UTCOffsetConfigMapKey); err != nil {
		return "", err
	}
	if !opposite {
		UTCOffset = -UTCOffset
	}
	return cronutil.ParseScheduleViaUTCOffset(schedule, UTCOffset)
}

func (s *Store) GetTiDBCluster(tidbClusterID string) (*v1alpha1.TidbCluster, error) {
	tc, err := s.Controller().GetTidbCluster(tidbClusterID, tidbClusterID, metav1.GetOptions{})
	if err != nil {
		if errors.IsNotFound(err) {
			msg := fmt.Errorf("tidb cluster[%s] not found: %v", tidbClusterID, err)
			return nil, errors.NewBadRequest(msg.Error())
		}
		msg := fmt.Errorf("failed to get tidb cluster[%s]: %v", tidbClusterID, err)
		return nil, errors.NewInternalError(msg)
	}
	return tc, nil
}