client.go

package interceptors

import (
	"context"
	stdError "errors"
	"fmt"
	"slices"

	"github.com/afex/hystrix-go/hystrix"
	"github.com/go-coldbrew/errors"
	"github.com/go-coldbrew/errors/notifier"
	"github.com/go-coldbrew/log"
	nrutil "github.com/go-coldbrew/tracing/newrelic"
	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/retry"
	"github.com/newrelic/go-agent/v3/integrations/nrgrpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/status"
)

// DefaultClientInterceptors are the set of default interceptors that should be applied to all client calls
func DefaultClientInterceptors(defaultOpts ...any) []grpc.UnaryClientInterceptor {
	ints := []grpc.UnaryClientInterceptor{}
	if len(defaultConfig.unaryClientInterceptors) > 0 {
		ints = append(ints, defaultConfig.unaryClientInterceptors...)
	}
	if defaultConfig.useCBClientInterceptors {
		callOptions := make([]grpc.CallOption, 0)
		for _, opt := range defaultOpts {
			if opt == nil {
				continue
			}
			if o, ok := opt.(grpc.CallOption); ok {
				callOptions = append(callOptions, o)
			}
		}
		ints = append(ints, ExecutorClientInterceptor(callOptions...))
		ints = append(ints,
			grpc_retry.UnaryClientInterceptor(),
			NewRelicClientInterceptor(),
			getClientMetrics().UnaryClientInterceptor(),
		)
	}
	return ints
}

// DefaultClientStreamInterceptors are the set of default interceptors that should be applied to all stream client calls
func DefaultClientStreamInterceptors(defaultOpts ...any) []grpc.StreamClientInterceptor {
	ints := []grpc.StreamClientInterceptor{}
	if len(defaultConfig.streamClientInterceptors) > 0 {
		ints = append(ints, defaultConfig.streamClientInterceptors...)
	}
	if defaultConfig.useCBClientInterceptors {
		if nrutil.GetNewRelicApp() != nil {
			ints = append(ints, nrgrpc.StreamClientInterceptor)
		}
		ints = append(ints, getClientMetrics().StreamClientInterceptor())
	}
	return ints
}

// DefaultClientInterceptor are the set of default interceptors that should be applied to all client calls
func DefaultClientInterceptor(defaultOpts ...any) grpc.UnaryClientInterceptor {
	return chainUnaryClient(DefaultClientInterceptors(defaultOpts...))
}

// DefaultClientStreamInterceptor are the set of default interceptors that should be applied to all stream client calls
func DefaultClientStreamInterceptor(defaultOpts ...any) grpc.StreamClientInterceptor {
	return chainStreamClient(DefaultClientStreamInterceptors(defaultOpts...))
}

// NewRelicClientInterceptor intercepts all client actions and reports them to newrelic.
// When NewRelic app is nil (no license key configured), returns a pass-through
// interceptor to avoid overhead.
func NewRelicClientInterceptor() grpc.UnaryClientInterceptor {
	app := nrutil.GetNewRelicApp()
	if app == nil {
		return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
			return invoker(ctx, method, req, reply, cc, opts...)
		}
	}
	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		if defaultConfig.filterFunc(ctx, method) {
			return nrgrpc.UnaryClientInterceptor(ctx, method, req, reply, cc, invoker, opts...)
		} else {
			return invoker(ctx, method, req, reply, cc, opts...)
		}
	}
}

// Deprecated: GRPCClientInterceptor is no longer needed. gRPC tracing is now handled
// by google.golang.org/grpc/stats/opentelemetry, configured via opentelemetry.DialOption()
// at the client level. This function is retained for backwards compatibility but returns
// a no-op interceptor.
func GRPCClientInterceptor(_ ...any) grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		return invoker(ctx, method, req, reply, cc, opts...)
	}
}

