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

import (
	"errors"
	"net/url"
	"strconv"
	"strings"
	"time"
)

type ValidatableParam interface {
	Validate() error
}

type SubmitJobParam struct {
	ClusterID          string `json:"cluster_id"`
	SQL                string `json:"sql"`
	CheckpointOn       bool   `json:"checkpoint_on"`
	CheckpointInterval int    `json:"checkpoint_interval"`
}

func (param *SubmitJobParam) Validate() error {
	if len(param.SQL) == 0 {
		return errors.New("sql is empty")
	}
	return nil
}

type SubmitJobData struct {
	ClusterID string `json:"cluster_id"`
}

type SubmitJobParamV2 struct {
	ClusterID   string `json:"cluster_id"`
	IngressName string `json:"ingress_name"`
	SinkName    string `json:"sink_name"`
}

func (param *SubmitJobParamV2) Validate() error {
	if len(param.ClusterID) == 0 ||
		len(param.IngressName) == 0 ||
		len(param.SinkName) == 0 {
		return errors.New("cluster Id, Ingress name or Sink name can't be empty")
	}
	return nil
}

type CreateSinkParam struct {
	SinkID   string `json:"sink_id"`
	Type     string `json:"type"`
	Table    string `json:"table"`    // only when type = 'tidb'
	Database string `json:"database"` // only when type = 'tidb'
	Schedule string `json:"schedule"`
}

func (param *CreateSinkParam) Validate() error {
	if param.SinkID == "" {
		return errors.New("the sink id is required")
	}
	if param.Type != "tidb" {
		return errors.New("only support TiDB at this time")
	}
	if param.Table == "" {
		return errors.New("table name must be specified")
	}
	return nil
}

type CreateSinkData struct {
	SinkID string `json:"sink_id"`
}

type ListJobParam struct {
}

func (param *ListJobParam) Validate() error {
	return nil
}

type ListJobItem struct {
	UserID    string `json:"user_id"`
	JobID     string `json:"job_id"`
	Status    string `json:"status"`
	Namespace string `json:"namespace"`
	JobType   int    `json:"job_type"`
}

type ListJobData struct {
	Items []ListJobItem `json:"items"`
}

type ListSinkItem struct {
	UserID   string `json:"user_id"`
	SinkID   string `json:"sink_id"`
	SinkType string `json:"type"`
	Table    string `json:"table"`
}

type ListSinkData struct {
	Items []ListSinkItem `json:"items"`
}

type RmSinkParam struct {
	SinkID string `json:"sink_id"`
}

func (param *RmSinkParam) Validate() error {
	if len(param.SinkID) == 0 {
		return errors.New("sink_id must be provided")
	}
	return nil
}

type RmSinkData struct {
	SinkID string `json:"sink_id"`
}

type LoginParam struct {
	State string `json:"state"`
}

type LoginData struct {
	URL string `json:"url"`
}

type CertificateParam struct {
	Code string `json:"code"`
}

// PEM format content
type ClientAuthData struct {
	Server    string `json:"server"`
	CA        string `json:"ca"`
	ClientKey string `json:"client-key"`
	Client    string `json:"client"`
}

type StopJobParam struct {
	JobID        string `json:"job_id"`
	Savepointing bool   `json:"savepointing"`
	Force        bool   `json:"force"`
}

func (param *StopJobParam) Validate() error {
	if len(param.JobID) == 0 {
		return errors.New("job_id must be provided")
	}
	return nil
}

type StopJobData struct {
	JobID string `json:"job_id"`
	Legal bool   `json:"legal"`
	Info  string `json:"info"`
}

type ResumeJobParam struct {
	JobID              string `json:"job_id"`
	CheckpointOn       bool   `json:"checkpoint_on"`
	CheckpointInterval int    `json:"checkpoint_interval"`
}

func (param *ResumeJobParam) Validate() error {
	if len(param.JobID) == 0 {
		return errors.New("job_id must be provided")
	}
	return nil
}

type ResumeJobData struct {
	JobID string `json:"job_id"`
	Info  string `json:"info"`
}

