// 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 v1alpha1

import (
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/validation/field"
	"math/rand"
	"sigs.k8s.io/controller-reconciler/pkg/finalizer"
	"sigs.k8s.io/controller-reconciler/pkg/status"
	"time"
)

// defaults and constant strings
const (
	PasswordCharNumSpace    = "abcdefghijklmnopqrstuvwxyz0123456789"
	PasswordCharSpace       = "abcdefghijklmnopqrstuvwxyz"
	defaultRedisImage       = "redis"
	defaultRedisVersion     = "4.0"
	defaultRedisPort        = "6379"
	defaultWorkerImage      = "gcr.io/airflow-operator/airflow"
	defaultSchedulerImage   = "gcr.io/airflow-operator/airflow"
	defaultFlowerImage      = "gcr.io/airflow-operator/airflow"
	GitsyncImage            = "gcr.io/google_containers/git-sync"
	GitsyncVersion          = "v3.0.1"
	GCSsyncImage            = "gcr.io/cloud-airflow-releaser/gcs-syncd"
	GCSsyncVersion          = "cloud_composer_service_2018-05-23-RC0"
	ExecutorLocal           = "Local"
	ExecutorCelery          = "Celery"
	ExecutorSequential      = "Sequential"
	ExecutorK8s             = "Kubernetes"
	defaultExecutor         = ExecutorLocal
	defaultBranch           = "master"
	defaultWorkerVersion    = "1.10.2"
	defaultSchedulerVersion = "1.10.2"
)

var (
	random = rand.New(rand.NewSource(time.Now().UnixNano()))
)

// RandomAlphanumericString generates a random password of some fixed length.
func RandomAlphanumericString(strlen int) []byte {
	result := make([]byte, strlen)
	for i := range result {
		result[i] = PasswordCharNumSpace[random.Intn(len(PasswordCharNumSpace))]
	}
	result[0] = PasswordCharSpace[random.Intn(len(PasswordCharSpace))]
	return result
}

var allowedExecutors = []string{ExecutorLocal, ExecutorSequential, ExecutorCelery, ExecutorK8s}

// MemoryStoreSpec defines the attributes and desired state of MemoryStore component
type MemoryStoreSpec struct {
	// Project defines the SQL instance project
	Project string `json:"project"`
	// Region defines the SQL instance region
	Region string `json:"region"`
	// AlternativeLocationID - alt
	// +optional.
	AlternativeLocationID string `json:"alternativeLocationId,omitempty"`
	// AuthorizedNetwork
	// +optional.
	AuthorizedNetwork string `json:"authorizedNetwork,omitempty"`
	// LocationID The zone where the instance will be provisioned.
	// +optional
	LocationID string `json:"locationId,omitempty"`
	// MemorySizeGb: Required. Redis memory size in GiB.
	MemorySizeGb int64 `json:"memorySizeGb,omitempty"`
	// RedisConfigs: Optional. Redis configuration parameters
	RedisConfigs map[string]string `json:"redisConfigs,omitempty"`
	// RedisVersion: Optional. The version of Redis software.
	RedisVersion string `json:"redisVersion,omitempty"`
	// Tier: Required. The service tier of the instance.
	Tier string `json:"tier,omitempty"`
	// Specifies the behavior Redis follows when the memory size limit is reached.
	MaxMemoryPolicy string `json:"maxMemoryPolicy,omitempty"`
	// Allows clients to subscribe to notifications on certain keyspace events
	NotifyKeyspaceEvents string `json:"notifyKeyspaceEvents,omitempty"`
	// Status
	Status MemoryStoreStatus `json:"status,omitempty"`
}

