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

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"io"
	"net/http"

	"github.com/pingcap/mimir/errors"
	"github.com/pingcap/mimir/pkg/protos"
)

type Client struct {
	endpoint   string
	httpClient *http.Client
}

func New(endpoint string, tlsCfg *tls.Config) *Client {

	return &Client{
		endpoint: endpoint,
		httpClient: &http.Client{
			Transport: &http.Transport{
				Proxy:           http.ProxyFromEnvironment,
				TLSClientConfig: tlsCfg,
			},
		},
	}
}

func (c *Client) SubmitJob(ctx context.Context, param *protos.SubmitJobParam) (*protos.SubmitJobData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}

	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/submit/flink", buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.SubmitJobData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) SubmitJobV2(ctx context.Context, param *protos.SubmitJobParamV2) (*protos.SubmitJobData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}

	req, err := http.NewRequest("POST", c.endpoint+"/api/v2/submit", buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.SubmitJobData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) ListJob(ctx context.Context, param *protos.ListJobParam) (*protos.ListJobData, error) {
	req, err := http.NewRequest("GET", c.endpoint+"/api/v1/list", nil)
	if err != nil {
		return nil, err
	}

	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.ListJobData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) CreateIngress(ctx context.Context, param *protos.CreateIngressParam) (*protos.CreateIngressData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}
	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/create-ingress", buf)
	if err != nil {
		return nil, err
	}

	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.CreateIngressData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) RmIngress(ctx context.Context, param *protos.RmIngressParam) (*protos.RmIngressData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}
	req, err := http.NewRequest("POST", c.endpoint+"/api/v2/rm-ingress", buf)
	if err != nil {
		return nil, err
	}

	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.RmIngressData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) ListIngress(ctx context.Context) (*protos.ListIngressData, error) {
	req, err := http.NewRequest("GET", c.endpoint+"/api/v2/list-ingress", nil)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.ListIngressData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) StopJob(ctx context.Context, param *protos.StopJobParam) (*protos.StopJobData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}
	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/stop", buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.StopJobData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) ResumeJob(ctx context.Context, param *protos.ResumeJobParam) (*protos.ResumeJobData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}
	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/resume", buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.ResumeJobData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) RmJob(ctx context.Context, param *protos.RmJobParam) (*protos.RmJobData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}

	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/rm", buf)
	if err != nil {
		return nil, fmt.Errorf("new request failed. err: %s", err)
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("do requst failed. err: %s", err)
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, fmt.Errorf("check http failed. err: %s", err)
	}

	response := struct {
		Data protos.RmJobData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, fmt.Errorf("decode response failed. err: %s", err)
	}
	return &response.Data, nil
}

func (c *Client) GetLogs(ctx context.Context, param *protos.GetLogsParam) (io.ReadCloser, error) {
	req, err := http.NewRequest("GET", c.endpoint+"/api/v1/logs/"+param.JobID, nil)
	if err != nil {
		return nil, err
	}
	req.URL.RawQuery = param.ToQuery().Encode()
	req = req.WithContext(ctx)
	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	return res.Body, nil
}

func (c *Client) Login(ctx context.Context, param *protos.LoginParam) (*protos.LoginData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}

	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/login", buf)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.LoginData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) GetCertificate(ctx context.Context, param *protos.CertificateParam) (*protos.ClientAuthData, error) {
	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/login/"+param.Code, nil)
	if err != nil {
		return nil, err
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, err
	}

	response := struct {
		Data protos.ClientAuthData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, err
	}
	return &response.Data, nil
}

func (c *Client) ShowJobSql(ctx context.Context, param *protos.ShowJobSqlParam) (*protos.ShowJobSqlData, error) {
	buf := new(bytes.Buffer)
	if err := json.NewEncoder(buf).Encode(param); err != nil {
		return nil, err
	}
	req, err := http.NewRequest("POST", c.endpoint+"/api/v1/show-sql", buf)
	if err != nil {
		return nil, fmt.Errorf("req failed")
	}
	req = req.WithContext(ctx)

	res, err := c.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("server do failed, %s", err)
	}
	defer res.Body.Close()

	if err := checkHTTPResponse(res); err != nil {
		return nil, fmt.Errorf("check http failed, %s", err)
	}

	response := struct {
		Data protos.ShowJobSqlData `json:"data"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&response); err != nil {
		return nil, fmt.Errorf("decode data %s", err)
	}
	return &response.Data, nil
}

// checkHTTPResponse checks if an HTTP response is with normal status codes
// parse response if the status code is not health
func checkHTTPResponse(res *http.Response) error {
	if res.StatusCode >= 200 && res.StatusCode < 400 {
		return nil
	}
	var ei errors.ErrorInfo
	body, err := io.ReadAll(res.Body)
	if err != nil {
		return fmt.Errorf("read response from %s: %s", res.Request.URL, err.Error())
	}

	if err := json.Unmarshal(body, &ei); err != nil {
		return fmt.Errorf("decode response(%s): %s, %d", string(body), err.Error(), res.StatusCode)
	}
	return fmt.Errorf("%s(%d)", ei.Message, ei.Code)
}