QueueTransaction processes and queues a transaction for execution. It handles both single transactions and split transactions, preparing them for processing by setting metadata, status, and managing their persistence and queueing. Parameters: - ctx context.Context: The context for the operation. -
(ctx context.Context, transaction *model.Transaction)
| 1783 | // - *model.Transaction: A pointer to the queued Transaction model. |
| 1784 | // - error: An error if the transaction could not be queued. |
| 1785 | func (l *LedgerForge) QueueTransaction(ctx context.Context, transaction *model.Transaction) (*model.Transaction, error) { |
| 1786 | ctx, span := tracer.Start(ctx, "QueueTransaction") |
| 1787 | defer span.End() |
| 1788 | |
| 1789 | // Initialize transaction metadata and status |
| 1790 | originalRef := transaction.Reference |
| 1791 | setTransactionMetadata(transaction) |
| 1792 | setTransactionStatus(transaction) |
| 1793 | originalTxnID := transaction.TransactionID |
| 1794 | |
| 1795 | // Handle split transactions if needed |
| 1796 | transactions, err := l.handleSplitTransactions(ctx, transaction) |
| 1797 | if err != nil { |
| 1798 | span.RecordError(err) |
| 1799 | return nil, err |
| 1800 | } |
| 1801 | |
| 1802 | // If SkipQueue is true, process synchronously |
| 1803 | if transaction.SkipQueue { |
| 1804 | _, err := l.processTxns(ctx, transaction, transactions, originalTxnID, originalRef) |
| 1805 | if err != nil { |
| 1806 | span.RecordError(err) |
| 1807 | return nil, err |
| 1808 | } |
| 1809 | |
| 1810 | // Update transaction status based on inflight flag |
| 1811 | if transaction.Inflight { |
| 1812 | transaction.Status = StatusInflight |
| 1813 | } else { |
| 1814 | transaction.Status = StatusApplied |
| 1815 | } |
| 1816 | } else { |
| 1817 | if transaction.MetaData != nil && !transaction.SkipQueue { |
| 1818 | transaction.MetaData["QUEUED_PARENT_TRANSACTION"] = originalTxnID |
| 1819 | } |
| 1820 | |
| 1821 | if strings.Contains(transaction.ParentTransaction, "bulk") { |
| 1822 | transaction.MetaData["QUEUED_PARENT_TRANSACTION"] = transaction.ParentTransaction |
| 1823 | } |
| 1824 | // For normal queue mode, process asynchronously |
| 1825 | processTransactionAsync(context.Background(), l, transaction, originalRef, originalTxnID, transactions) |
| 1826 | } |
| 1827 | |
| 1828 | span.AddEvent("Transaction successfully queued", trace.WithAttributes( |
| 1829 | attribute.String("transaction.id", transaction.TransactionID), |
| 1830 | )) |
| 1831 | |
| 1832 | if !transaction.InflightExpiryDate.IsZero() { |
| 1833 | err = l.queue.QueueInflightExpiry(ctx, transaction) |
| 1834 | if err != nil { |
| 1835 | span.RecordError(err) |
| 1836 | return nil, l.logAndRecordError(span, "failed to queue inflight expiry", err) |
| 1837 | } |
| 1838 | } |
| 1839 | |
| 1840 | // Queue an automatic commit task if inflight_commit_date is set. |
| 1841 | if !transaction.InflightCommitDate.IsZero() { |
| 1842 | err = l.queue.QueueInflightCommit(ctx, transaction) |