func (s *MemoryStoreSpec) validate(fp *field.Path) field.ErrorList {
	errs := field.ErrorList{}
	if s == nil {
		return errs
	}

	if s.Project == "" {
		errs = append(errs, field.Required(fp.Child("project"), "Missing memoryStore Project"))
	}
	if s.Region == "" {
		errs = append(errs, field.Required(fp.Child("region"), "Missing memoryStore Region"))
	}

	var allowedMaxMemoryPolicy = []string{"noeviction", "allkeys-lru", "volatile-lru", "allkeys-random", "volatile-random", "Volatile-ttl"}
	allowedMMP := false
	for _, policy := range allowedMaxMemoryPolicy {
		if policy == s.MaxMemoryPolicy {
			allowedMMP = true
		}
	}

	if !allowedMMP {
		errs = append(errs, field.Invalid(fp.Child("maxMemoryPolicy"), "", "Configuration is not allowed"))
	}

	var allowedNotifyKeyspaceEvents = []string{"", "K", "E", "g", "$", "l", "s", "h", "z", "x", "e", "A"}
	allowedNKE := false
	for _, event := range allowedNotifyKeyspaceEvents {
		if event == s.NotifyKeyspaceEvents {
			allowedNKE = true
		}
	}
	if !allowedNKE {
		errs = append(errs, field.Invalid(fp.Child("notifyKeyspaceEvent"), "", "Configuration is not allowed"))
	}

	return errs
}

// RedisSpec defines the attributes and desired state of Redis component
type RedisSpec struct {
	// Image defines the Redis Docker image name
	// +optional
	Image string `json:"image,omitempty"`
	// Version defines the Redis Docker image version.
	// +optional
	Version string `json:"version,omitempty"`
	// Flag when True generates RedisReplica CustomResource to be handled by Redis Operator
	// If False, a StatefulSet with 1 replica is created
	// +optional
	Operator bool `json:"operator,omitempty"`
	// Hostname or IP of existing Redis instance
	RedisHost string `json:"redisHost,omitempty"`
	// Port of existing Redis instance
	RedisPort string `json:"redisPort,omitempty"`
	// If the existing Redis instance uses password or not, as MemoryStore doesn't support password yet
	RedisPassword bool `json:"redisPassword,omitempty"`
	// Resources is the resource requests and limits for the pods.
	Resources corev1.ResourceRequirements `json:"resources,omitempty"`
	// VolumeClaimTemplate allows a user to specify volume claim for MySQL Server files
	// +optional
	VolumeClaimTemplate *corev1.PersistentVolumeClaim `json:"volumeClaimTemplate,omitempty"`
	// AdditionalArgs for redis-server
	// +optional
	AdditionalArgs string `json:"additionalargs,omitempty"`
}

func (s *RedisSpec) validate(fp *field.Path) field.ErrorList {
	errs := field.ErrorList{}
	if s == nil {
		return errs
	}
	if s.Operator == true {
		errs = append(errs, field.Invalid(fp.Child("operator"), "", "Operator is not supported in this version"))
	}
	return errs
}

// FlowerSpec defines the attributes to deploy Flower component
type FlowerSpec struct {
	// Image defines the Flower Docker image.
	// +optional
	Image string `json:"image,omitempty"`
	// Version defines the Flower Docker image version.
	// +optional
	Version string `json:"version,omitempty"`
	// Replicas defines the number of running Flower instances in a cluster
	Replicas int32 `json:"replicas,omitempty"`
	// Resources is the resource requests and limits for the pods.
	// +optional
	Resources corev1.ResourceRequirements `json:"resources,omitempty"`
}

func (s *FlowerSpec) validate(fp *field.Path) field.ErrorList {
	errs := field.ErrorList{}
	return errs
}

// SchedulerSpec defines the attributes and desired state of Airflow Scheduler
type SchedulerSpec struct {
	// Image defines the Airflow custom server Docker image.
	// +optional
	Image string `json:"image,omitempty"`
	// Version defines the Airflow Docker image version
	// +optional
	Version string `json:"version,omitempty"`
	// DBName defines the Airflow Database to be used
	// +optional
	DBName string `json:"database,omitempty"`
	// DBUser defines the Airflow Database user to be used
	// +optional
	DBUser string `json:"dbuser,omitempty"`
	// Resources is the resource requests and limits for the pods.
	Resources corev1.ResourceRequirements `json:"resources,omitempty"`
}

