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

import (
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	pb "beam.apache.org/playground/backend/internal/api/v1"
	"beam.apache.org/playground/backend/internal/logger"
)

const (
	serverIpKey                              = "SERVER_IP"
	serverPortKey                            = "SERVER_PORT"
	beamSdkKey                               = "BEAM_SDK"
	beamVersionKey                           = "BEAM_VERSION"
	workingDirKey                            = "APP_WORK_DIR"
	preparedModDirKey                        = "PREPARED_MOD_DIR"
	numOfParallelJobsKey                     = "NUM_PARALLEL_JOBS"
	cacheTypeKey                             = "CACHE_TYPE"
	cacheAddressKey                          = "CACHE_ADDRESS"
	beamPathKey                              = "BEAM_PATH"
	cacheKeyExpirationTimeKey                = "KEY_EXPIRATION_TIME"
	pipelineExecuteTimeoutKey                = "PIPELINE_EXPIRATION_TIMEOUT"
	protocolTypeKey                          = "PROTOCOL_TYPE"
	launchSiteKey                            = "LAUNCH_SITE"
	projectIdKey                             = "GOOGLE_CLOUD_PROJECT"
	pipelinesFolderKey                       = "PIPELINES_FOLDER_NAME"
	defaultPipelinesFolder                   = "executable_files"
	defaultLaunchSite                        = "local"
	defaultProtocol                          = "HTTP"
	defaultIp                                = "localhost"
	defaultPort                              = 8080
	defaultSdk                               = pb.Sdk_SDK_UNSPECIFIED
	defaultBeamVersion                       = "<unknown>"
	defaultBeamJarsPath                      = "/opt/apache/beam/jars/*"
	defaultDatasetsPath                      = "/opt/playground/backend/datasets"
	defaultKafkaEmulatorExecutablePath       = "/opt/playground/backend/kafka-emulator/beam-playground-kafka-emulator.jar"
	defaultCacheType                         = "local"
	defaultCacheAddress                      = "localhost:6379"
	defaultCacheKeyExpirationTime            = time.Minute * 15
	defaultPipelineExecuteTimeout            = time.Minute * 10
	jsonExt                                  = ".json"
	configFolderName                         = "configs"
	defaultNumOfParallelJobs                 = 20
	SDKConfigPathKey                         = "SDK_CONFIG"
	defaultSDKConfigPath                     = "../sdks.yaml"
	propertyPathKey                          = "PROPERTY_PATH"
	datasetsPathKey                          = "DATASETS_PATH"
	kafkaEmulatorExecutablePathKey           = "KAFKA_EMULATOR_EXECUTABLE_PATH"
	defaultPropertyPath                      = "."
	cacheRequestTimeoutKey                   = "CACHE_REQUEST_TIMEOUT"
	defaultCacheRequestTimeout               = time.Second * 5
	cleanupSnippetsFunctionsUrlKey           = "CLEANUP_SNIPPETS_FUNCTIONS_URL"
	defaultCleanupSnippetsFunctionsUrl       = "http://cleanup_snippets:8080/"
	putSnippetFunctionsUrlKey                = "PUT_SNIPPET_FUNCTIONS_URL"
	defaultPutSnippetFunctionsUrl            = "http://put_snippet:8080/"
	incrementSnippetViewsFunctionsUrlKey     = "INCREMENT_SNIPPET_VIEWS_FUNCTIONS_URL"
	defaultIncrementSnippetViewsFunctionsUrl = "http://increment_snippet_views:8080/"
)

// Environment operates with environment structures: NetworkEnvs, BeamEnvs, ApplicationEnvs
// Environment contains all environment variables which are used by the application
type Environment struct {
	NetworkEnvs     NetworkEnvs
	BeamSdkEnvs     BeamEnvs
	ApplicationEnvs ApplicationEnvs
}

// NewEnvironment is a constructor for Environment.
// Default values:
// LogWriters: by default using os.Stdout
// NetworkEnvs: by default using defaultIp, defaultPort and defaultProtocol from constants
// BeamEnvs: by default using pb.Sdk_SDK_JAVA
// ApplicationEnvs: required field not providing by default value
func NewEnvironment(networkEnvs NetworkEnvs, beamEnvs BeamEnvs, appEnvs ApplicationEnvs) *Environment {
	svc := Environment{}
	svc.NetworkEnvs = networkEnvs
	svc.BeamSdkEnvs = beamEnvs
	svc.ApplicationEnvs = appEnvs

	return &svc
}

