/*
 * Copyright 2024 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.
 */

// Service inline is a service deployment form of ByteDance's internal applications.
// Different Kitex services are merged together during the compilation period through the tool chain, and this capability is not yet opensource.
package server

import (
	"context"
	"reflect"
	"unsafe"

	"github.com/bytedance/gopkg/cloud/metainfo"

	"github.com/cloudwego/kitex/pkg/consts"
	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/utils"
)

var localAddr = utils.NewNetAddr("tcp", "127.0.0.1")

var invocationType = reflect.TypeOf(rpcinfo.NewServerInvocation()).Elem()

func constructServerCtxWithMetadata(cliCtx context.Context) (serverCtx context.Context) {
	serverCtx = context.Background()
	customVal := cliCtx.Value(consts.SERVICE_INLINE_CUSTOM_CTX_KEY)
	if customVal != nil {
		serverCtx = context.WithValue(serverCtx, consts.SERVICE_INLINE_CUSTOM_CTX_KEY, customVal)
	}
	// metainfo
	// forward transmission
	kvs := make(map[string]string, 16)
	metainfo.SaveMetaInfoToMap(cliCtx, kvs)
	if len(kvs) > 0 {
		serverCtx = metainfo.SetMetaInfoFromMap(serverCtx, kvs)
	}
	serverCtx = metainfo.TransferForward(serverCtx)
	// reverse transmission, backward mark
	serverCtx = metainfo.WithBackwardValuesToSend(serverCtx)
	return serverCtx
}

func (s *server) constructServerRPCInfo(svrCtx context.Context, cliRPCInfo rpcinfo.RPCInfo) (newServerCtx context.Context, svrRPCInfo rpcinfo.RPCInfo) {
	rpcStats := rpcinfo.AsMutableRPCStats(rpcinfo.NewRPCStats())
	if s.opt.StatsLevel != nil {
		rpcStats.SetLevel(*s.opt.StatsLevel)
	}
	// Export read-only views to external users and keep a mapping for internal users.
	svrRPCInfo = rpcinfo.NewRPCInfo(
		rpcinfo.EmptyEndpointInfo(),
		rpcinfo.FromBasicInfo(s.opt.Svr),
		rpcinfo.NewServerInvocation(),
		rpcinfo.AsMutableRPCConfig(s.opt.Configs).Clone().ImmutableView(),
		rpcStats.ImmutableView(),
	)
	rpcinfo.AsMutableEndpointInfo(svrRPCInfo.From()).SetAddress(localAddr)
	svrCtx = rpcinfo.NewCtxWithRPCInfo(svrCtx, svrRPCInfo)

	// handle common rpcinfo
	method := cliRPCInfo.To().Method()
	if ink, ok := svrRPCInfo.Invocation().(rpcinfo.InvocationSetter); ok {
		ink.SetMethodName(method)
		serviceName := cliRPCInfo.Invocation().Extra(consts.SERVICE_INLINE_SERVICE_NAME).(string)
		ink.SetServiceName(serviceName)
		svc := s.svcs.getService(serviceName)
		svcInfo := svc.svcInfo
		ink.SetMethodInfo(svcInfo.MethodInfo(context.Background(), method))
	}
	rpcinfo.AsMutableEndpointInfo(svrRPCInfo.To()).SetMethod(method)
	svrCtx = context.WithValue(svrCtx, consts.CtxKeyMethod, method)
	return svrCtx, svrRPCInfo
}

func (s *server) BuildServiceInlineInvokeChain() endpoint.Endpoint {
	// service inline will not call server.Run, so we should init server here
	s.init()

	svrTraceCtl := s.opt.TracerCtl
	if svrTraceCtl == nil {
		svrTraceCtl = &rpcinfo.TraceController{}
	}

	innerHandlerEp := s.invokeHandleEndpoint()
	mw := func(next endpoint.Endpoint) endpoint.Endpoint {
		return func(ctx context.Context, req, resp interface{}) (err error) {
			serverCtx := constructServerCtxWithMetadata(ctx)
			defer func() {
				// backward key
				kvs := metainfo.AllBackwardValuesToSend(serverCtx)
				if len(kvs) > 0 {
					metainfo.SetBackwardValuesFromMap(ctx, kvs)
				}
			}()
			ptr := ctx.Value(consts.SERVICE_INLINE_RPCINFO_KEY).(unsafe.Pointer)
			cliRPCInfo := *(*rpcinfo.RPCInfo)(ptr)
			serverCtx, svrRPCInfo := s.constructServerRPCInfo(serverCtx, cliRPCInfo)
			defer func() {
				rpcinfo.PutRPCInfo(svrRPCInfo)
			}()

			// server trace
			serverCtx = svrTraceCtl.DoStart(serverCtx, svrRPCInfo)
			serverCtx = context.WithValue(serverCtx, consts.SERVICE_INLINE_DATA_KEY, ctx.Value(consts.SERVICE_INLINE_DATA_KEY))
			err = next(serverCtx, req, resp)

			bizErr := svrRPCInfo.Invocation().BizStatusErr()
			if bizErr != nil {
				// In service inline scenario, cliRPCInfo is upstream rpcinfo and the dep is isolated.
				// We forcibly converted it to downstream rpcinfo through unsafe.Pointer,
				// so the _type in iface still points to the upstream type, which makes
				// cliRPCInfo.Invocation().(rpcinfo.InvocationSetter) cannot be asserted.
				//
				// To solve this problem, we need init a new invocation from reflect.NewAt to
				// pass the type assertion
				invI := reflect.NewAt(invocationType, unsafe.Pointer(reflect.ValueOf(cliRPCInfo.Invocation()).Pointer())).Interface()
				if cliSetter, ok := invI.(rpcinfo.InvocationSetter); ok {
					cliSetter.SetBizStatusErr(bizErr)
				}
			}
			// finish server trace
			// contextServiceInlineHandler may convert nil err to non nil err, so handle trace here
			svrTraceCtl.DoFinish(serverCtx, svrRPCInfo, err)

			return
		}
	}

	// In the internal service inline scenario, if the merged service does not detect the RPC Ingress Mesh being enabled
	// (for example, if the main service is an API service, the RPC Ingress Mesh will not be enabled),
	// the CtxEventQueueKey will be used to monitor configuration changes for debugging during server initialization.
	// If this key is not injected, it will lead to a panic during initialization.
	ctx := fillContext(s.opt)
	mws := []endpoint.Middleware{mw}
	smws := s.buildMiddlewares(ctx)
	mws = append(mws, smws...)
	return endpoint.Chain(mws...)(endpoint.Endpoint(innerHandlerEp))
}
