Function insertLineageOutboxesInTx

database/lineage.go:170–248

(ctx context.Context, tx *sql.Tx, outboxes []*model.LineageOutbox)

}

func insertLineageOutboxesInTx(ctx context.Context, tx *sql.Tx, outboxes []*model.LineageOutbox) error {
	if len(outboxes) == 0 {
		return nil
	}

	var query strings.Builder
	query.WriteString(`
		INSERT INTO ledgerforge.lineage_outbox
		(transaction_id, source_balance_id, destination_balance_id, provider, lineage_type, payload, status, max_attempts, created_at, inflight)
		VALUES
	`)

	args := make([]interface{}, 0, len(outboxes)*10)
	now := time.Now()
	for i, outbox := range outboxes {
		if outbox == nil {
			continue
		}
		if len(args) > 0 {
			query.WriteString(",")
		}
		base := len(args) + 1
		query.WriteString(fmt.Sprintf("($%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d)", base, base+1, base+2, base+3, base+4, base+5, base+6, base+7, base+8, base+9))
		args = append(args,
			outbox.TransactionID,
			outbox.SourceBalanceID,
			outbox.DestinationBalanceID,
			outbox.Provider,
			outbox.LineageType,
			outbox.Payload,
			model.OutboxStatusPending,
			outbox.MaxAttempts,
			now.Add(time.Duration(i)*time.Nanosecond),
			outbox.Inflight,
		)
	}

	if len(args) == 0 {
		return nil
	}

	query.WriteString(" RETURNING id, transaction_id")
	rows, err := tx.QueryContext(ctx, query.String(), args...)
	if err != nil {
		return apierror.NewAPIError(apierror.ErrInternalServer, "Failed to insert lineage outbox entries", err)
	}
	defer func() { _ = rows.Close() }()

	outboxesByTransactionID := make(map[string]*model.LineageOutbox, len(outboxes))
	for _, outbox := range outboxes {
		if outbox == nil {
			continue
		}
		outboxesByTransactionID[outbox.TransactionID] = outbox
	}

	inserted := 0
	for rows.Next() {