feat(aggregator): provider selection phase 1

controllers/accounts/profile.go

@@ -16,7 +16,6 @@ import (
 	"github.com/paycrest/aggregator/ent/providerfiataccount"
 	"github.com/paycrest/aggregator/ent/providerordertoken"
 	"github.com/paycrest/aggregator/ent/providerprofile"
-	"github.com/paycrest/aggregator/ent/provisionbucket"
 	"github.com/paycrest/aggregator/ent/senderordertoken"
 	"github.com/paycrest/aggregator/ent/senderprofile"
 	"github.com/paycrest/aggregator/ent/token"
@@ -35,15 +34,13 @@ var orderConf = config.OrderConfig()
 
 // ProfileController is a controller type for profile settings
 type ProfileController struct {
-	apiKeyService        *svc.APIKeyService
-	priorityQueueService *svc.PriorityQueueService
+	apiKeyService *svc.APIKeyService
 }
 
 // NewProfileController creates a new instance of ProfileController
 func NewProfileController() *ProfileController {
 	return &ProfileController{
-		apiKeyService:        svc.NewAPIKeyService(),
-		priorityQueueService: svc.NewPriorityQueueService(),
+		apiKeyService: svc.NewAPIKeyService(),
 	}
 }
 
@@ -666,8 +663,6 @@ func (ctrl *ProfileController) UpdateProviderProfile(ctx *gin.Context) {
 		txUpdate.SetVisibilityMode(providerprofile.VisibilityMode(payload.VisibilityMode))
 	}
 
-	var allBuckets []*ent.ProvisionBucket
-
 	// Process all token operations
 	for _, op := range tokenOperations {
 
@@ -731,35 +726,6 @@ func (ctrl *ProfileController) UpdateProviderProfile(ctx *gin.Context) {
 				return
 			}
 		}
-
-		// Collect buckets for this token
-		convertedMin := op.TokenPayload.MinOrderAmount.Mul(op.Rate)
-		convertedMax := op.TokenPayload.MaxOrderAmount.Mul(op.Rate)
-
-		buckets, err := tx.ProvisionBucket.
-			Query().
-			Where(
-				provisionbucket.And(
-					provisionbucket.HasCurrencyWith(fiatcurrency.IDEQ(op.Currency.ID)),
-					provisionbucket.MinAmountLTE(convertedMax), // providerMin ≤ bucketMax
-					provisionbucket.MaxAmountGTE(convertedMin), // providerMax ≥ bucketMin
-				),
-			).
-			All(reqCtx)
-		if err != nil {
-			if rollbackErr := tx.Rollback(); rollbackErr != nil {
-				logger.Errorf("Failed to rollback transaction: %v", rollbackErr)
-			}
-			logger.WithFields(logger.Fields{
-				"Error":      fmt.Sprintf("%v", err),
-				"ProviderID": provider.ID,
-				"MinAmount":  op.TokenPayload.MinOrderAmount,
-				"MaxAmount":  op.TokenPayload.MaxOrderAmount,
-			}).Errorf("Failed to assign provider to buckets")
-			u.APIResponse(ctx, http.StatusInternalServerError, "error", "Failed to update profile", nil)
-			return
-		}
-		allBuckets = append(allBuckets, buckets...)
 	}
 
 	// Process all fiat account operations (same pattern as tokens)
@@ -878,26 +844,6 @@ func (ctrl *ProfileController) UpdateProviderProfile(ctx *gin.Context) {
 		}
 	}
 
