ProcessLineageFromOutbox processes a lineage outbox entry. This is called by the outbox worker to perform deferred lineage processing. Parameters: - ctx context.Context: The context for the operation. - entry model.LineageOutbox: The outbox entry to process. Returns: - error: An error if processin
(ctx context.Context, entry model.LineageOutbox)
| 1461 | // Returns: |
| 1462 | // - error: An error if processing fails. |
| 1463 | func (l *LedgerForge) ProcessLineageFromOutbox(ctx context.Context, entry model.LineageOutbox) error { |
| 1464 | ctx, span := tracer.Start(ctx, "ProcessLineageFromOutbox") |
| 1465 | defer span.End() |
| 1466 | |
| 1467 | span.SetAttributes( |
| 1468 | attribute.String("outbox.id", fmt.Sprintf("%d", entry.ID)), |
| 1469 | attribute.String("outbox.transaction_id", entry.TransactionID), |
| 1470 | attribute.String("outbox.lineage_type", entry.LineageType), |
| 1471 | ) |
| 1472 | |
| 1473 | // Handle shadow commit/void operations |
| 1474 | switch entry.LineageType { |
| 1475 | case model.LineageTypeShadowCommit, model.LineageTypeShadowVoid: |
| 1476 | // Extract parent transaction ID from payload |
| 1477 | var payload struct { |
| 1478 | ParentTransactionID string `json:"parent_transaction_id"` |
| 1479 | } |
| 1480 | if err := json.Unmarshal(entry.Payload, &payload); err != nil { |
| 1481 | return fmt.Errorf("failed to unmarshal shadow work payload: %w", err) |
| 1482 | } |
| 1483 | parentTxnID := payload.ParentTransactionID |
| 1484 | if parentTxnID == "" { |
| 1485 | return fmt.Errorf("parent_transaction_id missing in shadow work payload") |
| 1486 | } |
| 1487 | |
| 1488 | if entry.LineageType == model.LineageTypeShadowCommit { |
| 1489 | span.AddEvent("Processing shadow commit from outbox", trace.WithAttributes( |
| 1490 | attribute.String("parent.transaction_id", parentTxnID), |
| 1491 | )) |
| 1492 | if err := l.commitShadowTransactions(ctx, parentTxnID, nil); err != nil { |
| 1493 | return fmt.Errorf("failed to commit shadow transactions: %w", err) |
| 1494 | } |
| 1495 | span.AddEvent("Shadow commit completed from outbox") |
| 1496 | } else { |
| 1497 | span.AddEvent("Processing shadow void from outbox", trace.WithAttributes( |
| 1498 | attribute.String("parent.transaction_id", parentTxnID), |
| 1499 | )) |
| 1500 | if err := l.voidShadowTransactions(ctx, parentTxnID); err != nil { |
| 1501 | return fmt.Errorf("failed to void shadow transactions: %w", err) |
| 1502 | } |
| 1503 | span.AddEvent("Shadow void completed from outbox") |
| 1504 | } |
| 1505 | return nil |
| 1506 | } |
| 1507 | |
| 1508 | // Handle regular lineage processing (credit, debit, both) |
| 1509 | // Fetch the transaction |
| 1510 | txn, err := l.GetTransaction(ctx, entry.TransactionID) |
| 1511 | if err != nil { |
| 1512 | return fmt.Errorf("failed to get transaction: %w", err) |
| 1513 | } |
| 1514 | |
| 1515 | txn.Inflight = entry.Inflight |
| 1516 | |
| 1517 | // Fetch balances |
| 1518 | var sourceBalance, destinationBalance *model.Balance |
| 1519 | if entry.SourceBalanceID != "" { |
| 1520 | sourceBalance, err = l.datasource.GetBalanceByIDLite(entry.SourceBalanceID) |