func (s *SchedulerSpec) validate(fp *field.Path) field.ErrorList {
	return field.ErrorList{}
}

// WorkerSpec defines the attributes and desired state of Airflow workers
type WorkerSpec struct {
	// Image defines the Airflow worker Docker image.
	// +optional
	Image string `json:"image,omitempty"`
	// Version defines the Airflow worker Docker image version
	// +optional
	Version string `json:"version,omitempty"`
	// Replicas is the count of number of workers
	Replicas int32 `json:"replicas,omitempty"`
	// Resources is the resource requests and limits for the pods.
	Resources corev1.ResourceRequirements `json:"resources,omitempty"`
}

func (s *WorkerSpec) validate(fp *field.Path) field.ErrorList {
	return field.ErrorList{}
}

//GCSSpec defines the atributed needed to sync from a git repo
type GCSSpec struct {
	// Bucket describes the GCS bucket
	Bucket string `json:"bucket,omitempty"`
	// Once syncs initially and quits (use init container instead of sidecar)
	Once bool `json:"once,omitempty"`
}

func (s *GCSSpec) validate(fp *field.Path) field.ErrorList {
	errs := field.ErrorList{}
	if s == nil {
		return errs
	}
	if s.Bucket == "" {
		errs = append(errs, field.Required(fp.Child("bucket"), "bucket required"))
	}
	if s.Once == true {
		errs = append(errs, field.NotSupported(fp.Child("once"), "true", []string{}))
	}
	return errs
}

//GitSpec defines the atributed needed to sync from a git repo
type GitSpec struct {
	// Repo describes the http/ssh uri for git repo
	Repo string `json:"repo,"`
	// Branch describes the branch name to be synced
	Branch string `json:"branch,omitempty"`
	// Rev is the git hash to be used for syncing
	Rev string `json:"rev,omitempty"`
	// User for git access
	User string `json:"user,omitempty"`
	// Once syncs initially and quits (use init container instead of sidecar)
	Once bool `json:"once,omitempty"`
	// Reference to git credentials (user, password, ssh etc)
	CredSecretRef *corev1.LocalObjectReference `json:"cred,omitempty"`
}

func (s *GitSpec) validate(fp *field.Path) field.ErrorList {
	errs := field.ErrorList{}
	if s == nil {
		return errs
	}

	if s.Repo == "" {
		errs = append(errs, field.Required(fp.Child("repo"), "repo required"))
	}
	if s.CredSecretRef != nil && s.CredSecretRef.Name == "" {
		errs = append(errs, field.Required(fp.Child("cred", "name"), "name missing"))
	}
	//errs = append(errs, field.NotSupported(fp.Child("cred"), "", []string{}))
	return errs
}

// DagSpec defines where the DAGs are located and how to access them
type DagSpec struct {
	// DagSubdir is the directory under source where the dags are present
	DagSubdir string `json:"subdir,omitempty"`
	// GitSpec defines details to pull DAGs from a git repo using
	// github.com/kubernetes/git-sync sidecar
	Git *GitSpec `json:"git,omitempty"`
	// NfsPVSpec
	NfsPV *corev1.PersistentVolumeClaim `json:"nfspv,omitempty"`
	// Storage has s3 compatible storage spec for copying files from
	Storage *StorageSpec `json:"storage,omitempty"`
	// Gcs config which uses storage spec
	GCS *GCSSpec `json:"gcs,omitempty"`
}

