/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 triple

import (
	"context"
	"errors"
	"reflect"
	"sync"
	"time"
)

import (
	"github.com/dubbogo/gost/log/logger"

	"github.com/dubbogo/grpc-go/metadata"

	tripleConstant "github.com/dubbogo/triple/pkg/common/constant"
	triConfig "github.com/dubbogo/triple/pkg/config"
	"github.com/dubbogo/triple/pkg/triple"

	"github.com/dustin/go-humanize"
)

import (
	"dubbo.apache.org/dubbo-go/v3/common"
	"dubbo.apache.org/dubbo-go/v3/common/constant"
	"dubbo.apache.org/dubbo-go/v3/config"
	"dubbo.apache.org/dubbo-go/v3/global"
	"dubbo.apache.org/dubbo-go/v3/protocol/base"
	"dubbo.apache.org/dubbo-go/v3/protocol/result"
	dubbotls "dubbo.apache.org/dubbo-go/v3/tls"
)

// same as dubbo_invoker.go attachmentKey
var attachmentKey = []string{
	constant.InterfaceKey, constant.GroupKey, constant.TokenKey, constant.TimeoutKey,
	constant.VersionKey, tripleConstant.TripleServiceGroup, tripleConstant.TripleServiceVersion,
}

// DubboInvoker is implement of protocol.Invoker, a dubboInvoker refer to one service and ip.
type DubboInvoker struct {
	base.BaseInvoker
	// the net layer client, it is focus on network communication.
	client *triple.TripleClient
	// quitOnce is used to make sure DubboInvoker is only destroyed once
	quitOnce sync.Once
	// timeout for service(interface) level.
	timeout time.Duration
	// clientGuard is the client lock of dubbo invoker
	clientGuard *sync.RWMutex
}

// NewDubbo3Invoker constructor
func NewDubbo3Invoker(url *common.URL) (*DubboInvoker, error) {
	// TODO: Temporary compatibility with old APIs, can be removed later
	rt := config.GetConsumerConfig().RequestTimeout
	if consumerConfRaw, ok := url.GetAttribute(constant.ConsumerConfigKey); ok {
		if consumerConf, ok := consumerConfRaw.(*global.ConsumerConfig); ok {
			rt = consumerConf.RequestTimeout
		}
	}

	timeout := url.GetParamDuration(constant.TimeoutKey, rt)
	// for triple pb serialization. The bean name from provider is the provider reference key,
	// which can't locate the target consumer stub, so we use interface key..
	interfaceKey := url.GetParam(constant.InterfaceKey, "")
	consumerService := config.GetConsumerServiceByInterfaceName(interfaceKey)

	dubboSerializerType := url.GetParam(constant.SerializationKey, constant.ProtobufSerialization)
	triCodecType := tripleConstant.CodecType(dubboSerializerType)
	// new triple client
	opts := []triConfig.OptionFunction{
		triConfig.WithClientTimeout(timeout),
		triConfig.WithCodecType(triCodecType),
		triConfig.WithLocation(url.Location),
		triConfig.WithHeaderAppVersion(url.GetParam(constant.AppVersionKey, "")),
		triConfig.WithHeaderGroup(url.GetParam(constant.GroupKey, "")),
		triConfig.WithLogger(logger.GetLogger()),
	}
	maxCallRecvMsgSize := constant.DefaultMaxCallRecvMsgSize
	if maxCall, err := humanize.ParseBytes(url.GetParam(constant.MaxCallRecvMsgSize, "")); err == nil && maxCall != 0 {
		maxCallRecvMsgSize = int(maxCall)
	}
	maxCallSendMsgSize := constant.DefaultMaxCallSendMsgSize
	if maxCall, err := humanize.ParseBytes(url.GetParam(constant.MaxCallSendMsgSize, "")); err == nil && maxCall != 0 {
		maxCallSendMsgSize = int(maxCall)
	}
	opts = append(opts, triConfig.WithGRPCMaxCallRecvMessageSize(maxCallRecvMsgSize))
	opts = append(opts, triConfig.WithGRPCMaxCallSendMessageSize(maxCallSendMsgSize))

	// grpc keepalive config
	// Deprecated：use tripleconfig
	// TODO: remove KeepAliveInterval and KeepAliveInterval in version 4.0.0
	keepAliveInterval := url.GetParamDuration(constant.KeepAliveInterval, constant.DefaultKeepAliveInterval)
	keepAliveTimeout := url.GetParamDuration(constant.KeepAliveTimeout, constant.DefaultKeepAliveTimeout)

	tripleConfRaw, ok := url.GetAttribute(constant.TripleConfigKey)
	if ok {
		tripleConf := tripleConfRaw.(*global.TripleConfig)
		// TODO: handle ParseDuration error
		keepAliveInterval, _ = time.ParseDuration(tripleConf.KeepAliveInterval)
		keepAliveTimeout, _ = time.ParseDuration(tripleConf.KeepAliveTimeout)
	}

	opts = append(opts, triConfig.WithGRPCKeepAliveTimeInterval(keepAliveInterval))
	opts = append(opts, triConfig.WithGRPCKeepAliveTimeout(keepAliveTimeout))

	tracingKey := url.GetParam(constant.TracingConfigKey, "")
	if tracingKey != "" {
		tracingConfig := config.GetTracingConfig(tracingKey)
		if tracingConfig != nil {
			if tracingConfig.Name == "jaeger" {
				if tracingConfig.ServiceName == "" {
					tracingConfig.ServiceName = config.GetApplicationConfig().Name
				}
				opts = append(opts, triConfig.WithJaegerConfig(
					tracingConfig.Address,
					tracingConfig.ServiceName,
					*tracingConfig.UseAgent,
				))
			} else {
				logger.Warnf("unsupported tracing name %s, now triple only support jaeger", tracingConfig.Name)
			}
		}
	}

	triOption := triConfig.NewTripleOption(opts...)

	// TODO: remove config TLSConfig
	// delete this branch
	tlsConfig := config.GetRootConfig().TLSConfig
	if tlsConfig != nil {
		triOption.CACertFile = tlsConfig.CACertFile
		triOption.TLSCertFile = tlsConfig.TLSCertFile
		triOption.TLSKeyFile = tlsConfig.TLSKeyFile
		triOption.TLSServerName = tlsConfig.TLSServerName
		logger.Infof("DUBBO3 Client initialized the TLSConfig configuration")
	} else if tlsConfRaw, ok := url.GetAttribute(constant.TLSConfigKey); ok {
		// use global TLSConfig handle tls
		tlsConf, ok := tlsConfRaw.(*global.TLSConfig)
		if !ok {
			return nil, errors.New("DUBBO3 Client initialized the TLSConfig configuration failed")
		}
		if dubbotls.IsClientTLSValid(tlsConf) {
			triOption.CACertFile = tlsConf.CACertFile
			triOption.TLSCertFile = tlsConf.TLSCertFile
			triOption.TLSKeyFile = tlsConf.TLSKeyFile
			triOption.TLSServerName = tlsConf.TLSServerName
			logger.Infof("DUBBO3 Client initialized the TLSConfig configuration")
		}
	}

	client, err := triple.NewTripleClient(consumerService, triOption)

	if err != nil {
		return nil, err
	}

	return &DubboInvoker{
		BaseInvoker: *base.NewBaseInvoker(url),
		client:      client,
		timeout:     timeout,
		clientGuard: &sync.RWMutex{},
	}, nil
}

