InsertLineageOutboxInTx inserts a lineage outbox entry within an existing database transaction. This ensures the outbox entry is committed atomically with the main transaction.
(ctx context.Context, tx *sql.Tx, outbox *model.LineageOutbox)
| 143 | // InsertLineageOutboxInTx inserts a lineage outbox entry within an existing database transaction. |
| 144 | // This ensures the outbox entry is committed atomically with the main transaction. |
| 145 | func (d Datasource) InsertLineageOutboxInTx(ctx context.Context, tx *sql.Tx, outbox *model.LineageOutbox) error { |
| 146 | query := ` |
| 147 | INSERT INTO ledgerforge.lineage_outbox |
| 148 | (transaction_id, source_balance_id, destination_balance_id, provider, lineage_type, payload, status, max_attempts, created_at, inflight) |
| 149 | VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) |
| 150 | RETURNING id |
| 151 | ` |
| 152 | err := tx.QueryRowContext(ctx, query, |
| 153 | outbox.TransactionID, |
| 154 | outbox.SourceBalanceID, |
| 155 | outbox.DestinationBalanceID, |
| 156 | outbox.Provider, |
| 157 | outbox.LineageType, |
| 158 | outbox.Payload, |
| 159 | model.OutboxStatusPending, |
| 160 | outbox.MaxAttempts, |
| 161 | time.Now(), |
| 162 | outbox.Inflight, |
| 163 | ).Scan(&outbox.ID) |
| 164 | if err != nil { |
| 165 | return apierror.NewAPIError(apierror.ErrInternalServer, "Failed to insert lineage outbox entry", err) |
| 166 | } |
| 167 | return nil |
| 168 | } |
| 169 | |
| 170 | func insertLineageOutboxesInTx(ctx context.Context, tx *sql.Tx, outboxes []*model.LineageOutbox) error { |
| 171 | if len(outboxes) == 0 { |
no test coverage detected