wallets listing dedupe

api/billing/fees/internal/service/fees/service.go

@@ -5,7 +5,6 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"strings"
 	"time"
 
@@ -33,6 +32,8 @@ import (
 )
 
 type Service struct {
+	feesv1.UnimplementedFeeEngineServer
+
 	logger     mlogger.Logger
 	storage    storage.Repository
 	producer   msg.Producer
@@ -42,7 +43,6 @@ type Service struct {
 	resolver   FeeResolver
 	announcer  *discovery.Announcer
 	invokeURI  string
-	feesv1.UnimplementedFeeEngineServer
 }
 
 func NewService(logger mlogger.Logger, repo storage.Repository, producer msg.Producer, opts ...Option) *Service {
@@ -52,6 +52,7 @@ func NewService(logger mlogger.Logger, repo storage.Repository, producer msg.Pro
 		producer: producer,
 		clock:    clockpkg.NewSystem(),
 	}
+
 	initMetrics()
 
 	for _, opt := range opts {
@@ -61,9 +62,11 @@ func NewService(logger mlogger.Logger, repo storage.Repository, producer msg.Pro
 	if svc.clock == nil {
 		svc.clock = clockpkg.NewSystem()
 	}
+
 	if svc.calculator == nil {
 		svc.calculator = internalcalculator.New(svc.logger, svc.oracle)
 	}
+
 	if svc.resolver == nil {
 		svc.resolver = resolver.New(repo.Plans(), svc.logger)
 	}
@@ -83,25 +86,12 @@ func (s *Service) Shutdown() {
 	if s == nil {
 		return
 	}
+
 	if s.announcer != nil {
 		s.announcer.Stop()
 	}
 }
 
-func (s *Service) startDiscoveryAnnouncer() {
-	if s == nil || s.producer == nil {
-		return
-	}
-	announce := discovery.Announcement{
-		Service:    "BILLING_FEES",
-		Operations: []string{"fee.calc"},
-		InvokeURI:  s.invokeURI,
-		Version:    appversion.Create().Short(),
-	}
-	s.announcer = discovery.NewAnnouncer(s.logger, s.producer, string(mservice.FeePlans), announce)
-	s.announcer.Start()
-}
-
 func (s *Service) QuoteFees(ctx context.Context, req *feesv1.QuoteFeesRequest) (resp *feesv1.QuoteFeesResponse, err error) {
 	var (
 		meta   *feesv1.RequestMeta
@@ -111,23 +101,29 @@ func (s *Service) QuoteFees(ctx context.Context, req *feesv1.QuoteFeesRequest) (
 		meta = req.GetMeta()
 		intent = req.GetIntent()
 	}
+
 	logger := s.logger.With(requestLogFields(meta, intent)...)
 
 	start := s.clock.Now()
+
 	trigger := feesv1.Trigger_TRIGGER_UNSPECIFIED
 	if intent != nil {
 		trigger = intent.GetTrigger()
 	}
+
 	var fxUsed bool
+
 	defer func() {
 		statusLabel := statusFromError(err)
 		linesCount := 0
 		appliedCount := 0
+
 		if err == nil && resp != nil {
 			fxUsed = resp.GetFxUsed() != nil
 			linesCount = len(resp.GetLines())
 			appliedCount = len(resp.GetApplied())
 		}
+
 		observeMetrics("quote", trigger, statusLabel, fxUsed, time.Since(start))
 
 		logFields := []zap.Field{
@@ -140,8 +136,10 @@ func (s *Service) QuoteFees(ctx context.Context, req *feesv1.QuoteFeesRequest) (
 		}
 		if err != nil {
 			logger.Warn("QuoteFees finished", append(logFields, zap.Error(err))...)
+
 			return
 		}
+
 		logger.Info("QuoteFees finished", logFields...)
 	}()
 
@@ -155,12 +153,14 @@ func (s *Service) QuoteFees(ctx context.Context, req *feesv1.QuoteFeesRequest) (
 	if parseErr != nil {
 		logger.Warn("QuoteFees invalid organization_ref", zap.Error(parseErr))
 		err = status.Error(codes.InvalidArgument, "invalid organization_ref")
+
 		return nil, err
 	}
 
-	lines, applied, fx, computeErr := s.computeQuote(ctx, orgRef, req.GetIntent(), req.GetPolicy(), req.GetMeta().GetTrace())
+	lines, applied, fxResult, computeErr := s.computeQuote(ctx, orgRef, req.GetIntent(), req.GetPolicy(), req.GetMeta().GetTrace())
 	if computeErr != nil {
 		err = computeErr
+
 		return nil, err
 	}
 
@@ -168,8 +168,9 @@ func (s *Service) QuoteFees(ctx context.Context, req *feesv1.QuoteFeesRequest) (
 		Meta:    &feesv1.ResponseMeta{Trace: req.GetMeta().GetTrace()},
 		Lines:   lines,
 		Applied: applied,
-		FxUsed:  fx,
+		FxUsed:  fxResult,
 	}
+
 	return resp, nil
 }
 
@@ -182,48 +183,17 @@ func (s *Service) PrecomputeFees(ctx context.Context, req *feesv1.PrecomputeFees
 		meta = req.GetMeta()
 		intent = req.GetIntent()
 	}
+
 	logger := s.logger.With(requestLogFields(meta, intent)...)
 
 	start := s.clock.Now()
+
 	trigger := feesv1.Trigger_TRIGGER_UNSPECIFIED
 	if intent != nil {
 		trigger = intent.GetTrigger()
 	}
-	var (
-		fxUsed    bool
-		expiresAt time.Time
-	)
-	defer func() {
-		statusLabel := statusFromError(err)
-		linesCount := 0
-		appliedCount := 0
-		if err == nil && resp != nil {
-			fxUsed = resp.GetFxUsed() != nil
-			linesCount = len(resp.GetLines())
-			appliedCount = len(resp.GetApplied())
-			if ts := resp.GetExpiresAt(); ts != nil {
-				expiresAt = ts.AsTime()
-			}
-		}
-		observeMetrics("precompute", trigger, statusLabel, fxUsed, time.Since(start))
 
-		logFields := []zap.Field{
-			zap.String("status", statusLabel),
-			zap.Duration("duration", time.Since(start)),
-			zap.Bool("fx_used", fxUsed),
-			zap.String("trigger", trigger.String()),
-			zap.Int("lines", linesCount),
-			zap.Int("applied_rules", appliedCount),
-		}
-		if !expiresAt.IsZero() {
-			logFields = append(logFields, zap.Time("expires_at", expiresAt))
-		}
-		if err != nil {
-			logger.Warn("PrecomputeFees finished", append(logFields, zap.Error(err))...)
-			return
-		}
-		logger.Info("PrecomputeFees finished", logFields...)
-	}()
+	defer func() { s.observePrecomputeFees(logger, err, resp, trigger, start) }()
 
 	logger.Debug("PrecomputeFees request received")
 
@@ -237,12 +207,14 @@ func (s *Service) PrecomputeFees(ctx context.Context, req *feesv1.PrecomputeFees
 	if parseErr != nil {
 		logger.Warn("PrecomputeFees invalid organization_ref", zap.Error(parseErr))
 		err = status.Error(codes.InvalidArgument, "invalid organization_ref")
+
 		return nil, err
 	}
 
-	lines, applied, fx, computeErr := s.computeQuoteWithTime(ctx, orgRef, req.GetIntent(), nil, req.GetMeta().GetTrace(), now)
+	lines, applied, fxResult, computeErr := s.computeQuoteWithTime(ctx, orgRef, req.GetIntent(), nil, req.GetMeta().GetTrace(), now)
 	if computeErr != nil {
 		err = computeErr
+
 		return nil, err
 	}
 
@@ -250,7 +222,8 @@ func (s *Service) PrecomputeFees(ctx context.Context, req *feesv1.PrecomputeFees
 	if ttl <= 0 {
 		ttl = 60000
 	}
-	expiresAt = now.Add(time.Duration(ttl) * time.Millisecond)
+
+	expiresAt := now.Add(time.Duration(ttl) * time.Millisecond)
 
 	payload := feeQuoteTokenPayload{
 		OrganizationRef: req.GetMeta().GetOrganizationRef(),
@@ -263,6 +236,7 @@ func (s *Service) PrecomputeFees(ctx context.Context, req *feesv1.PrecomputeFees
 	if token, err = encodeTokenPayload(payload); err != nil {
 		logger.Warn("Failed to encode fee quote token", zap.Error(err))
 		err = status.Error(codes.Internal, "failed to encode fee quote token")
+
 		return nil, err
 	}
 
@@ -272,8 +246,9 @@ func (s *Service) PrecomputeFees(ctx context.Context, req *feesv1.PrecomputeFees
 		ExpiresAt:     timestamppb.New(expiresAt),
 		Lines:         lines,
 		Applied:       applied,
-		FxUsed:        fx,
+		FxUsed:        fxResult,
 	}
+
 	return resp, nil
 }
 
@@ -282,49 +257,23 @@ func (s *Service) ValidateFeeToken(ctx context.Context, req *feesv1.ValidateFeeT
 	if req != nil {
 		tokenLen = len(strings.TrimSpace(req.GetFeeQuoteToken()))
 	}
+
 	logger := s.logger.With(zap.Int("token_length", tokenLen))
 
 	start := s.clock.Now()
-	trigger := feesv1.Trigger_TRIGGER_UNSPECIFIED
-	var (
-		fxUsed       bool
-		resultReason string
-	)
-	defer func() {
-		statusLabel := statusFromError(err)
-		if err == nil && resp != nil {
-			if !resp.GetValid() {
-				statusLabel = "invalid"
-			}
-			fxUsed = resp.GetFxUsed() != nil
-			if resp.GetIntent() != nil {
-				trigger = resp.GetIntent().GetTrigger()
-			}
-		}
-		observeMetrics("validate", trigger, statusLabel, fxUsed, time.Since(start))
 
-		logFields := []zap.Field{
-			zap.String("status", statusLabel),
-			zap.Duration("duration", time.Since(start)),
-			zap.Bool("fx_used", fxUsed),
-			zap.String("trigger", trigger.String()),
-			zap.Bool("valid", resp != nil && resp.GetValid()),
-		}
-		if resultReason != "" {
-			logFields = append(logFields, zap.String("reason", resultReason))
-		}
-		if err != nil {
-			logger.Warn("ValidateFeeToken finished", append(logFields, zap.Error(err))...)
-			return
-		}
-		logger.Info("ValidateFeeToken finished", logFields...)
-	}()
+	trigger := feesv1.Trigger_TRIGGER_UNSPECIFIED
+
+	var resultReason string
+
+	defer func() { s.observeValidateFeeToken(logger, err, resp, trigger, resultReason, start) }()
 
 	logger.Debug("ValidateFeeToken request received")
 
 	if req == nil || strings.TrimSpace(req.GetFeeQuoteToken()) == "" {
 		resultReason = "missing_token"
 		err = status.Error(codes.InvalidArgument, "fee_quote_token is required")
+
 		return nil, err
 	}
 
@@ -333,8 +282,11 @@ func (s *Service) ValidateFeeToken(ctx context.Context, req *feesv1.ValidateFeeT
 	payload, decodeErr := decodeTokenPayload(req.GetFeeQuoteToken())
 	if decodeErr != nil {
 		resultReason = "invalid_token"
+
 		logger.Warn("Failed to decode fee quote token", zap.Error(decodeErr))
+
 		resp = &feesv1.ValidateFeeTokenResponse{Meta: &feesv1.ResponseMeta{}, Valid: false, Reason: "invalid_token"}
+
 		return resp, nil
 	}
 
@@ -346,22 +298,29 @@ func (s *Service) ValidateFeeToken(ctx context.Context, req *feesv1.ValidateFeeT
 
 	if now.UnixMilli() > payload.ExpiresAtUnixMs {
 		resultReason = "expired"
+
 		logger.Info("Fee quote token expired")
+
 		resp = &feesv1.ValidateFeeTokenResponse{Meta: &feesv1.ResponseMeta{}, Valid: false, Reason: "expired"}
+
 		return resp, nil
 	}
 
 	orgRef, parseErr := bson.ObjectIDFromHex(payload.OrganizationRef)
 	if parseErr != nil {
 		resultReason = "invalid_token"
+
 		logger.Warn("Token contained invalid organization reference", zap.Error(parseErr))
+
 		resp = &feesv1.ValidateFeeTokenResponse{Meta: &feesv1.ResponseMeta{}, Valid: false, Reason: "invalid_token"}
+
 		return resp, nil
 	}
 
-	lines, applied, fx, computeErr := s.computeQuoteWithTime(ctx, orgRef, payload.Intent, nil, payload.Trace, now)
+	lines, applied, fxResult, computeErr := s.computeQuoteWithTime(ctx, orgRef, payload.Intent, nil, payload.Trace, now)
 	if computeErr != nil {
 		err = computeErr
+
 		return nil, err
 	}
 
@@ -371,8 +330,9 @@ func (s *Service) ValidateFeeToken(ctx context.Context, req *feesv1.ValidateFeeT
 		Intent:  payload.Intent,
 		Lines:   lines,
 		Applied: applied,
-		FxUsed:  fx,
+		FxUsed:  fxResult,
 	}
+
 	return resp, nil
 }
 
@@ -380,24 +340,31 @@ func (s *Service) validateQuoteRequest(req *feesv1.QuoteFeesRequest) error {
 	if req == nil {
 		return status.Error(codes.InvalidArgument, "request is required")
 	}
+
 	if req.GetMeta() == nil || strings.TrimSpace(req.GetMeta().GetOrganizationRef()) == "" {
 		return status.Error(codes.InvalidArgument, "meta.organization_ref is required")
 	}
+
 	if req.GetIntent() == nil {
 		return status.Error(codes.InvalidArgument, "intent is required")
 	}
+
 	if req.GetIntent().GetTrigger() == feesv1.Trigger_TRIGGER_UNSPECIFIED {
 		return status.Error(codes.InvalidArgument, "intent.trigger is required")
 	}
+
 	if req.GetIntent().GetBaseAmount() == nil {
 		return status.Error(codes.InvalidArgument, "intent.base_amount is required")
 	}
+
 	if strings.TrimSpace(req.GetIntent().GetBaseAmount().GetAmount()) == "" {
 		return status.Error(codes.InvalidArgument, "intent.base_amount.amount is required")
 	}
+
 	if strings.TrimSpace(req.GetIntent().GetBaseAmount().GetCurrency()) == "" {
 		return status.Error(codes.InvalidArgument, "intent.base_amount.currency is required")
 	}
+
 	return nil
 }
 
@@ -405,6 +372,7 @@ func (s *Service) validatePrecomputeRequest(req *feesv1.PrecomputeFeesRequest) e
 	if req == nil {
 		return status.Error(codes.InvalidArgument, "request is required")
 	}
+
 	return s.validateQuoteRequest(&feesv1.QuoteFeesRequest{Meta: req.GetMeta(), Intent: req.GetIntent()})
 }
 
@@ -413,17 +381,13 @@ func (s *Service) computeQuote(ctx context.Context, orgRef bson.ObjectID, intent
 }
 
 func (s *Service) computeQuoteWithTime(ctx context.Context, orgRef bson.ObjectID, intent *feesv1.Intent, _ *feesv1.PolicyOverrides, trace *tracev1.TraceContext, now time.Time) ([]*feesv1.DerivedPostingLine, []*feesv1.AppliedRule, *feesv1.FXUsed, error) {
-	bookedAt := now
-	if intent.GetBookedAt() != nil && intent.GetBookedAt().IsValid() {
-		bookedAt = intent.GetBookedAt().AsTime()
-	}
+	bookedAt := resolvedBookedAt(intent, now)
 
-	logFields := []zap.Field{
-		zap.Time("booked_at_used", bookedAt),
-	}
+	logFields := []zap.Field{zap.Time("booked_at_used", bookedAt)}
 	if !orgRef.IsZero() {
 		logFields = append(logFields, zap.String("organization_ref", orgRef.Hex()))
 	}
+
 	logFields = append(logFields, logFieldsFromIntent(intent)...)
 	logFields = append(logFields, logFieldsFromTrace(trace)...)
 	logger := s.logger.With(logFields...)
@@ -436,22 +400,13 @@ func (s *Service) computeQuoteWithTime(ctx context.Context, orgRef bson.ObjectID
 	plan, rule, err := s.resolver.ResolveFeeRule(ctx, orgPtr, convertTrigger(intent.GetTrigger()), bookedAt, intent.GetAttributes())
 	if err != nil {
 		s.logger.Warn("Failed to resolve fee rule", zap.Error(err))
-		switch {
-		case errors.Is(err, merrors.ErrNoData):
-			return nil, nil, nil, status.Error(codes.NotFound, fmt.Sprintf("fee rule not found: %s", err.Error()))
-		case errors.Is(err, merrors.ErrDataConflict):
-			return nil, nil, nil, status.Error(codes.FailedPrecondition, fmt.Sprintf("conflicting fee rules: %s", err.Error()))
-		case errors.Is(err, storage.ErrConflictingFeePlans):
-			return nil, nil, nil, status.Error(codes.FailedPrecondition, fmt.Sprintf("conflicting fee plans: %s", err.Error()))
-		case errors.Is(err, storage.ErrFeePlanNotFound):
-			return nil, nil, nil, status.Error(codes.NotFound, fmt.Sprintf("fee plan not found: %s", err.Error()))
-		default:
-			return nil, nil, nil, status.Error(codes.Internal, fmt.Sprintf("failed to resolve fee rule: %s", err.Error()))
-		}
+
+		return nil, nil, nil, mapResolveError(err)
 	}
 
 	originalRules := plan.Rules
 	plan.Rules = []model.FeeRule{*rule}
+
 	defer func() {
 		plan.Rules = originalRules
 	}()
@@ -461,13 +416,115 @@ func (s *Service) computeQuoteWithTime(ctx context.Context, orgRef bson.ObjectID
 		if errors.Is(calcErr, merrors.ErrInvalidArg) {
 			return nil, nil, nil, status.Error(codes.InvalidArgument, calcErr.Error())
 		}
+
 		logger.Warn("Failed to compute fee quote", zap.Error(calcErr))
+
 		return nil, nil, nil, status.Error(codes.Internal, "failed to compute fee quote")
 	}
 
 	return result.Lines, result.Applied, result.FxUsed, nil
 }
 
+func resolvedBookedAt(intent *feesv1.Intent, now time.Time) time.Time {
+	if intent.GetBookedAt() != nil && intent.GetBookedAt().IsValid() {
+		return intent.GetBookedAt().AsTime()
+	}
+
+	return now
+}
+
+func mapResolveError(err error) error {
+	switch {
+	case errors.Is(err, merrors.ErrNoData):
+		return status.Error(codes.NotFound, "fee rule not found: "+err.Error())
+	case errors.Is(err, merrors.ErrDataConflict):
+		return status.Error(codes.FailedPrecondition, "conflicting fee rules: "+err.Error())
+	case errors.Is(err, storage.ErrConflictingFeePlans):
+		return status.Error(codes.FailedPrecondition, "conflicting fee plans: "+err.Error())
+	case errors.Is(err, storage.ErrFeePlanNotFound):
+		return status.Error(codes.NotFound, "fee plan not found: "+err.Error())
+	default:
+		return status.Error(codes.Internal, "failed to resolve fee rule: "+err.Error())
+	}
+}
+
+func (s *Service) observePrecomputeFees(logger mlogger.Logger, err error, resp *feesv1.PrecomputeFeesResponse, trigger feesv1.Trigger, start time.Time) {
+	statusLabel := statusFromError(err)
+	fxUsed := false
+	linesCount := 0
+	appliedCount := 0
+
+	var expiresAt time.Time
+
+	if err == nil && resp != nil {
+		fxUsed = resp.GetFxUsed() != nil
+		linesCount = len(resp.GetLines())
+		appliedCount = len(resp.GetApplied())
+
+		if ts := resp.GetExpiresAt(); ts != nil {
+			expiresAt = ts.AsTime()
+		}
+	}
+
+	observeMetrics("precompute", trigger, statusLabel, fxUsed, time.Since(start))
+
+	logFields := []zap.Field{
+		zap.String("status", statusLabel),
+		zap.Duration("duration", time.Since(start)),
+		zap.Bool("fx_used", fxUsed),
+		zap.String("trigger", trigger.String()),
+		zap.Int("lines", linesCount),
+		zap.Int("applied_rules", appliedCount),
+	}
+	if !expiresAt.IsZero() {
+		logFields = append(logFields, zap.Time("expires_at", expiresAt))
+	}
+	if err != nil {
+		logger.Warn("PrecomputeFees finished", append(logFields, zap.Error(err))...)
+
+		return
+	}
+
+	logger.Info("PrecomputeFees finished", logFields...)
+}
+
+func (s *Service) observeValidateFeeToken(logger mlogger.Logger, err error, resp *feesv1.ValidateFeeTokenResponse, trigger feesv1.Trigger, resultReason string, start time.Time) {
+	statusLabel := statusFromError(err)
+	fxUsed := false
+
+	if err == nil && resp != nil {
+		if !resp.GetValid() {
+			statusLabel = "invalid"
+		}
+
+		fxUsed = resp.GetFxUsed() != nil
+		if resp.GetIntent() != nil {
+			trigger = resp.GetIntent().GetTrigger()
+		}
+	}
+
+	observeMetrics("validate", trigger, statusLabel, fxUsed, time.Since(start))
+
+	logFields := []zap.Field{
+		zap.String("status", statusLabel),
+		zap.Duration("duration", time.Since(start)),
+		zap.Bool("fx_used", fxUsed),
+		zap.String("trigger", trigger.String()),
+		zap.Bool("valid", resp != nil && resp.GetValid()),
+	}
+	if resultReason != "" {
+		logFields = append(logFields, zap.String("reason", resultReason))
+	}
+
+	if err != nil {
+		logger.Warn("ValidateFeeToken finished", append(logFields, zap.Error(err))...)
+
+		return
+	}
+
+	logger.Info("ValidateFeeToken finished", logFields...)
+}
+
 type feeQuoteTokenPayload struct {
 	OrganizationRef string                `json:"organization_ref"`
 	Intent          *feesv1.Intent        `json:"intent"`
@@ -480,17 +537,36 @@ func encodeTokenPayload(payload feeQuoteTokenPayload) (string, error) {
 	if err != nil {
 		return "", merrors.Internal("fees: failed to serialize token payload")
 	}
+
 	return base64.StdEncoding.EncodeToString(data), nil
 }
 
 func decodeTokenPayload(token string) (feeQuoteTokenPayload, error) {
 	var payload feeQuoteTokenPayload
+
 	data, err := base64.StdEncoding.DecodeString(token)
 	if err != nil {
 		return payload, merrors.InvalidArgument("fees: invalid token encoding")
 	}
+
 	if err := json.Unmarshal(data, &payload); err != nil {
 		return payload, merrors.InvalidArgument("fees: invalid token payload")
 	}
+
 	return payload, nil
 }
+
+func (s *Service) startDiscoveryAnnouncer() {
+	if s == nil || s.producer == nil {
+		return
+	}
+
+	announce := discovery.Announcement{
+		Service:    "BILLING_FEES",
+		Operations: []string{"fee.calc"},
+		InvokeURI:  s.invokeURI,
+		Version:    appversion.Create().Short(),
+	}
+	s.announcer = discovery.NewAnnouncer(s.logger, s.producer, mservice.FeePlans, announce)
+	s.announcer.Start()
+}