func (di *DubboInvoker) setClient(client *triple.TripleClient) {
	di.clientGuard.Lock()
	defer di.clientGuard.Unlock()

	di.client = client
}

func (di *DubboInvoker) getClient() *triple.TripleClient {
	di.clientGuard.RLock()
	defer di.clientGuard.RUnlock()

	return di.client
}

// Invoke call remoting.
func (di *DubboInvoker) Invoke(ctx context.Context, invocation base.Invocation) result.Result {
	var (
		result result.RPCResult
	)

	if !di.BaseInvoker.IsAvailable() {
		// Generally, the case will not happen, because the invoker has been removed
		// from the invoker list before destroy,so no new request will enter the destroyed invoker
		logger.Warnf("this dubboInvoker is destroyed")
		result.SetError(base.ErrDestroyedInvoker)
		return &result
	}

	client := di.getClient()
	if client == nil {
		result.SetError(base.ErrClientClosed)
		return &result
	}

	for _, k := range attachmentKey {
		var paramKey string
		switch k {
		case tripleConstant.TripleServiceGroup:
			paramKey = constant.GroupKey
		case tripleConstant.TripleServiceVersion:
			paramKey = constant.VersionKey
		default:
			paramKey = k
		}

		if v := di.GetURL().GetParam(paramKey, ""); len(v) > 0 {
			invocation.SetAttachment(k, v)
		}
	}

	// append interface id to ctx
	gRPCMD := make(metadata.MD, 0)
	// triple will convert attachment value to []string
	for k, v := range invocation.Attachments() {
		if str, ok := v.(string); ok {
			gRPCMD.Set(k, str)
			continue
		}
		if str, ok := v.([]string); ok {
			gRPCMD.Set(k, str...)
			continue
		}
		logger.Warnf("[Triple Protocol]Triple attachment value with key = %s is invalid, which should be string or []string", k)
	}
	ctx = metadata.NewOutgoingContext(ctx, gRPCMD)
	ctx = context.WithValue(ctx, tripleConstant.InterfaceKey, di.BaseInvoker.GetURL().GetParam(constant.InterfaceKey, ""))
	in := make([]reflect.Value, 0, 16)
	in = append(in, reflect.ValueOf(ctx))

	if len(invocation.ParameterValues()) > 0 {
		in = append(in, invocation.ParameterValues()...)
	}

	methodName := invocation.MethodName()
	triAttachmentWithErr := client.Invoke(methodName, in, invocation.Reply())
	result.SetError(triAttachmentWithErr.GetError())
	result.SetAttachments(make(map[string]any))
	for k, v := range triAttachmentWithErr.GetAttachments() {
		result.AddAttachment(k, v)
	}
	result.SetResult(invocation.Reply())
	return &result
}

// IsAvailable check if invoker is available, now it is useless
func (di *DubboInvoker) IsAvailable() bool {
	client := di.getClient()
	if client != nil {
		// FIXME: here can't check if tcp server is started now!!!
		return client.IsAvailable()
	}
	return false
}

// Destroy destroy dubbo3 client invoker.
func (di *DubboInvoker) Destroy() {
	di.quitOnce.Do(func() {
		di.BaseInvoker.Destroy()
		client := di.getClient()
		if client != nil {
			di.setClient(nil)
			client.Close()
		}
	})
}