-	// Deduplicate buckets to prevent duplicate many-to-many edges
-	seenBuckets := make(map[int]bool)
-	var dedupedBuckets []*ent.ProvisionBucket
-	for _, bucket := range allBuckets {
-		if !seenBuckets[bucket.ID] {
-			seenBuckets[bucket.ID] = true
-			dedupedBuckets = append(dedupedBuckets, bucket)
-		}
-	}
-
-	// Update provider profile with deduplicated buckets
-	// When Tokens field is present, always clear existing buckets first
-	if payload.Tokens != nil {
-		txUpdate.ClearProvisionBuckets()
-		// Only add buckets if there are any matches
-		if len(dedupedBuckets) > 0 {
-			txUpdate.AddProvisionBuckets(dedupedBuckets...)
-		}
-	}
-
 	// Save provider profile update within the transaction
 	_, err = txUpdate.Save(reqCtx)
 	if err != nil {

controllers/index.go

@@ -33,6 +33,7 @@ import (
 	tokenEnt "github.com/paycrest/aggregator/ent/token"
 	"github.com/paycrest/aggregator/ent/user"
 	svc "github.com/paycrest/aggregator/services"
+	"github.com/paycrest/aggregator/services/assignment"
 	"github.com/paycrest/aggregator/services/common"
 	"github.com/paycrest/aggregator/services/email"
 	explorer "github.com/paycrest/aggregator/services/explorer"
@@ -61,7 +62,7 @@ var orderConf = config.OrderConfig()
 // Controller is the default controller for other endpoints
 type Controller struct {
 	orderService          types.OrderService
-	priorityQueueService  *svc.PriorityQueueService
+	priorityQueueService  *assignment.Service
 	receiveAddressService *svc.ReceiveAddressService
 	kycService            types.KYCProvider
 	slackService          *svc.SlackService
@@ -77,7 +78,7 @@ func NewController() *Controller {
 
 	return &Controller{
 		orderService:          orderSvc.NewOrderEVM(),
-		priorityQueueService:  svc.NewPriorityQueueService(),
+		priorityQueueService:  assignment.New(),
 		receiveAddressService: svc.NewReceiveAddressService(),
 		kycService:            smile.NewSmileIDService(),
 		slackService:          svc.NewSlackService(serverConf.SlackWebhookURL),

controllers/provider/provider.go

@@ -33,6 +33,7 @@ import (
 	"github.com/paycrest/aggregator/ent/token"
 	"github.com/paycrest/aggregator/ent/transactionlog"
 	"github.com/paycrest/aggregator/services"
+	"github.com/paycrest/aggregator/services/assignment"
 	"github.com/paycrest/aggregator/services/balance"
 	"github.com/paycrest/aggregator/services/contracts"
 	orderService "github.com/paycrest/aggregator/services/order"
@@ -448,9 +449,6 @@ func (ctrl *ProviderController) handleExportPaymentOrders(ctx *gin.Context, prov
 		WithToken(func(tq *ent.TokenQuery) {
 			tq.WithNetwork()
 		}).
-		WithProvisionBucket(func(pbq *ent.ProvisionBucketQuery) {
-			pbq.WithCurrency()
-		}).
 		Limit(maxExportLimit).
 		Order(ent.Desc(paymentorder.FieldCreatedAt), ent.Desc(paymentorder.FieldID)).
 		All(reqCtx)
@@ -525,8 +523,10 @@ func (ctrl *ProviderController) handleExportPaymentOrders(ctx *gin.Context, prov
 		}
 
 		var currencyCode string
-		if paymentOrder.Edges.ProvisionBucket != nil && paymentOrder.Edges.ProvisionBucket.Edges.Currency != nil {
-			currencyCode = paymentOrder.Edges.ProvisionBucket.Edges.Currency.Code
+		if paymentOrder.Institution != "" {
+			if inst, ierr := u.GetInstitutionByCode(reqCtx, paymentOrder.Institution, true); ierr == nil && inst != nil && inst.Edges.FiatCurrency != nil {
+				currencyCode = inst.Edges.FiatCurrency.Code
+			}
 		}
 
 		row := []string{
@@ -590,7 +590,6 @@ func (ctrl *ProviderController) AcceptOrder(ctx *gin.Context) {
 			WithProvider().
 			WithSenderProfile().
 			WithToken(func(tq *ent.TokenQuery) { tq.WithNetwork() }).
-			WithProvisionBucket(func(pbq *ent.ProvisionBucketQuery) { pbq.WithCurrency() }).
 			Only(reqCtx)
 		if fallbackErr != nil || fallbackOrder == nil {
 			if fallbackErr != nil && !ent.IsNotFound(fallbackErr) {
@@ -628,9 +627,7 @@ func (ctrl *ProviderController) AcceptOrder(ctx *gin.Context) {
 		}
 		if fallbackOrder.Direction == paymentorder.DirectionOnramp {
 			result["amount"] = fallbackOrder.Amount.Add(fallbackOrder.SenderFee).Mul(fallbackOrder.Rate).RoundBank(0).String()
-			if fallbackOrder.Edges.ProvisionBucket != nil && fallbackOrder.Edges.ProvisionBucket.Edges.Currency != nil {
-				result["currency"] = fallbackOrder.Edges.ProvisionBucket.Edges.Currency.Code
-			} else if fallbackOrder.Institution != "" {
+			if fallbackOrder.Institution != "" {
 				inst, instErr := u.GetInstitutionByCode(reqCtx, fallbackOrder.Institution, true)
 				if instErr == nil && inst != nil && inst.Edges.FiatCurrency != nil {
 					result["currency"] = inst.Edges.FiatCurrency.Code
@@ -1159,9 +1156,6 @@ func (ctrl *ProviderController) FulfillOrder(ctx *gin.Context) {
 				tq.WithNetwork()
 			})
 			poq.WithProvider()
-			poq.WithProvisionBucket(func(pbq *ent.ProvisionBucketQuery) {
-				pbq.WithCurrency()
-			})
 		}).
 		Only(reqCtx)
 	if err != nil {
@@ -1214,9 +1208,7 @@ func (ctrl *ProviderController) FulfillOrder(ctx *gin.Context) {
 				WithOrder(func(poq *ent.PaymentOrderQuery) {
 					poq.WithToken(func(tq *ent.TokenQuery) {
 						tq.WithNetwork()
-					}).WithProvider().WithProvisionBucket(func(pbq *ent.ProvisionBucketQuery) {
-						pbq.WithCurrency()
-					})
+					}).WithProvider()
 				}).
 				Only(reqCtx)
 			if err != nil {
@@ -1408,10 +1400,7 @@ func (ctrl *ProviderController) handlePayoutFulfillment(ctx *gin.Context, orderI
 		}
 		providerID := fulfillment.Edges.Order.Edges.Provider.ID
 		currency := ""
-		if fulfillment.Edges.Order.Edges.ProvisionBucket != nil && fulfillment.Edges.Order.Edges.ProvisionBucket.Edges.Currency != nil {
-			currency = fulfillment.Edges.Order.Edges.ProvisionBucket.Edges.Currency.Code
-		}
-		if currency == "" && fulfillment.Edges.Order.Institution != "" {
+		if fulfillment.Edges.Order.Institution != "" {
 			inst, instErr := u.GetInstitutionByCode(reqCtx, fulfillment.Edges.Order.Institution, true)
 			if instErr == nil && inst != nil && inst.Edges.FiatCurrency != nil {
 				currency = inst.Edges.FiatCurrency.Code
@@ -1421,8 +1410,8 @@ func (ctrl *ProviderController) handlePayoutFulfillment(ctx *gin.Context, orderI
 			logger.WithFields(logger.Fields{
 				"OrderID": orderID.String(),
 				"TxID":    payload.TxID,
-			}).Errorf("FulfillOrder: order missing provision bucket and could not resolve currency from institution")
-			u.APIResponse(ctx, http.StatusInternalServerError, "error", "Order missing provider or provision bucket", nil)
+			}).Errorf("FulfillOrder: could not resolve fiat currency from institution")
+			u.APIResponse(ctx, http.StatusInternalServerError, "error", "Order missing provider or institution currency", nil)
 			_ = tx.Rollback()
 			return
 		}
@@ -1463,6 +1452,10 @@ func (ctrl *ProviderController) handlePayoutFulfillment(ctx *gin.Context, orderI
 			return
 		}
 
+		if scoreErr := assignment.ApplyProviderScoreChange(reqCtx, orderID, assignment.ScoreEventFulfilledValidated, decimal.NewFromFloat(assignment.RewardFulfilledValidated)); scoreErr != nil {
+			logger.WithFields(logger.Fields{"OrderID": orderID.String(), "Error": scoreErr.Error()}).Warnf("FulfillOrder: score reward")
+		}
+
 		// Clean up order exclude list from Redis (best effort, don't fail if it errors)
 		orderKey := fmt.Sprintf("order_exclude_list_%s", orderID)
 		_ = storage.RedisClient.Del(reqCtx, orderKey).Err()
@@ -1549,10 +1542,7 @@ func (ctrl *ProviderController) handlePayoutFulfillment(ctx *gin.Context, orderI
 		}
 		providerID := fulfillment.Edges.Order.Edges.Provider.ID
 		currency := ""
-		if fulfillment.Edges.Order.Edges.ProvisionBucket != nil && fulfillment.Edges.Order.Edges.ProvisionBucket.Edges.Currency != nil {
-			currency = fulfillment.Edges.Order.Edges.ProvisionBucket.Edges.Currency.Code
-		}
-		if currency == "" && fulfillment.Edges.Order.Institution != "" {
+		if fulfillment.Edges.Order.Institution != "" {
 			inst, instErr := u.GetInstitutionByCode(reqCtx, fulfillment.Edges.Order.Institution, true)
 			if instErr == nil && inst != nil && inst.Edges.FiatCurrency != nil {
 				currency = inst.Edges.FiatCurrency.Code
@@ -1562,7 +1552,7 @@ func (ctrl *ProviderController) handlePayoutFulfillment(ctx *gin.Context, orderI
 			logger.WithFields(logger.Fields{
 				"OrderID": orderID.String(),
 				"TxID":    payload.TxID,
-			}).Errorf("FulfillOrder: order missing provision bucket and could not resolve currency from institution")
+			}).Errorf("FulfillOrder: could not resolve fiat currency from institution")
 			return
 		}
 		amount := fulfillment.Edges.Order.Amount.Mul(fulfillment.Edges.Order.Rate).RoundBank(0)
@@ -1578,6 +1568,9 @@ func (ctrl *ProviderController) handlePayoutFulfillment(ctx *gin.Context, orderI
 			}).Errorf("failed to release reserved balance for failed validation")
 			// Don't return error here as the order status is already updated
 		}
+		if scoreErr := assignment.ApplyProviderScoreChange(reqCtx, orderID, assignment.ScoreEventValidationFailed, decimal.NewFromFloat(assignment.PenaltyValidationFailed)); scoreErr != nil {
+			logger.WithFields(logger.Fields{"OrderID": orderID.String(), "Error": scoreErr.Error()}).Warnf("FulfillOrder: score validation failed penalty")
+		}
 
 	default:
 		// Create log and update order in one transaction so the log is never orphaned
@@ -1791,6 +1784,9 @@ func (ctrl *ProviderController) handlePayinFulfillment(ctx *gin.Context, orderID
 				logger.Errorf("Failed to release token balance for payin validation failure (order %s): %v", orderID, relErr)
 			}
 		}
+		if scoreErr := assignment.ApplyProviderScoreChange(reqCtx, orderID, assignment.ScoreEventValidationFailed, decimal.NewFromFloat(assignment.PenaltyValidationFailed)); scoreErr != nil {
+			logger.WithFields(logger.Fields{"OrderID": orderID.String(), "Error": scoreErr.Error()}).Warnf("FulfillOrder payin: score validation failed penalty")
+		}
 		u.APIResponse(ctx, http.StatusOK, "success", "Fulfillment validation failed", nil)
 		return
 	case paymentorderfulfillment.ValidationStatusSuccess:
@@ -2024,6 +2020,10 @@ func (ctrl *ProviderController) handlePayinFulfillment(ctx *gin.Context, orderID
 		return
 	}
 
+	if scoreErr := assignment.ApplyProviderScoreChange(reqCtx, orderID, assignment.ScoreEventFulfilledValidated, decimal.NewFromFloat(assignment.RewardFulfilledValidated)); scoreErr != nil {
+		logger.WithFields(logger.Fields{"OrderID": orderID.String(), "Error": scoreErr.Error()}).Warnf("FulfillOrder payin: score reward")
+	}
+
 	u.APIResponse(ctx, http.StatusOK, "success", "Order submitted for settlement", nil)
 }
 
@@ -2358,9 +2358,6 @@ func (ctrl *ProviderController) CancelOrder(ctx *gin.Context) {
 			tq.WithNetwork()
 		}).
 		WithProvider().
-		WithProvisionBucket(func(pbq *ent.ProvisionBucketQuery) {
-			pbq.WithCurrency()
-		}).
 		Only(reqCtx)
 	if err != nil {
 		logger.WithFields(logger.Fields{
@@ -2381,49 +2378,8 @@ func (ctrl *ProviderController) CancelOrder(ctx *gin.Context) {
 	} else if payload.Reason != "Insufficient funds" {
 		cancellationCount += 1
 		orderUpdate.AppendCancellationReasons([]string{payload.Reason})
-	} else if payload.Reason == "Insufficient funds" && order.Edges.ProvisionBucket != nil && order.Edges.ProvisionBucket.Edges.Currency != nil {
-		// Search for the specific provider in the queue using a Redis list (private orders with nil bucket are not in the queue)
-		redisKey := fmt.Sprintf("bucket_%s_%s_%s", order.Edges.ProvisionBucket.Edges.Currency.Code, order.Edges.ProvisionBucket.MinAmount, order.Edges.ProvisionBucket.MaxAmount)
-
-		// Check if the provider ID exists in the list
-		for index := -1; ; index-- {
-			providerData, err := storage.RedisClient.LIndex(reqCtx, redisKey, int64(index)).Result()
-			if err != nil {
-				break
-			}
-
-			// Extract the id from the data (format "providerID:token:network:rate:minAmount:maxAmount")
-			parts := strings.Split(providerData, ":")
-			if len(parts) != 6 {
-				logger.WithFields(logger.Fields{
-					"Provider Data": providerData,
-				}).Error("Invalid provider data format")
-				continue // Skip this entry due to invalid format
-			}
-
-			if parts[0] == provider.ID {
-				// Remove the provider from the list
-				placeholder := "DELETED_PROVIDER" // Define a placeholder value
-				_, err := storage.RedisClient.LSet(reqCtx, redisKey, int64(index), placeholder).Result()
-				if err != nil {
-					logger.WithFields(logger.Fields{
-						"Error": fmt.Sprintf("%v", err),
-						"Index": index,
-					}).Errorf("Failed to set placeholder at index %d: %v", index, err)
-				}
-
-				// Remove all occurences of the placeholder from the list
-				_, err = storage.RedisClient.LRem(reqCtx, redisKey, 0, placeholder).Result()
-				if err != nil {
-					logger.WithFields(logger.Fields{
-						"Error":       fmt.Sprintf("%v", err),
-						"Placeholder": placeholder,
-					}).Errorf("Failed to remove placeholder from circular queue: %v", err)
-				}
-
-				break
-			}
-		}
+	} else if payload.Reason == "Insufficient funds" {
+		orderUpdate.AppendCancellationReasons([]string{payload.Reason})
 	}
 
 	// Update order status to cancelled
@@ -2447,10 +2403,7 @@ func (ctrl *ProviderController) CancelOrder(ctx *gin.Context) {
 	// Release reserved balance for this cancelled order
 	providerID := order.Edges.Provider.ID
 	currency := ""
-	if order.Edges.ProvisionBucket != nil && order.Edges.ProvisionBucket.Edges.Currency != nil {
-		currency = order.Edges.ProvisionBucket.Edges.Currency.Code
-	}
-	if currency == "" && order.Institution != "" {
+	if order.Institution != "" {
 		inst, instErr := u.GetInstitutionByCode(reqCtx, order.Institution, true)
 		if instErr == nil && inst != nil && inst.Edges.FiatCurrency != nil {
 			currency = inst.Edges.FiatCurrency.Code
@@ -2490,6 +2443,17 @@ func (ctrl *ProviderController) CancelOrder(ctx *gin.Context) {
 		logger.Errorf("error setting TTL for order %s exclude_list on Redis: %v", orderID, err)
 	}
 
+	switch {
+	case payload.Reason == "Insufficient funds":
+		if scoreErr := assignment.ApplyProviderScoreChange(reqCtx, orderID, assignment.ScoreEventCancelInsufficientFunds, decimal.NewFromFloat(assignment.PenaltyCancelInsufficientFunds)); scoreErr != nil {
+			logger.WithFields(logger.Fields{"OrderID": orderID.String(), "Error": scoreErr.Error()}).Warnf("CancelOrder: score insufficient funds penalty")
+		}
+	case assignment.IsProviderFaultCancelReason(payload.Reason):
+		if scoreErr := assignment.ApplyProviderScoreChange(reqCtx, orderID, assignment.ScoreEventCancelProviderFault, decimal.NewFromFloat(assignment.PenaltyCancelProviderFault)); scoreErr != nil {
+			logger.WithFields(logger.Fields{"OrderID": orderID.String(), "Error": scoreErr.Error()}).Warnf("CancelOrder: score provider-fault penalty")
+		}
+	}
+
 	u.APIResponse(ctx, http.StatusOK, "success", "Order cancelled successfully", nil)
 }