func (s *DagSpec) validate(fp *field.Path) field.ErrorList {
	errs := field.ErrorList{}
	if s == nil {
		return errs
	}
	if s.NfsPV != nil {
		errs = append(errs, field.NotSupported(fp.Child("nfspv"), "", []string{}))
	}
	if s.Storage != nil {
		errs = append(errs, field.NotSupported(fp.Child("storage"), "", []string{}))
	}
	errs = append(errs, s.Git.validate(fp.Child("git"))...)
	errs = append(errs, s.GCS.validate(fp.Child("git"))...)
	return errs
}

// SecretEnv secret env
type SecretEnv struct {
	// Env - env variable name
	Env string `json:"env,omitempty"`
	// Secret - secret name
	Secret string `json:"secret,omitempty"`
	// Field - field name
	Field string `json:"field,omitempty"`
}

// ClusterConfig is used to capture the config for Airflow
type ClusterConfig struct {
	// Airflow defines a list of kv pairs that describe env variables injected into the nodes
	// +optional
	AirflowEnv map[string]string `json:"airflow,omitempty"`
	// AirflowSecret defines a list of secret envs
	// +optional
	AirflowSecretEnv []SecretEnv `json:"airflowsecret,omitempty"`
}

// AirflowClusterSpec defines the desired state of AirflowCluster
type AirflowClusterSpec struct {
	// Selector for fitting pods to nodes whose labels match the selector.
	// https://kubernetes.io/docs/concepts/configuration/assign-pod-node/
	// +optional
	NodeSelector map[string]string `json:"nodeSelector,omitempty"`
	// Define scheduling constraints for pods.
	// +optional
	Affinity *corev1.Affinity `json:"affinity,omitempty"`
	// Custom annotations to be added to the pods.
	// +optional
	Annotations map[string]string `json:"annotations,omitempty"`
	// Custom labels to be added to the pods.
	// +optional
	Labels map[string]string `json:"labels,omitempty"`
	// Airflow Executor desired: local,celery,kubernetes
	// +optional
	Executor string `json:"executor,omitempty"`
	// Airflow config as env list
	// +optional
	Config ClusterConfig `json:"config,omitempty"`
	// Spec for MemoryStore component
	// +optional
	MemoryStore *MemoryStoreSpec `json:"memoryStore,omitempty"`
	// Spec for Redis component.
	// +optional
	Redis *RedisSpec `json:"redis,omitempty"`
	// Spec for Airflow Scheduler component.
	// +optional
	Scheduler *SchedulerSpec `json:"scheduler,omitempty"`
	// Spec for Airflow Workers
	// +optional
	Worker *WorkerSpec `json:"worker,omitempty"`
	// Spec for Airflow UI component.
	// +optional
	UI *AirflowUISpec `json:"ui,omitempty"`
	// Spec for Flower component.
	// +optional
	Flower *FlowerSpec `json:"flower,omitempty"`
	// Spec for DAG source and location
	// +optional
	DAGs *DagSpec `json:"dags,omitempty"`
	// AirflowBaseRef is a reference to the AirflowBase CR
	AirflowBaseRef *corev1.LocalObjectReference `json:"airflowbase,omitempty"`
}

// SchedulerStatus defines the observed state of Airflow Scheduler
type SchedulerStatus struct {
	// DagCount is a count of number of Dags observed
	DagCount int32 `json:"dagcount,omitempty"`
	// RunCount is a count of number of Dag Runs observed
	RunCount int32 `json:"runcount,omitempty"`
}

// MemoryStoreStatus defines the observed state of MemoryStore
type MemoryStoreStatus struct {
	// CreateTime: Output only. The time the instance was created.
	CreateTime string `json:"createTime,omitempty"`
	// CurrentLocationID: Output only. The current zone where the Redis
	// endpoint is placed.
	CurrentLocationID string `json:"currentLocationId,omitempty"`
	// StatusMessage: Output only. Additional information about the current
	// status of this instance, if available.
	StatusMessage string `json:"statusMessage,omitempty"`
	// Host: Output only. Hostname or IP address of the exposed Redis endpoint used by
	// clients to connect to the service.
	Host string `json:"host,omitempty"`
	// Port: Output only. The port number of the exposed Redis endpoint.
	Port int64 `json:"port,omitempty"`
	// State: Output only. The current state of this instance.
	State                string `json:"state,omitempty"`
	status.Meta          `json:",inline"`
	status.ComponentMeta `json:",inline"`
}

