/*
 * 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 server defines the Options of server
package server

import (
	"context"
	"os"
	"os/signal"
	"syscall"

	"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/diagnosis"
	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/endpoint/sep"
	"github.com/cloudwego/kitex/pkg/event"
	"github.com/cloudwego/kitex/pkg/gofunc"
	"github.com/cloudwego/kitex/pkg/limit"
	"github.com/cloudwego/kitex/pkg/limiter"
	"github.com/cloudwego/kitex/pkg/proxy"
	"github.com/cloudwego/kitex/pkg/registry"
	"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/rpcinfo"
	"github.com/cloudwego/kitex/pkg/serviceinfo"
	"github.com/cloudwego/kitex/pkg/stats"
	"github.com/cloudwego/kitex/pkg/streaming"
	"github.com/cloudwego/kitex/pkg/transmeta"
	"github.com/cloudwego/kitex/pkg/utils"
)

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 {
	UnaryMiddlewares        []endpoint.UnaryMiddleware
	UnaryMiddlewareBuilders []endpoint.UnaryMiddlewareBuilder
}

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...)
	}
}

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

type StreamOptions struct {
	EventHandler                 streaming.EventHandler
	StreamMiddlewares            []sep.StreamMiddleware
	StreamMiddlewareBuilders     []sep.StreamMiddlewareBuilder
	StreamRecvMiddlewares        []sep.StreamRecvMiddleware
	StreamRecvMiddlewareBuilders []sep.StreamRecvMiddlewareBuilder
	StreamSendMiddlewares        []sep.StreamSendMiddleware
	StreamSendMiddlewareBuilders []sep.StreamSendMiddlewareBuilder
}

func (o *StreamOptions) InitMiddlewares(ctx context.Context) {
	if len(o.StreamMiddlewareBuilders) > 0 {
		streamMiddlewares := make([]sep.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([]sep.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([]sep.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() sep.StreamRecvEndpoint {
	return sep.StreamRecvChain(o.StreamRecvMiddlewares...)(func(ctx context.Context, stream streaming.ServerStream, message interface{}) (err error) {
		return stream.RecvMsg(ctx, message)
	})
}

func (o *StreamOptions) BuildSendChain() sep.StreamSendEndpoint {
	return sep.StreamSendChain(o.StreamSendMiddlewares...)(func(ctx context.Context, stream streaming.ServerStream, message interface{}) (err error) {
		return stream.SendMsg(ctx, message)
	})
}

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

// Options is used to initialize the server.
type Options struct {
	Svr      *rpcinfo.EndpointBasicInfo
	Configs  rpcinfo.RPCConfig
	LockBits int
	Once     *configutil.OptionOnce

	UnaryOptions  UnaryOptions
	StreamOptions StreamOptions

	MetaHandlers []remote.MetaHandler

	RemoteOpt  *remote.ServerOption
	ErrHandle  func(context.Context, error) error
	ExitSignal func() <-chan error
	Proxy      proxy.ReverseProxy

	// Registry is used for service registry.
	Registry registry.Registry
	// RegistryInfo is used to in registry.
	RegistryInfo *registry.Info

	ACLRules []acl.RejectFunc
	Limit    Limit

	MWBs []endpoint.MiddlewareBuilder

	Bus    event.Bus
	Events event.Queue

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

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

	BackupOpt backup.Options

	Streaming stream.StreamingConfig // deprecated, use StreamOptions instead

	RefuseTrafficWithoutServiceName bool
	EnableContextTimeout            bool
}

type Limit struct {
	Limits        *limit.Option
	LimitReporter limiter.LimitReporter
	ConLimit      limiter.ConcurrencyLimiter
	QPSLimit      limiter.RateLimiter

	// QPSLimitPostDecode is true to indicate that the QPS limiter takes effect in
	// the OnMessage callback, and false for the OnRead callback.
	// Usually when the server is multiplexed, Kitex set it to True by default.
	QPSLimitPostDecode bool
}

// NewOptions creates a default options.
func NewOptions(opts []Option) *Options {
	ropt := newServerRemoteOption()
	o := &Options{
		Svr:          &rpcinfo.EndpointBasicInfo{},
		Configs:      rpcinfo.NewRPCConfig(),
		Once:         configutil.NewOptionOnce(),
		MetaHandlers: []remote.MetaHandler{transmeta.MetainfoServerHandler},
		RemoteOpt:    ropt,
		DebugService: diagnosis.NoopService,
		ExitSignal:   DefaultSysExitSignal,

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

		TracerCtl: &rpcinfo.TraceController{},
		Registry:  registry.NoopRegistry,
	}
	ApplyOptions(opts, o)

	rpcinfo.AsMutableRPCConfig(o.Configs).LockConfig(o.LockBits)
	if o.StatsLevel == nil {
		level := stats.LevelDisabled
		if o.TracerCtl.HasTracer() {
			level = stats.LevelDetailed
		}
		o.StatsLevel = &level
	}
	return o
}

// ApplyOptions applies the given options.
func ApplyOptions(opts []Option, o *Options) {
	for _, op := range opts {
		op.F(o, &o.DebugInfo)
	}
}

func DefaultSysExitSignal() <-chan error {
	errCh := make(chan error, 1)
	gofunc.GoFunc(context.Background(), func() {
		sig := SysExitSignal()
		defer signal.Stop(sig)
		<-sig
		errCh <- nil
	})
	return errCh
}

func SysExitSignal() chan os.Signal {
	signals := make(chan os.Signal, 1)
	notifications := []os.Signal{syscall.SIGINT, syscall.SIGTERM}
	if !signal.Ignored(syscall.SIGHUP) {
		notifications = append(notifications, syscall.SIGHUP)
	}
	signal.Notify(signals, notifications...)
	return signals
}
