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

import (
	"fmt"
	"io/ioutil"
	"os"
	"path"

	"github.com/spf13/cobra"

	"github.com/pingcap/mimir/pkg/constant"
	"github.com/pingcap/mimir/util"
)

func newPlayCmd() *cobra.Command {
	cmd := &cobra.Command{
		Use:   `play`,
		Short: `play with flink SQL`,
		RunE: func(cmd *cobra.Command, args []string) error {
			home := constant.HomeDir()

			endpoint = getEndpoint(path.Join(home, "tls"))

			if !flinkInstalled(home) {
				fmt.Printf("flink not found in directory `%s`, downloading from %s:\n", home, constant.DefaultFlinkDownloadUrl)
				if err := installFlink(home); err != nil {
					return err
				}
			}

			cluster, err := newFlinkCluster(path.Join(home, "flink"))
			if err != nil {
				return err
			}
			if endpoint == "local" || endpoint == "none" {
				endpoint = ""
			}

			fmt.Println("bootstraping...")
			if err := cluster.bootstrap(endpoint); err != nil {
				return err
			}
			if err := cluster.bootSQLClient(); err != nil {
				return err
			}
			cluster.terminate()
			return nil
		},
	}

	return cmd
}

func flinkInstalled(home string) bool {
	return util.IsExist(path.Join(home, "flink"))
}

func installFlink(home string) error {
	tempDir, err := ioutil.TempDir("/tmp", "flink-*")
	if err != nil {
		return err
	}
	defer os.Remove(tempDir)

	reader, err := util.Download(constant.DefaultFlinkDownloadUrl, "", 0)
	if err != nil {
		return err
	}
	defer reader.Close()

	if err := util.Untar(reader, tempDir); err != nil {
		return err
	}

	return util.Copy(path.Join(tempDir, "flink-"+constant.FlinkVersion), path.Join(home, "flink"))
}