// AirflowClusterStatus defines the observed state of AirflowCluster
type AirflowClusterStatus struct {
	status.Meta          `json:",inline"`
	status.ComponentMeta `json:",inline"`
}

// AirflowCluster represents the Airflow Scheduler and workers for a single DAG folder
// function. At a minimum they need a SQL service (MySQL or SQLProxy) and Airflow UI.
// In addition for an installation with minimal external dependencies, NFS and Airflow UI
// are also added.
// +kubebuilder:object:root=true
type AirflowCluster struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   AirflowClusterSpec   `json:"spec,omitempty"`
	Status AirflowClusterStatus `json:"status,omitempty"`
}

// Helper functions for the resources

// ApplyDefaults the AirflowCluster
func (b *AirflowCluster) ApplyDefaults() {
	if b.Spec.Redis != nil {
		if b.Spec.Redis.Image == "" {
			b.Spec.Redis.Image = defaultRedisImage
		}
		if b.Spec.Redis.Version == "" {
			b.Spec.Redis.Version = defaultRedisVersion
		}
		if b.Spec.Redis.RedisHost == "" {
			if b.Spec.Redis.Image == "" {
				b.Spec.Redis.Image = defaultRedisImage
			}
			if b.Spec.Redis.Version == "" {
				b.Spec.Redis.Version = defaultRedisVersion
			}
		} else if b.Spec.Redis.RedisPort == "" {
			b.Spec.Redis.RedisPort = defaultRedisPort
		}
	}
	if b.Spec.Scheduler != nil {
		if b.Spec.Scheduler.Image == "" {
			b.Spec.Scheduler.Image = defaultSchedulerImage
		}
		if b.Spec.Scheduler.Version == "" {
			b.Spec.Scheduler.Version = defaultSchedulerVersion
		}
		if b.Spec.Scheduler.DBName == "" {
			b.Spec.Scheduler.DBName = string(RandomAlphanumericString(16))
		}
		if b.Spec.Scheduler.DBUser == "" {
			b.Spec.Scheduler.DBUser = string(RandomAlphanumericString(16))
		}
	}
	if b.Spec.UI != nil {
		if b.Spec.UI.Image == "" {
			b.Spec.UI.Image = defaultUIImage
		}
		if b.Spec.UI.Version == "" {
			b.Spec.UI.Version = defaultUIVersion
		}
		if b.Spec.UI.Replicas == 0 {
			b.Spec.UI.Replicas = 1
		}
	}
	if b.Spec.Flower != nil {
		if b.Spec.Flower.Image == "" {
			b.Spec.Flower.Image = defaultFlowerImage
		}
		if b.Spec.Flower.Version == "" {
			b.Spec.Flower.Version = defaultFlowerVersion
		}
		if b.Spec.Flower.Replicas == 0 {
			b.Spec.Flower.Replicas = 1
		}
	}
	if b.Spec.Executor == "" {
		b.Spec.Executor = defaultExecutor
	}
	if b.Spec.Worker != nil {
		if b.Spec.Worker.Image == "" {
			b.Spec.Worker.Image = defaultWorkerImage
		}
		if b.Spec.Worker.Version == "" {
			b.Spec.Worker.Version = defaultWorkerVersion
		}
		if b.Spec.Worker.Replicas == 0 {
			b.Spec.Worker.Replicas = 1
		}
		if b.Spec.Executor == ExecutorK8s {
			b.Spec.Worker.Replicas = 0
		}
	}
	if b.Spec.DAGs != nil {
		if b.Spec.DAGs.Git != nil {
			if b.Spec.DAGs.Git.Branch == "" {
				b.Spec.DAGs.Git.Branch = defaultBranch
			}
		}
	}
	b.Status.ComponentList = status.ComponentList{}
	finalizer.EnsureStandard(b)
}