// GetApplicationEnvsFromOsEnvs returns ApplicationEnvs.
// Lookups in os environment variables and tries to take values for all (exclude working dir) ApplicationEnvs parameters.
// In case some value doesn't exist sets default values:
//   - pipeline execution timeout: 10 minutes
//   - cache expiration time: 15 minutes
//   - type of cache: local
//   - cache address: localhost:6379
//
// If os environment variables don't contain a value for app working dir - returns error.
func GetApplicationEnvsFromOsEnvs() (*ApplicationEnvs, error) {
	pipelineExecuteTimeout := getEnvAsDuration(pipelineExecuteTimeoutKey, defaultPipelineExecuteTimeout, "couldn't convert provided pipeline execute timeout. Using default %s\n")
	cacheExpirationTime := getEnvAsDuration(cacheKeyExpirationTimeKey, defaultCacheKeyExpirationTime, "couldn't convert provided cache expiration time. Using default %s\n")
	cacheType := getEnv(cacheTypeKey, defaultCacheType)
	cacheAddress := getEnv(cacheAddressKey, defaultCacheAddress)
	launchSite := getEnv(launchSiteKey, defaultLaunchSite)
	projectId := os.Getenv(projectIdKey)
	pipelinesFolder := getEnv(pipelinesFolderKey, defaultPipelinesFolder)
	sdkConfigPath := getEnv(SDKConfigPathKey, defaultSDKConfigPath)
	propertyPath := getEnv(propertyPathKey, defaultPropertyPath)
	datasetsPath := getEnv(datasetsPathKey, defaultDatasetsPath)
	kafkaEmulatorExecutablePath := getEnv(kafkaEmulatorExecutablePathKey, defaultKafkaEmulatorExecutablePath)
	cacheRequestTimeout := getEnvAsDuration(cacheRequestTimeoutKey, defaultCacheRequestTimeout, "couldn't convert provided cache request timeout. Using default %s\n")
	cleanupSnippetsFunctionsUrl := getEnv(cleanupSnippetsFunctionsUrlKey, defaultCleanupSnippetsFunctionsUrl)
	putSnippetFunctionsUrl := getEnv(putSnippetFunctionsUrlKey, defaultPutSnippetFunctionsUrl)
	incrementSnippetViewsFunctionsUrl := getEnv(incrementSnippetViewsFunctionsUrlKey, defaultIncrementSnippetViewsFunctionsUrl)

	if value, present := os.LookupEnv(workingDirKey); present {
		return NewApplicationEnvs(
			value,
			launchSite,
			projectId,
			pipelinesFolder,
			sdkConfigPath,
			propertyPath,
			kafkaEmulatorExecutablePath,
			datasetsPath,
			cleanupSnippetsFunctionsUrl,
			putSnippetFunctionsUrl,
			incrementSnippetViewsFunctionsUrl,
			NewCacheEnvs(
				cacheType,
				cacheAddress,
				cacheExpirationTime,
			),
			pipelineExecuteTimeout,
			cacheRequestTimeout,
		), nil
	}
	return nil, errors.New("APP_WORK_DIR env should be provided with os.env")
}

// GetNetworkEnvsFromOsEnvs returns NetworkEnvs.
// Lookups in os environment variables and takes values for ip and port.
// In case some value doesn't exist sets default values:
//   - ip:	localhost
//   - port: 8080
func GetNetworkEnvsFromOsEnvs() (*NetworkEnvs, error) {
	ip := getEnv(serverIpKey, defaultIp)
	port := defaultPort
	protocol := getEnv(protocolTypeKey, defaultProtocol)
	var err error
	if value, present := os.LookupEnv(serverPortKey); present {
		port, err = strconv.Atoi(value)
		if err != nil {
			return nil, err
		}
	}
	return NewNetworkEnvs(ip, port, protocol), nil
}

