CreateBulkTransactions handles the creation of multiple transactions in a batch. If atomic is true: Any failure will cause all transactions to be rolled back (or voided if inflight). If atomic is false: Failures will stop processing but previous transactions remain unaffected. If run_async is true:
(ctx context.Context, req *model.BulkTransactionRequest)
| 2517 | // If atomic is false: Failures will stop processing but previous transactions remain unaffected. |
| 2518 | // If run_async is true: Processing happens in background with webhook notifications. |
| 2519 | func (l *LedgerForge) CreateBulkTransactions(ctx context.Context, req *model.BulkTransactionRequest) (*model.BulkTransactionResult, error) { |
| 2520 | ctx, span := tracer.Start(ctx, "LedgerForge.CreateBulkTransactions") |
| 2521 | defer span.End() |
| 2522 | |
| 2523 | // Generate batch ID (parent transaction ID) |
| 2524 | batchID := model.GenerateUUIDWithSuffix("bulk") |
| 2525 | span.SetAttributes(attribute.String("batch.id", batchID)) |
| 2526 | |
| 2527 | // Check if this should be run asynchronously |
| 2528 | if req.RunAsync { |
| 2529 | if !asyncBulkSemaphore.TryAcquire(1) { |
| 2530 | return nil, apierror.NewAPIError( |
| 2531 | apierror.ErrRateLimited, |
| 2532 | "too many async bulk operations in progress, try again later", |
| 2533 | nil, |
| 2534 | ) |
| 2535 | } |
| 2536 | |
| 2537 | // Start processing in background |
| 2538 | go func() { |
| 2539 | defer asyncBulkSemaphore.Release(1) |
| 2540 | |
| 2541 | // Create a background context with timeout |
| 2542 | bgCtx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) |
| 2543 | defer cancel() |
| 2544 | |
| 2545 | logrus.Infof("Starting async bulk transaction batch %s with %d transactions (atomic: %v, inflight: %v)", |
| 2546 | batchID, len(req.Transactions), req.Atomic, req.Inflight) |
| 2547 | |
| 2548 | // Process transactions in batch |
| 2549 | err := l.processBulkTransactions(bgCtx, req.Transactions, batchID, req.Inflight, req.SkipQueue) |
| 2550 | |
| 2551 | if err != nil { |
| 2552 | // Handle failure (rollback if atomic, send webhook) |
| 2553 | l.handleAsyncBulkTransactionFailure(bgCtx, err, batchID, req.Atomic, req.Inflight) |
| 2554 | } else { |
| 2555 | // Send webhook notification for success |
| 2556 | status := "inflight" |
| 2557 | if !req.Inflight { |
| 2558 | status = "applied" |
| 2559 | } |
| 2560 | l.sendBulkTransactionWebhook(batchID, status, "", len(req.Transactions)) |
| 2561 | logrus.Infof("Completed async bulk transaction batch %s successfully", batchID) |
| 2562 | } |
| 2563 | }() |
| 2564 | |
| 2565 | // Return immediate response indicating async processing started |
| 2566 | return &model.BulkTransactionResult{ |
| 2567 | BatchID: batchID, |
| 2568 | Status: "processing", // Indicate that it's running in the background |
| 2569 | }, nil |
| 2570 | } |
| 2571 | |
| 2572 | // Synchronous processing |
| 2573 | logrus.Infof("Starting sync bulk transaction batch %s with %d transactions (atomic: %v, inflight: %v)", |
| 2574 | batchID, len(req.Transactions), req.Atomic, req.Inflight) |
| 2575 | |
| 2576 | // Process transactions in batch |