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

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"time"

	"github.com/gorilla/mux"
	"github.com/heptiolabs/healthcheck"
	"github.com/pingcap/fn"
	"github.com/pingcap/log"
	"go.uber.org/zap"

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

type Endpoint struct {
	apiService service.APIService
}

func New(apiSvc service.APIService) (*Endpoint, error) {
	return &Endpoint{
		apiService: apiSvc,
	}, nil
}

func (ep *Endpoint) SubmitJob(request *http.Request) (*protos.SubmitJobData, error) {
	// TODO: add more error logging
	param := new(protos.SubmitJobParam)
	err := json.NewDecoder(request.Body).Decode(param)
	if err != nil {
		return nil, errors.ErrInvalidParameter("param can not be decoded")
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	ctx := request.Context()
	return ep.apiService.SubmitJob(ctx, param)
}

func (ep *Endpoint) SubmitJobV2(request *http.Request) (*protos.SubmitJobData, error) {
	// TODO: add more error logging
	param := new(protos.SubmitJobParamV2)
	err := json.NewDecoder(request.Body).Decode(param)
	if err != nil {
		return nil, errors.ErrInvalidParameter("param can not be decoded")
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	ctx := request.Context()
	return ep.apiService.SubmitJobV2(ctx, param)
}

func (ep *Endpoint) ListJob(ctx context.Context, request *http.Request) (*protos.ListJobData, error) {
	param := new(protos.ListJobParam)
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	return ep.apiService.ListJob(ctx, param)
}

func (ep *Endpoint) StopJob(ctx context.Context, request *http.Request) (*protos.StopJobData, error) {
	param := new(protos.StopJobParam)
	if err := json.NewDecoder(request.Body).Decode(param); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	return ep.apiService.StopJob(ctx, param)
}

func (ep *Endpoint) ResumeJob(ctx context.Context, request *http.Request) (*protos.ResumeJobData, error) {
	param := new(protos.ResumeJobParam)
	if err := json.NewDecoder(request.Body).Decode(param); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	return ep.apiService.ResumeJob(ctx, param)
}

func (ep *Endpoint) RmJob(ctx context.Context, request *http.Request) (*protos.RmJobData, error) {
	param := new(protos.RmJobParam)
	if err := json.NewDecoder(request.Body).Decode(param); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	return ep.apiService.RmJob(ctx, param)
}

func (ep *Endpoint) GetLogs(ctx context.Context, request *http.Request) (io.ReadCloser, error) {
	id, ok := mux.Vars(request)["id"]
	if !ok {
		return nil, errors.ErrInvalidParameter("missing param job_id")
	}
	log.Info("get job logs", zap.String("id", id))
	param := &protos.GetLogsParam{
		JobID: id,
	}
	if err := param.FromQuery(request.URL.Query()); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	if request.URL.Query().Get("since_time") != "" {
		t, err := time.Parse(time.RFC3339, request.URL.Query().Get("since_time"))
		if err != nil {
			return nil, errors.ErrInvalidParameter(err.Error())
		}
		param.SinceTime = &t
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(err.Error())
	}
	stream, err := ep.apiService.GetLogs(ctx, param)
	if err != nil {
		log.Error("get job logs failed", zap.Error(err))
		return nil, err
	}
	return stream, nil
}

func (ep *Endpoint) ShowSql(ctx context.Context, request *http.Request) (*protos.ShowJobSqlData, error) {
	param := new(protos.ShowJobSqlParam)
	if err := json.NewDecoder(request.Body).Decode(param); err != nil {
		return nil, errors.ErrInvalidParameter(fmt.Sprintf("decode param error: %v", err))
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(fmt.Sprintf("param invalid: %v", err))
	}
	return ep.apiService.ShowSql(ctx, param)
}

func (ep *Endpoint) CreateIngress(ctx context.Context, request *http.Request) (*protos.CreateIngressData, error) {
	param := new(protos.CreateIngressParam)
	if err := json.NewDecoder(request.Body).Decode(param); err != nil {
		return nil, errors.ErrInvalidParameter(fmt.Sprintf("decode param error: %v", err))
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(fmt.Sprintf("param invalid: %v", err))
	}
	return ep.apiService.CreateIngress(ctx, param)
}

func (ep *Endpoint) RmIngress(ctx context.Context, request *http.Request) (*protos.RmIngressData, error) {
	param := new(protos.RmIngressParam)
	if err := json.NewDecoder(request.Body).Decode(param); err != nil {
		return nil, errors.ErrInvalidParameter(fmt.Sprintf("decode param error: %v", err))
	}
	if err := param.Validate(); err != nil {
		return nil, errors.ErrInvalidParameter(fmt.Sprintf("param invalid: %v", err))
	}
	return ep.apiService.RmIngress(ctx, param)
}

func (ep *Endpoint) ListIngress(ctx context.Context, request *http.Request) (*protos.ListIngressData, error) {
	return ep.apiService.ListIngress(ctx)
}

func (ep *Endpoint) FlinkSessionProxy(prefix string) http.Handler {
	return ep.apiService.FlinkSessionProxy(prefix)
}

func (ep *Endpoint) MountOnMux(router *mux.Router) {
	apiV1 := router.PathPrefix("/api/v1").Subrouter()
	apiV1.Handle("/submit/flink", fn.Wrap(ep.SubmitJob)).Methods(http.MethodPost)
	apiV1.Handle("/list", fn.Wrap(ep.ListJob)).Methods(http.MethodGet)
	apiV1.Handle("/stop", fn.Wrap(ep.StopJob)).Methods(http.MethodPost)
	apiV1.Handle("/resume", fn.Wrap(ep.ResumeJob)).Methods(http.MethodPost)
	apiV1.Handle("/logs/{id}", &StreamResponser{ep.GetLogs}).Methods(http.MethodGet)
	apiV1.Handle("/rm", fn.Wrap(ep.RmJob)).Methods(http.MethodPost)
	apiV1.Handle("/show-sql", fn.Wrap(ep.ShowSql)).Methods(http.MethodPost)
	apiV1.Handle("/create-ingress", fn.Wrap(ep.CreateIngress)).Methods(http.MethodPost)
	router.PathPrefix("/flink-api").Handler(ep.FlinkSessionProxy("/flink-api"))

	apiV2 := router.PathPrefix("/api/v2").Subrouter()
	apiV2.Handle("/sink/create", fn.Wrap(ep.CreateSink)).Methods(http.MethodPost)
	apiV2.Handle("/sink/list", fn.Wrap(ep.ListSink)).Methods(http.MethodGet)
	apiV2.Handle("/sink/rm", fn.Wrap(ep.RmSink)).Methods(http.MethodPost)
	apiV1.Handle("/rm-ingress", fn.Wrap(ep.RmIngress)).Methods(http.MethodPost)
	apiV2.Handle("/submit", fn.Wrap(ep.SubmitJobV2)).Methods(http.MethodPost)
	apiV2.Handle("/list-ingress", fn.Wrap(ep.ListIngress)).Methods(http.MethodGet)

	log.Info("registering health checkers")
	healthHandler := healthcheck.NewHandler()
	healthHandler.AddLivenessCheck("apiserver", livenessCheck)
	healthHandler.AddReadinessCheck("apiserver", readinessCheck)
	router.Handle("/live", healthHandler)
	router.Handle("/ready", healthHandler)
}

// Ping is for health checking.
func Ping(resp http.ResponseWriter, req *http.Request) {
	if _, err := resp.Write([]byte("OK")); err != nil {
		log.Error("write error", zap.Error(err))
	}
}

func livenessCheck() error {
	return nil
}

func readinessCheck() error {
	return nil
}