// ConfigureBeamEnvs returns BeamEnvs.
// Lookups in os environment variables and takes value for Apache Beam SDK.
// If os environment variables don't contain a value for Apache Beam SDK - returns error.
// Configures ExecutorConfig with config file.
func ConfigureBeamEnvs(workDir string) (*BeamEnvs, error) {
	sdk := pb.Sdk_SDK_UNSPECIFIED
	preparedModDir, modDirExist := os.LookupEnv(preparedModDirKey)
	numOfParallelJobs := getEnvAsInt(numOfParallelJobsKey, defaultNumOfParallelJobs)

	beamVersion := getEnv(beamVersionKey, defaultBeamVersion)

	if value, present := os.LookupEnv(beamSdkKey); present {

		switch value {
		case pb.Sdk_SDK_JAVA.String():
			sdk = pb.Sdk_SDK_JAVA
		case pb.Sdk_SDK_GO.String():
			sdk = pb.Sdk_SDK_GO
			if !modDirExist {
				return nil, errors.New("env PREPARED_MOD_DIR must be specified in the environment variables for GO sdk")
			}
		case pb.Sdk_SDK_PYTHON.String():
			sdk = pb.Sdk_SDK_PYTHON
		case pb.Sdk_SDK_SCIO.String():
			sdk = pb.Sdk_SDK_SCIO
		}
	}
	if sdk == pb.Sdk_SDK_UNSPECIFIED {
		return NewBeamEnvs(sdk, beamVersion, nil, preparedModDir, numOfParallelJobs), nil
	}
	configPath := filepath.Join(workDir, configFolderName, sdk.String()+jsonExt)
	executorConfig, err := createExecutorConfig(sdk, configPath)
	if err != nil {
		return nil, err
	}
	return NewBeamEnvs(sdk, beamVersion, executorConfig, preparedModDir, numOfParallelJobs), nil
}

// createExecutorConfig creates ExecutorConfig that corresponds to specific Apache Beam SDK.
// Configures ExecutorConfig with config file which is located at configPath.
func createExecutorConfig(apacheBeamSdk pb.Sdk, configPath string) (*ExecutorConfig, error) {
	executorConfig, err := getConfigFromJson(configPath)
	if err != nil {
		return nil, err
	}
	switch apacheBeamSdk {
	case pb.Sdk_SDK_JAVA:
		args, err := ConcatBeamJarsToString()
		if err != nil {
			return nil, fmt.Errorf("error during proccessing jars: %s", err.Error())
		}
		executorConfig.CompileArgs = append(executorConfig.CompileArgs, args)
		executorConfig.RunArgs[1] = fmt.Sprintf("%s%s", executorConfig.RunArgs[1], args)
		executorConfig.TestArgs[1] = fmt.Sprintf("%s%s", executorConfig.TestArgs[1], args)
	case pb.Sdk_SDK_GO:
		// Go sdk doesn't need any additional arguments from the config file
	case pb.Sdk_SDK_PYTHON:
		// Python sdk doesn't need any additional arguments from the config file
	case pb.Sdk_SDK_SCIO:
		// Scala sdk doesn't need any additional arguments from the config file
	}
	return executorConfig, nil
}

func ConcatBeamJarsToString() (string, error) {
	jars, err := filepath.Glob(getEnv(beamPathKey, defaultBeamJarsPath))
	if err != nil {
		return "", err
	}
	args := strings.Join(jars, ":")
	return args, nil
}

// getConfigFromJson reads a json file to ExecutorConfig
func getConfigFromJson(configPath string) (*ExecutorConfig, error) {
	file, err := ioutil.ReadFile(configPath)
	if err != nil {
		return nil, err
	}
	executorConfig := ExecutorConfig{}
	err = json.Unmarshal(file, &executorConfig)
	if err != nil {
		return nil, err
	}
	return &executorConfig, err
}

// getEnv returns an environment variable or default value
func getEnv(key, defaultValue string) string {
	if value, ok := os.LookupEnv(key); ok {
		return value
	}
	return defaultValue
}

// getEnvAsInt returns an environment variable or default value as integer
func getEnvAsInt(key string, defaultValue int) int {
	if value, present := os.LookupEnv(key); present {
		convertedValue, err := strconv.Atoi(value)
		if err != nil {
			logger.Errorf("Incorrect value for %s. Should be integer. Will be used default value: %d", key, defaultValue)
			return defaultValue
		} else {
			if convertedValue <= 0 {
				logger.Errorf("Incorrect value for %s. Should be a positive integer value but it is %d. Will be used default value: %d", key, convertedValue, defaultValue)
				return defaultValue
			} else {
				return convertedValue
			}
		}
	}
	return defaultValue
}

// getEnvAsDuration returns an environment variable or default value as duration
func getEnvAsDuration(key string, defaultValue time.Duration, errMsg string) time.Duration {
	if value, present := os.LookupEnv(key); present {
		if converted, err := time.ParseDuration(value); err == nil {
			return converted
		} else {
			log.Printf(errMsg, defaultValue)
		}
	}
	return defaultValue
}