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

import (
	"bytes"
	"errors"
)

// ParseFlinkRunApplicationOutput
// return flink cluster id & jm endpoint
func ParseFlinkRunApplicationOutput(out []byte) (string, string) {
	if len(out) == 0 {
		return "", ""
	}

	lines := bytes.Split(bytes.TrimSpace(out), []byte("\n"))
	bs := lines[len(lines)-1]
	fields := bytes.Fields(bs)
	if len(fields) == 0 || len(fields) != 16 {
		return "", ""
	}
	return string(fields[10]), string(fields[15])
}

// ParseFlinkListOutput, return flink job ids in this flink kubernetes application
// TODO: structured parsing
func ParseFlinkListOutput(out []byte) ([]string, error) {
	flinkJobIds := make([]string, 0)
	if len(out) == 0 {
		return flinkJobIds, nil
	}

	lines := bytes.Split(bytes.TrimSpace(out), []byte("\n"))

	descriptorLine := lines[0]
	if !bytes.Contains(descriptorLine, []byte("successfully")) {
		return flinkJobIds, errors.New("flink list jobs failed")
	}

	jobLinePtr := 2
	for jobLinePtr < len(lines) {
		fields := bytes.Fields(lines[jobLinePtr])
		if false ||
			len(fields) < 3 ||
			bytes.Equal(fields[0], []byte("No")) ||
			len(fields) < 4 ||
			bytes.Contains(fields[0], bytes.Repeat([]byte("-"), 8)) ||
			len(fields) < 6 {
			jobLinePtr++
			continue
		}
		flinkJobIds = append(flinkJobIds, string(fields[3]))
		jobLinePtr++
	}

	return flinkJobIds, nil
}

func ParseFlinkStopOutput(out []byte) (string, error) {
	lines := bytes.Split(bytes.TrimSpace(out), []byte("\n"))

	if len(lines) != 3 {
		return "", errors.New("invalid flink stop output")
	}

	cols := bytes.Fields(lines[2])
	if len(cols) != 4 {
		return "", errors.New("invalid flink stop output")
	}

	// TODO: validate fs schema
	return string(cols[3]), nil
}