/*
 * 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 service

import (
	"context"
	go_errors "errors"
	"fmt"
	"os/exec"
	"strings"

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap/mimir/errors"
	"github.com/pingcap/mimir/util"
)

func (svc *apiService) runApplication(ctx context.Context, namespace, clusterID, secretName string, customizedProperty map[string]string) (string, string, error) {
	if len(svc.flinkClientPath) == 0 {
		return "", "", errors.ErrMaliciousConfig("flink client path is empty")
	}
	if len(svc.flinkImage) == 0 {
		return "", "", errors.ErrMaliciousConfig("flink image is empty")
	}
	if len(svc.executorCMD) == 0 {
		return "", "", errors.ErrMaliciousConfig("executor cmd is empty")
	}
	property := make(map[string]string)
	for k, v := range customizedProperty {
		property[k] = v
	}

	property["kubernetes.service-account"] = svc.flinkClusterServiceAccount
	property["kubernetes.container.image"] = svc.flinkImage
	// Set restart strategy
	property["restart-strategy"] = "fixed-delay"
	property["restart-strategy.fixed-delay.attempts"] = "10"
	property["restart-strategy.fixed-delay.delay"] = "30 s"
	// TODO: set labels correctly
	property["kubernetes.taskmanager.labels"] = "env:qa"
	property["kubernetes.jobmanager.labels"] = "env:qa"
	property["kubernetes.secrets"] = fmt.Sprintf("%s:/opt/sql", secretName)
	if len(svc.flinkRestExposedType) != 0 {
		property["kubernetes.rest-service.exposed.type"] = svc.flinkRestExposedType
	}
	if len(namespace) > 0 {
		property["kubernetes.namespace"] = namespace
	}
	if len(clusterID) > 0 {
		property["kubernetes.cluster-id"] = clusterID
	}
	property["state.backend"] = "rocksdb"
	property["state.checkpoints.dir"] = strings.TrimRight(svc.flinkPersistenceConfig.CheckpointBaseDir, "/") + "/" + clusterID
	property["state.savepoints.dir"] = strings.TrimRight(svc.flinkPersistenceConfig.SavepointBaseDir, "/") + "/" + clusterID
	if len(svc.flinkPersistenceConfig.S3.AccessKey) > 0 && len(svc.flinkPersistenceConfig.S3.SecretKey) > 0 {
		property["s3.access-key"] = svc.flinkPersistenceConfig.S3.AccessKey
		property["s3.secret-key"] = svc.flinkPersistenceConfig.S3.SecretKey
	}
	property["s3.endpoint"] = svc.flinkPersistenceConfig.S3.Endpoint

	args := svc.buildFlinkArgs([]string{"run-application", "-t", "kubernetes-application"}, property)
	args = append(args, svc.executorCMD...)
	log.Debug("args", zap.Strings("", args))

	cmd := exec.CommandContext(ctx, svc.flinkClientPath, args...)
	out, err := cmd.CombinedOutput()
	log.Debug(string(out))
	if err != nil {
		log.Error("exec flink-cli error", zap.Error(err))
		return "", "", errors.ErrInternal(err.Error())
	}
	clusterID, jmEndpoint := util.ParseFlinkRunApplicationOutput(out)
	return clusterID, jmEndpoint, nil
}

func (svc *apiService) buildFlinkArgs(args []string, property map[string]string) []string {
	if len(svc.k8sConfigFile) != 0 {
		property["kubernetes.config.file"] = svc.k8sConfigFile
	}
	for k, v := range property {
		args = append(args, fmt.Sprintf("-D%s=%s", k, v))
	}

	return args

}

// queryFlinkJobs
// TODO: clusterID should be RUNNING when do query, or flink command would fail
func (svc *apiService) queryFlinkJobs(ctx context.Context, namespace, clusterID string) ([]string, error) {
	property := make(map[string]string)
	property["kubernetes.cluster-id"] = clusterID
	property["kubernetes.namespace"] = namespace
	args := svc.buildFlinkArgs([]string{"list", "-t", "kubernetes-application"}, property)
	log.Debug("queryFlinkJobs", zap.Strings("args", args))

	out, err := execFlinkCommand(ctx, svc.flinkClientPath, args)
	if err != nil {
		return nil, fmt.Errorf("error querying flink jobs: %w", err)
	}

	flinkJobIDs, err := util.ParseFlinkListOutput(out)
	if err != nil {
		return nil, fmt.Errorf("parse flink cmd output error: %w", err)
	}
	if len(flinkJobIDs) == 0 {
		return nil, go_errors.New("no flink job found")
	}

	return flinkJobIDs, nil
}

func (svc *apiService) cancelFlinkJob(ctx context.Context, namespace, clusterID, flinkJobID string) error {
	property := make(map[string]string)
	property["kubernetes.cluster-id"] = clusterID
	property["kubernetes.namespace"] = namespace
	args := svc.buildFlinkArgs([]string{"cancel", "-t", "kubernetes-application"}, property)
	args = append(args, flinkJobID)
	log.Debug("cancelFlinkJob", zap.Strings("args", args))

	if _, err := execFlinkCommand(ctx, svc.flinkClientPath, args); err != nil {
		return fmt.Errorf("error cancelling flink jobs: %w", err)
	}

	return nil
}

func (svc *apiService) stopFlinkJob(ctx context.Context, namespace, clusterID, flinkJobID string) (string, error) {
	property := make(map[string]string)
	property["kubernetes.cluster-id"] = clusterID
	property["kubernetes.namespace"] = namespace
	args := svc.buildFlinkArgs([]string{"stop", "-t", "kubernetes-application"}, property)
	args = append(args, flinkJobID)
	log.Debug("stopFlinkJob", zap.Strings("args", args))

	out, err := execFlinkCommand(ctx, svc.flinkClientPath, args)
	if err != nil {
		return "", fmt.Errorf("error stopping flink jobs: %w", err)
	}

	savepointPath, err := util.ParseFlinkStopOutput(out)
	if err != nil {
		return "", fmt.Errorf("error stopping flink jobs: %w", err)
	}

	return savepointPath, nil
}

func execFlinkCommand(ctx context.Context, flinkClientPath string, args []string) ([]byte, error) {
	cmd := exec.CommandContext(ctx, flinkClientPath, args...)
	out, err := cmd.CombinedOutput()
	log.Debug(string(out))
	if err != nil {
		return out, fmt.Errorf("exec flink cmd error: %s", out)
	}

	return out, nil
}