/*
 * Copyright 2021 PingCAP, Inc.
 *
 * Licensed 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,
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package config

import (
	"errors"

	"github.com/pingcap/log"
	k8s "k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/pingcap/mimir/model"
)

// AppOptions is the overall config
type AppOptions struct {
	Server       *ServerConfig      `yaml:"server"`
	MySql        *model.Config      `yaml:"mysql"`
	Log          *log.Config        `yaml:"log"`
	Flink        *FlinkConfig       `yaml:"flink"`
	K8S          *K8SConfig         `yaml:"k8s"`
	Security     SecurityConfig     `yaml:"security"`
	Kafka        *KafkaConf         `yaml:"kafka"`
	Sink         SinkConfig         `yaml:"sink"`
	TableChecker TableCheckerConfig `yaml:"table-checker"`
	Mappers      []Mapper           `yaml:"mappers"`
}

type Mapper struct {
	Image string `yaml:"image"`
}

type KafkaConf struct {
	BrokerList string `yaml:"broker_list"`
	User       string `yaml:"user"`
	Passwd     string `yaml:"passwd"`
}

func (kc *KafkaConf) Validate() error {
	if len(kc.BrokerList) == 0 || kc.User == "" || kc.Passwd == "" {
		return errors.New("kafka conf is empty")
	}
	return nil
}

func (opt *AppOptions) FillWithDefaults() {
	opt.Server.FillWithDefaults()
	opt.MySql.FillWithDefaults()
	opt.Flink.FillWithDefaults()
}

func (opt *AppOptions) Validate() error {
	if opt.Server == nil {
		return errors.New("config server is empty")
	}
	if opt.Log == nil {
		return errors.New("config log is empty")
	}
	if opt.Flink == nil {
		return errors.New("config flink is empty")
	}
	if opt.K8S == nil {
		return errors.New("config k8s is empty")
	}
	if opt.MySql == nil {
		return errors.New("config mysql is empty")
	}
	return nil
}

// ServerConfig define config for Http Server
type ServerConfig struct {
	Addr string `yaml:"addr"` // bind address include port

	RTimeout       int `yaml:"request_timeout"`  // request timeout, in second
	WTimeout       int `yaml:"response_timeout"` // response timeout, in second
	DTimeout       int `yaml:"idle_timeout"`     // http connection idle timeout, in second
	MaxHeaderBytes int `yaml:"max_header_bytes"` // unit in byte
}

func (cfg *ServerConfig) FillWithDefaults() {
	if cfg == nil {
		return
	}
	if cfg.RTimeout == 0 {
		cfg.RTimeout = 3
	}
	if cfg.WTimeout == 0 {
		cfg.WTimeout = 10
	}
	if cfg.DTimeout == 0 {
		cfg.DTimeout = 30
	}
	if cfg.MaxHeaderBytes == 0 {
		cfg.MaxHeaderBytes = 1 << 20
	}
}

type FlinkConfig struct {
	SessionEndpoint string                 `yaml:"session_endpoint"`
	ClientPath      string                 `yaml:"client_path"`
	Image           string                 `yaml:"image"`
	RestExposedType string                 `yaml:"rest_exposed_type"`
	CMD             []string               `yaml:"cmd"`
	ServiceAccount  string                 `yaml:"service_account"`
	Persistence     FlinkPersistenceConfig `yaml:"persistence"`
}

func (c *FlinkConfig) FillWithDefaults() {
	c.Persistence.FillWithDefaults()
}

type FlinkPersistenceConfig struct {
	DefaultCheckpointInterval int      `yaml:"default_checkpoint_interval"`
	CheckpointBaseDir         string   `yaml:"checkpoint_base_dir"`
	SavepointBaseDir          string   `yaml:"savepoint_base_dir"`
	S3                        S3Config `yaml:"s3"`
}

func (c *FlinkPersistenceConfig) FillWithDefaults() {
	if c.DefaultCheckpointInterval == 0 {
		c.DefaultCheckpointInterval = 5000
	}
}

type S3Config struct {
	AccessKey string `yaml:"access_key"`
	SecretKey string `yaml:"secret_key"`
	Endpoint  string `yaml:"endpoint"`
}

type K8SConfig struct {
	InCluster      bool   `yaml:"in_cluster"`
	KubeConfigPath string `yaml:"kube_config_path"`
	MasterURL      string `yaml:"master_url"`
}

func (k8sCfg *K8SConfig) BuildClientset() (clientSet *k8s.Clientset, err error) {
	cfg, err := k8sCfg.buildRestConfig()
	if err != nil {
		return nil, err
	}

	clientSet, err = k8s.NewForConfig(cfg)
	if err != nil {
		return nil, err
	}
	return clientSet, nil
}

func (k8sCfg *K8SConfig) buildRestConfig() (*rest.Config, error) {
	var (
		cfg *rest.Config
		err error
	)

	if k8sCfg.InCluster {
		cfg, err = rest.InClusterConfig()
	} else {
		cfg, err = clientcmd.BuildConfigFromFlags(k8sCfg.MasterURL, k8sCfg.KubeConfigPath)
	}

	return cfg, err
}

func (k8sCfg *K8SConfig) BuildControllerClient() (client.Client, error) {
	cfg, err := k8sCfg.buildRestConfig()
	if err != nil {
		return nil, err
	}

	return client.New(cfg, client.Options{})
}

type SecurityConfig struct {
	TrustedCAFile string `yaml:"trusted-ca-file"`
	CertFile      string `yaml:"cert-file"`
	KeyFile       string `yaml:"key-file"`
}

type SinkConfig struct {
	DBSecret  string `yaml:"db-secret"`
	Namespace string `yaml:"namespace"`
}

type TableCheckerConfig struct {
	Image string `yaml:"image"`
}