// Validate the AirflowCluster
func (b *AirflowCluster) Validate() error {
	errs := field.ErrorList{}
	spec := field.NewPath("spec")

	errs = append(errs, b.Spec.MemoryStore.validate(spec.Child("memorystore"))...)
	errs = append(errs, b.Spec.Redis.validate(spec.Child("redis"))...)
	errs = append(errs, b.Spec.Scheduler.validate(spec.Child("scheduler"))...)
	errs = append(errs, b.Spec.Worker.validate(spec.Child("worker"))...)
	errs = append(errs, b.Spec.DAGs.validate(spec.Child("dags"))...)
	errs = append(errs, b.Spec.UI.validate(spec.Child("ui"))...)
	errs = append(errs, b.Spec.Flower.validate(spec.Child("flower"))...)

	allowed := false
	for _, executor := range allowedExecutors {
		if executor == b.Spec.Executor {
			allowed = true
		}
	}
	if !allowed {
		errs = append(errs, field.NotSupported(spec.Child("executor"), b.Spec.Executor, allowedExecutors))
	}

	if b.Spec.Scheduler == nil {
		errs = append(errs, field.Required(spec.Child("scheduler"), "scheduler required"))
	}

	if b.Spec.Executor == ExecutorCelery {
		if b.Spec.Redis == nil && b.Spec.MemoryStore == nil {
			errs = append(errs, field.Required(spec.Child("redis"), "redis/memoryStore required for Celery executor"))
		}
		if b.Spec.Worker == nil {
			errs = append(errs, field.Required(spec.Child("worker"), "worker required for Celery executor"))
		}
	}
	if b.Spec.Executor == ExecutorK8s {
		if b.Spec.Worker == nil {
			errs = append(errs, field.Required(spec.Child("worker"), "worker required for Celery executor"))
		}
	}

	if b.Spec.Flower != nil {
		if b.Spec.Executor != ExecutorCelery {
			errs = append(errs, field.Required(spec.Child("executor"), "celery executor required for Flower"))
		}
	}

	if b.Spec.AirflowBaseRef == nil {
		errs = append(errs, field.Required(spec.Child("airflowbase"), "airflowbase reference missing"))
	} else if b.Spec.AirflowBaseRef.Name == "" {
		errs = append(errs, field.Required(spec.Child("airflowbase", "name"), "name missing"))
	}

	return errs.ToAggregate()
}

// NewAirflowCluster return a defaults filled AirflowCluster object
func NewAirflowCluster(name, namespace, executor, base string, dags *DagSpec) *AirflowCluster {
	c := AirflowCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Labels:    map[string]string{},
			Namespace: namespace,
		},
	}
	c.Spec = AirflowClusterSpec{}
	c.Spec.Executor = executor
	c.Spec.Scheduler = &SchedulerSpec{}
	c.Spec.UI = &AirflowUISpec{}
	if executor == ExecutorCelery {
		c.Spec.Redis = &RedisSpec{}
		c.Spec.MemoryStore = &MemoryStoreSpec{}
		c.Spec.Redis = &RedisSpec{}
		c.Spec.Worker = &WorkerSpec{}
		c.Spec.Flower = &FlowerSpec{}
	}
	c.Spec.DAGs = dags
	c.Spec.AirflowBaseRef = &corev1.LocalObjectReference{Name: base}
	c.ApplyDefaults()
	return &c
}

// AirflowClusterList contains a list of AirflowCluster
// +kubebuilder:object:root=true
type AirflowClusterList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []AirflowCluster `json:"items"`
}

func init() {
	SchemeBuilder.Register(&AirflowCluster{}, &AirflowClusterList{})
}