PrepareLineageOutbox creates a LineageOutbox entry for atomic insertion with the transaction. This ensures lineage processing intent is captured in the same database transaction, guaranteeing no lineage work is lost even if subsequent async operations fail. Parameters: - ctx context.Context: The co
(ctx context.Context, txn *model.Transaction, sourceBalance, destinationBalance *model.Balance)
| 1385 | // Returns: |
| 1386 | // - *model.LineageOutbox: The outbox entry to insert, or nil if no lineage processing needed. |
| 1387 | func (l *LedgerForge) PrepareLineageOutbox(ctx context.Context, txn *model.Transaction, sourceBalance, destinationBalance *model.Balance) *model.LineageOutbox { |
| 1388 | _, span := tracer.Start(ctx, "PrepareLineageOutbox") |
| 1389 | defer span.End() |
| 1390 | |
| 1391 | provider := l.getLineageProvider(txn) |
| 1392 | |
| 1393 | // Determine what type of lineage processing is needed |
| 1394 | needsCredit := provider != "" && destinationBalance != nil && destinationBalance.TrackFundLineage |
| 1395 | needsDebit := sourceBalance != nil && sourceBalance.TrackFundLineage |
| 1396 | |
| 1397 | if !needsCredit && !needsDebit { |
| 1398 | span.AddEvent("No lineage processing needed") |
| 1399 | return nil |
| 1400 | } |
| 1401 | |
| 1402 | lineageType := "both" |
| 1403 | if needsCredit && !needsDebit { |
| 1404 | lineageType = "credit" |
| 1405 | } else if needsDebit && !needsCredit { |
| 1406 | lineageType = "debit" |
| 1407 | } |
| 1408 | |
| 1409 | // Build the payload with transaction data needed for processing |
| 1410 | payload := LineageOutboxPayload{ |
| 1411 | Amount: txn.Amount, |
| 1412 | PreciseAmount: txn.PreciseAmount.String(), |
| 1413 | Currency: txn.Currency, |
| 1414 | Precision: txn.Precision, |
| 1415 | Reference: txn.Reference, |
| 1416 | SkipQueue: true, |
| 1417 | Inflight: txn.Inflight, |
| 1418 | } |
| 1419 | payloadBytes, err := json.Marshal(payload) |
| 1420 | if err != nil { |
| 1421 | logrus.Errorf("failed to marshal lineage outbox payload: %v", err) |
| 1422 | span.RecordError(err) |
| 1423 | return nil |
| 1424 | } |
| 1425 | |
| 1426 | var srcID, dstID string |
| 1427 | if sourceBalance != nil { |
| 1428 | srcID = sourceBalance.BalanceID |
| 1429 | } |
| 1430 | if destinationBalance != nil { |
| 1431 | dstID = destinationBalance.BalanceID |
| 1432 | } |
| 1433 | |
| 1434 | outbox := &model.LineageOutbox{ |
| 1435 | TransactionID: txn.TransactionID, |
| 1436 | SourceBalanceID: srcID, |
| 1437 | DestinationBalanceID: dstID, |
| 1438 | Provider: provider, |
| 1439 | LineageType: lineageType, |
| 1440 | Payload: payloadBytes, |
| 1441 | MaxAttempts: 5, |
| 1442 | Inflight: txn.Inflight, // explicit column, don't rely on JSON payload |
| 1443 | } |
| 1444 |