/*
 * Copyright 2021 CloudWeGo Authors
 *
 * 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,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

// Package client defines the Options of client
package client

import (
	"context"
	"time"

	"github.com/cloudwego/localsession/backup"

	"github.com/cloudwego/kitex/internal/configutil"
	"github.com/cloudwego/kitex/internal/stream"
	"github.com/cloudwego/kitex/pkg/acl"
	"github.com/cloudwego/kitex/pkg/circuitbreak"
	connpool2 "github.com/cloudwego/kitex/pkg/connpool"
	"github.com/cloudwego/kitex/pkg/diagnosis"
	"github.com/cloudwego/kitex/pkg/discovery"
	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/endpoint/cep"
	"github.com/cloudwego/kitex/pkg/event"
	"github.com/cloudwego/kitex/pkg/fallback"
	"github.com/cloudwego/kitex/pkg/http"
	"github.com/cloudwego/kitex/pkg/loadbalance"
	"github.com/cloudwego/kitex/pkg/loadbalance/lbcache"
	"github.com/cloudwego/kitex/pkg/proxy"
	"github.com/cloudwego/kitex/pkg/remote"
	"github.com/cloudwego/kitex/pkg/remote/codec/protobuf"
	"github.com/cloudwego/kitex/pkg/remote/codec/thrift"
	"github.com/cloudwego/kitex/pkg/remote/connpool"
	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2"
	"github.com/cloudwego/kitex/pkg/remote/trans/nphttp2/grpc"
	"github.com/cloudwego/kitex/pkg/remote/trans/ttstream"
	"github.com/cloudwego/kitex/pkg/retry"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/serviceinfo"
	"github.com/cloudwego/kitex/pkg/stats"
	"github.com/cloudwego/kitex/pkg/transmeta"
	"github.com/cloudwego/kitex/pkg/utils"
	"github.com/cloudwego/kitex/pkg/warmup"
	"github.com/cloudwego/kitex/transport"
)

func init() {
	remote.PutPayloadCode(serviceinfo.Thrift, thrift.NewThriftCodec())
	remote.PutPayloadCode(serviceinfo.Protobuf, protobuf.NewProtobufCodec())
}

type UnaryOption struct {
	F func(o *UnaryOptions, di *utils.Slice)
}

type UnaryOptions struct {
	opts *Options

	// middlewares
	UnaryMiddlewares        []endpoint.UnaryMiddleware
	UnaryMiddlewareBuilders []endpoint.UnaryMiddlewareBuilder

	// retry policy
	RetryMethodPolicies map[string]retry.Policy
	RetryContainer      *retry.Container
	RetryWithResult     *retry.ShouldResultRetry

	// fallback policy
	Fallback *fallback.Policy
}

func (o *UnaryOptions) InitMiddlewares(ctx context.Context) {
	if len(o.UnaryMiddlewareBuilders) > 0 {
		unaryMiddlewares := make([]endpoint.UnaryMiddleware, 0, len(o.UnaryMiddlewareBuilders))
		for _, mwb := range o.UnaryMiddlewareBuilders {
			unaryMiddlewares = append(unaryMiddlewares, mwb(ctx))
		}
		o.UnaryMiddlewares = append(o.UnaryMiddlewares, unaryMiddlewares...)
	}
}

func (o *UnaryOptions) SetUnaryRPCTimeout(d time.Duration) {
	rpcinfo.AsMutableRPCConfig(o.opts.Configs).SetRPCTimeout(d)
	o.opts.Locks.Bits |= rpcinfo.BitRPCTimeout
}

type StreamOption struct {
	F func(o *StreamOptions, di *utils.Slice)
}

type StreamOptions struct {
	EventHandler                 stream.StreamEventHandler
	RecvTimeout                  time.Duration
	StreamMiddlewares            []cep.StreamMiddleware
	StreamMiddlewareBuilders     []cep.StreamMiddlewareBuilder
	StreamRecvMiddlewares        []cep.StreamRecvMiddleware
	StreamRecvMiddlewareBuilders []cep.StreamRecvMiddlewareBuilder
	StreamSendMiddlewares        []cep.StreamSendMiddleware
	StreamSendMiddlewareBuilders []cep.StreamSendMiddlewareBuilder
}

func (o *StreamOptions) InitMiddlewares(ctx context.Context) {
	if len(o.StreamMiddlewareBuilders) > 0 {
		streamMiddlewares := make([]cep.StreamMiddleware, 0, len(o.StreamMiddlewareBuilders))
		for _, mwb := range o.StreamMiddlewareBuilders {
			streamMiddlewares = append(streamMiddlewares, mwb(ctx))
		}
		o.StreamMiddlewares = append(o.StreamMiddlewares, streamMiddlewares...)
	}
	if len(o.StreamRecvMiddlewareBuilders) > 0 {
		streamRecvMiddlewares := make([]cep.StreamRecvMiddleware, 0, len(o.StreamRecvMiddlewareBuilders))
		for _, mwb := range o.StreamRecvMiddlewareBuilders {
			streamRecvMiddlewares = append(streamRecvMiddlewares, mwb(ctx))
		}
		o.StreamRecvMiddlewares = append(o.StreamRecvMiddlewares, streamRecvMiddlewares...)
	}
	if len(o.StreamSendMiddlewareBuilders) > 0 {
		streamSendMiddlewares := make([]cep.StreamSendMiddleware, 0, len(o.StreamSendMiddlewareBuilders))
		for _, mwb := range o.StreamSendMiddlewareBuilders {
			streamSendMiddlewares = append(streamSendMiddlewares, mwb(ctx))
		}
		o.StreamSendMiddlewares = append(o.StreamSendMiddlewares, streamSendMiddlewares...)
	}
}

func (o *StreamOptions) BuildRecvChain(recvEndpoint cep.StreamRecvEndpoint) cep.StreamRecvEndpoint {
	return cep.StreamRecvChain(o.StreamRecvMiddlewares...)(recvEndpoint)
}

func (o *StreamOptions) BuildSendChain(sendEndpoint cep.StreamSendEndpoint) cep.StreamSendEndpoint {
	return cep.StreamSendChain(o.StreamSendMiddlewares...)(sendEndpoint)
}

// Options is used to initialize a client.
type Options struct {
	Cli     *rpcinfo.EndpointBasicInfo
	Svr     *rpcinfo.EndpointBasicInfo
	Configs rpcinfo.RPCConfig
	Locks   *ConfigLocks
	Once    *configutil.OptionOnce

	UnaryOptions  UnaryOptions
	StreamOptions StreamOptions

	MetaHandlers []remote.MetaHandler

	RemoteOpt        *remote.ClientOption
	Proxy            proxy.ForwardProxy
	Resolver         discovery.Resolver
	HTTPResolver     http.Resolver
	Balancer         loadbalance.Loadbalancer
	BalancerCacheOpt *lbcache.Options
	PoolCfg          *connpool2.IdleConfig
	ErrHandle        func(context.Context, error) error
	Targets          string
	CBSuite          *circuitbreak.CBSuite
	Timeouts         rpcinfo.TimeoutProvider

	ACLRules []acl.RejectFunc

	MWBs  []endpoint.MiddlewareBuilder
	IMWBs []endpoint.MiddlewareBuilder

	Bus          event.Bus
	Events       event.Queue
	ExtraTimeout time.Duration

	// DebugInfo should only contain objects that are suitable for json serialization.
	DebugInfo    utils.Slice
	DebugService diagnosis.Service

	// Observability
	TracerCtl  *rpcinfo.TraceController
	StatsLevel *stats.Level

	CloseCallbacks []func() error
	WarmUpOption   *warmup.ClientOption

	// GRPC
	GRPCConnPoolSize uint32
	GRPCConnectOpts  *grpc.ConnectOptions

	// TTHeaderStreaming
	TTHeaderStreamingOptions TTHeaderStreamingOptions

	// XDS
	XDSEnabled          bool
	XDSRouterMiddleware endpoint.Middleware

	// Context backup
	CtxBackupHandler backup.BackupHandler

	Streaming stream.StreamingConfig // deprecated, use StreamOptions instead

	// TailOptions is used to store options that are executed after all options are applied.
	// just ignore it unless you clearly know what you are doing.
	TailOptions []Option
}

// Apply applies all options.
func (o *Options) Apply(opts []Option) {
	for _, op := range opts {
		op.F(o, &o.DebugInfo)
	}
	for _, op := range o.TailOptions {
		op.F(o, &o.DebugInfo)
	}
	o.TailOptions = nil
}

// Option is the only way to config client.
type Option struct {
	F func(o *Options, di *utils.Slice)
}

// NewOptions creates a new option.
func NewOptions(opts []Option) *Options {
	o := &Options{
		Cli:          &rpcinfo.EndpointBasicInfo{Tags: make(map[string]string)},
		Svr:          &rpcinfo.EndpointBasicInfo{Tags: make(map[string]string)},
		MetaHandlers: []remote.MetaHandler{transmeta.MetainfoClientHandler},
		RemoteOpt:    newClientRemoteOption(),
		Configs:      rpcinfo.NewRPCConfig(),
		Locks:        NewConfigLocks(),
		Once:         configutil.NewOptionOnce(),
		HTTPResolver: http.NewDefaultResolver(),
		DebugService: diagnosis.NoopService,

		Bus:    event.NewEventBus(),
		Events: event.NewQueue(event.MaxEventNum),

		TracerCtl: &rpcinfo.TraceController{},

		GRPCConnectOpts: new(grpc.ConnectOptions),
	}
	// Since Kitex v0.13.0, the default transport protocol has been changed to framed from buffered.
	rpcinfo.AsMutableRPCConfig(o.Configs).SetTransportProtocol(transport.Framed)
	o.UnaryOptions.opts = o
	o.Apply(opts)

	o.initRemoteOpt()

	if o.UnaryOptions.RetryContainer != nil && o.DebugService != nil {
		o.DebugService.RegisterProbeFunc(diagnosis.RetryPolicyKey, o.UnaryOptions.RetryContainer.Dump)
	}

	if o.StatsLevel == nil {
		level := stats.LevelDisabled
		if o.TracerCtl.HasTracer() {
			level = stats.LevelDetailed
		}
		o.StatsLevel = &level
	}
	return o
}

func (o *Options) initRemoteOpt() {
	var zero connpool2.IdleConfig

	// configure grpc
	if o.Configs.TransportProtocol()&transport.GRPC == transport.GRPC {
		if o.PoolCfg != nil && *o.PoolCfg == zero {
			// grpc unary short connection
			o.GRPCConnectOpts.ShortConn = true
		}
		o.RemoteOpt.ConnPool = nphttp2.NewConnPool(o.Svr.ServiceName, o.GRPCConnPoolSize, *o.GRPCConnectOpts)
		o.RemoteOpt.CliHandlerFactory = nphttp2.NewCliTransHandlerFactory()
	}
	// configure grpc streaming
	if o.Configs.TransportProtocol()&(transport.GRPC|transport.GRPCStreaming) == transport.GRPCStreaming {
		if o.PoolCfg != nil && *o.PoolCfg == zero {
			// grpc unary short connection
			o.GRPCConnectOpts.ShortConn = true
		}
		o.RemoteOpt.GRPCStreamingConnPool = nphttp2.NewConnPool(o.Svr.ServiceName, o.GRPCConnPoolSize, *o.GRPCConnectOpts)
		o.RemoteOpt.GRPCStreamingCliHandlerFactory = nphttp2.NewCliTransHandlerFactory()
	}
	// configure ttheader streaming
	if o.Configs.TransportProtocol()&transport.TTHeaderStreaming == transport.TTHeaderStreaming {
		if o.PoolCfg != nil && *o.PoolCfg == zero {
			// configure short conn pool
			o.TTHeaderStreamingOptions.TransportOptions = append(o.TTHeaderStreamingOptions.TransportOptions, ttstream.WithClientShortConnPool())
		}
		o.RemoteOpt.TTHeaderStreamingCliHandlerFactory = ttstream.NewCliTransHandlerFactory(o.TTHeaderStreamingOptions.TransportOptions...)
	}
	if o.RemoteOpt.ConnPool == nil {
		if o.PoolCfg != nil {
			if *o.PoolCfg == zero {
				o.RemoteOpt.ConnPool = connpool.NewShortPool(o.Svr.ServiceName)
			} else {
				o.RemoteOpt.ConnPool = connpool.NewLongPool(o.Svr.ServiceName, *o.PoolCfg)
			}
		} else {
			o.RemoteOpt.ConnPool = connpool.NewLongPool(
				o.Svr.ServiceName,
				connpool2.IdleConfig{
					MaxIdlePerAddress: 10,
					MaxIdleGlobal:     100,
					MaxIdleTimeout:    time.Minute,
				},
			)
		}
	}
}

// InitRetryContainer init retry container and add close callback
func (o *Options) InitRetryContainer() {
	if o.UnaryOptions.RetryContainer == nil {
		o.UnaryOptions.RetryContainer = retry.NewRetryContainerWithPercentageLimit()
		o.CloseCallbacks = append(o.CloseCallbacks, o.UnaryOptions.RetryContainer.Close)
	}
}