type GetLogsParam struct {
	JobID        string     `json:"job_id"`
	Follow       bool       `json:"follow"`
	Timestamps   bool       `json:"timestamps"`
	Previous     bool       `json:"previous"`
	SinceSeconds *int64     `json:"since"`
	LimitBytes   *int64     `json:"limit_bytes"`
	TailLines    *int64     `json:"tail"`
	SinceTime    *time.Time `json:"since_time"`
}

func (param *GetLogsParam) Validate() error {
	if len(param.JobID) == 0 {
		return errors.New("job_id must be provided")
	}
	return nil
}

type RmJobParam struct {
	JobID string `json:"job_id"`
	Force bool   `json:"force"`
}

func (param *RmJobParam) Validate() error {
	if len(param.JobID) == 0 {
		return errors.New("job_id must be provided")
	}
	return nil
}

type RmJobData struct {
	JobID string `json:"job_id"`
	Legal bool   `json:"legal"`
}

func (param *GetLogsParam) FromQuery(values url.Values) error {
	param.Follow = values.Get("follow") == "true"
	param.Timestamps = values.Get("timestamps") == "true"
	param.Previous = values.Get("previous") == "true"

	if values.Get("since") != "" {
		x, err := strconv.Atoi(values.Get("since"))
		if err != nil {
			return err
		}
		y := int64(x)
		param.SinceSeconds = &y
	}

	if values.Get("limit_bytes") != "" {
		x, err := strconv.Atoi(values.Get("limit_bytes"))
		if err != nil {
			return err
		}
		y := int64(x)
		param.LimitBytes = &y
	}

	if values.Get("tail") != "" {
		x, err := strconv.Atoi(values.Get("tail"))
		if err != nil {
			return err
		}
		y := int64(x)
		param.TailLines = &y
	}

	if values.Get("since_time") != "" {
		t, err := time.Parse(time.RFC3339, values.Get("since_time"))
		if err != nil {
			return err
		}
		param.SinceTime = &t
	}

	return nil
}

func (param *GetLogsParam) ToQuery() url.Values {
	values := url.Values{}
	if param.Follow {
		values.Set("follow", "true")
	}
	if param.Timestamps {
		values.Set("timestamps", "true")
	}
	if param.Previous {
		values.Set("previous", "true")
	}
	if param.SinceSeconds != nil {
		values.Set("since", strconv.Itoa(int(*param.SinceSeconds)))
	}
	if param.LimitBytes != nil {
		values.Set("limit_bytes", strconv.Itoa(int(*param.LimitBytes)))
	}
	if param.TailLines != nil {
		values.Set("tail", strconv.Itoa(int(*param.TailLines)))
	}
	if param.SinceTime != nil {
		values.Set("since_time", param.SinceTime.Format(time.RFC3339))
	}
	return values
}

type ShowJobSqlParam struct {
	ClusterID string `json:"cluster_id"`
}

type ShowJobSqlData struct {
	Sql string `json:"sql"`
}

func (param *ShowJobSqlParam) Validate() error {
	if len(param.ClusterID) == 0 {
		return errors.New("cluster id cant be empty")
	}
	return nil
}

type CreateIngressParam struct {
	Name string `json:"name"`
	Type string `json:"type"`
}

type CreateIngressData struct {
	Endpoint string `json:"endpoint"`
}

func (p *CreateIngressParam) Validate() error {
	if p.Name == "" {
		return errors.New("name Can't be empty")
	}
	p.Type = strings.ToLower(p.Type)
	if p.Type != "kafka" && p.Type != "sqs" {
		return errors.New("we only support Type in {kafka, sqs}")
	}
	return nil
}

type RmIngressParam struct {
	IngressID string `json:"ingress_id"`
	Type      string `json:"type"`
}

type RmIngressData struct {
	IngressID string `json:"ingress_id"`
}

func (p *RmIngressParam) Validate() error {
	if p.IngressID == "" {
		return errors.New("ingressid Can't be empty")
	}
	p.Type = strings.ToLower(p.Type)
	if p.Type != "kafka" && p.Type != "sqs" {
		return errors.New("we only support Type in {kafka, sqs}")
	}
	return nil
}

type ListIngressItem struct {
	UserID    string `json:"user_id"`
	IngressID string `json:"ingress_id"`
	Type      string `json:"type"`
	Endpoint  string `json:"endpoint"`
}

type ListIngressData struct {
	Items []ListIngressItem `json:"items"`
}