Skip to content

feat(aggregator): provider selection phase 1 #736

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chibie wants to merge 8 commits into
base: main
Choose a base branch
from
+17,726 −9,291
Open

feat(aggregator): provider selection phase 1 #736

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
4e93bc0
feat(aggregator): provider selection phase 1
chibie Mar 24, 2026
6e50197
Merge branch 'main' into feat/provider-selection-redesign
chibie Mar 24, 2026
3350691
fix(assignment): harden assign flow, rates, retries, score history, a…
chibie Mar 24, 2026
b8a58bd
refactor(order): streamline error handling in payment order validation
chibie Mar 24, 2026
5249316
refactor(tests): simplify Redis bucket handling in sender tests
chibie Mar 24, 2026
062eded
feat(assignment): implement Redis locking for order assignment and co…
chibie Mar 24, 2026
b7777a7
feat(assignment): enhance order assignment logic with Redis locking a…
chibie Mar 24, 2026
0abbaee
feat(assignment): improve order assignment logic with network resolut…
chibie Mar 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 2 additions & 56 deletions controllers/accounts/profile.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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(),
}
}

Expand Down Expand Up @@ -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 {

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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 {
Expand Down
5 changes: 3 additions & 2 deletions controllers/index.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand All @@ -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),
Expand Down
116 changes: 40 additions & 76 deletions controllers/provider/provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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{
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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
Expand All @@ -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
}
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)
}

Expand Down Expand Up @@ -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{
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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)
}

Expand Down
Loading
Loading