// ExecutorClientInterceptor returns a unary client interceptor that wraps each
// RPC in an [Executor]. The executor provides resilience logic such as circuit
// breaking, retries, or bulkheading.
//
// Executor resolution order: per-call [WithExecutor] > global [SetDefaultExecutor].
// When no executor is configured, the interceptor falls back to
// [HystrixClientInterceptor] for backward compatibility. When the caller
// explicitly opts out via [WithoutExecutor] or [WithoutHystrix], the RPC is
// invoked directly as a passthrough.
//
// Excluded errors and codes (set via [WithExcludedErrors] / [WithExcludedCodes])
// are reported as nil to the executor, preventing them from tripping circuit
// breakers or retry logic. The original error is still returned to the caller.
func ExecutorClientInterceptor(defaultOpts ...grpc.CallOption) grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		o := clientOptions{}
		for _, opt := range defaultOpts {
			if opt != nil {
				if co, ok := opt.(clientOption); ok {
					co.process(&o)
				}
			}
		}
		for _, opt := range opts {
			if opt != nil {
				if co, ok := opt.(clientOption); ok {
					co.process(&o)
				}
			}
		}

		// Resolve executor: per-call > global > hystrix fallback
		exec := defaultConfig.defaultExecutor
		if o.hasExecutor {
			exec = o.executor
		}
		if exec == nil {
			if o.hasExecutor {
				// Caller explicitly opted out (WithoutExecutor / WithoutHystrix).
				return invoker(ctx, method, req, reply, cc, opts...)
			}
			// No executor configured; fall back to Hystrix for backward compat.
			return hystrixFallback(ctx, method, req, reply, cc, invoker, opts, o)
		}

		var invokerErr error
		executorErr := exec(ctx, method, func(execCtx context.Context) (err error) {
			defer func() {
				if r := recover(); r != nil {
					err = errors.Wrap(fmt.Errorf("panic in executor method: %s", method), "Executor")
					log.Error(execCtx, "panic", r, "method", method)
				}
			}()
			defer notifier.NotifyOnPanic(execCtx, method)
			invokerErr = invoker(execCtx, method, req, reply, cc, opts...)
			for _, excludedErr := range o.excludedErrors {
				if stdError.Is(invokerErr, excludedErr) {
					return nil
				}
			}
			if st, ok := status.FromError(invokerErr); ok {
				if slices.Contains(o.excludedCodes, st.Code()) {
					return nil
				}
			}
			return invokerErr
		})
		if invokerErr != nil {
			return invokerErr
		}
		return executorErr
	}
}

// hystrixFallback runs the RPC through the deprecated Hystrix circuit breaker.
// It is the shared Hystrix implementation used both as the executor fallback
// when no executor is configured (neither global nor per-call) and by
// [HystrixClientInterceptor], preserving backward compatibility for services
// that have not migrated.
func hystrixFallback(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts []grpc.CallOption, o clientOptions) error {
	if o.disableHystrix {
		return invoker(ctx, method, req, reply, cc, opts...)
	}
	hystrixCmd := o.hystrixName
	if hystrixCmd == "" {
		hystrixCmd = method
	}
	newCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	var invokerErr error
	hystrixErr := hystrix.Do(hystrixCmd, func() (err error) {
		defer func() {
			if r := recover(); r != nil {
				err = errors.Wrap(fmt.Errorf("panic inside hystrix method: %s, req: %v, reply: %v", method, req, reply), "Hystrix")
				log.Error(ctx, "panic", r, "method", method, "req", req, "reply", reply)
			}
		}()
		defer notifier.NotifyOnPanic(newCtx, method)
		invokerErr = invoker(newCtx, method, req, reply, cc, opts...)
		for _, excludedErr := range o.excludedErrors {
			if stdError.Is(invokerErr, excludedErr) {
				return nil
			}
		}
		if st, ok := status.FromError(invokerErr); ok {
			if slices.Contains(o.excludedCodes, st.Code()) {
				return nil
			}
		}
		return invokerErr
	}, nil)
	if invokerErr != nil {
		return invokerErr
	}
	return hystrixErr
}

// Deprecated: HystrixClientInterceptor wraps the unmaintained hystrix-go library.
// Use [SetDefaultExecutor] with a failsafe-go executor instead. Will be removed in v1.
//
// See [ExecutorClientInterceptor] for the replacement.
func HystrixClientInterceptor(defaultOpts ...grpc.CallOption) grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		o := clientOptions{}
		for _, opt := range defaultOpts {
			if opt != nil {
				if co, ok := opt.(clientOption); ok {
					co.process(&o)
				}
			}
		}
		for _, opt := range opts {
			if opt != nil {
				if co, ok := opt.(clientOption); ok {
					co.process(&o)
				}
			}
		}
		return hystrixFallback(ctx, method, req, reply, cc, invoker, opts, o)
	}
}