Method ClaimPendingOutboxEntries

database/lineage.go:279–349

ClaimPendingOutboxEntries claims a batch of pending outbox entries for processing. It uses SELECT FOR UPDATE SKIP LOCKED to allow concurrent processors.

(ctx context.Context, batchSize int, lockDuration time.Duration)

// ClaimPendingOutboxEntries claims a batch of pending outbox entries for processing.
// It uses SELECT FOR UPDATE SKIP LOCKED to allow concurrent processors.
func (d Datasource) ClaimPendingOutboxEntries(ctx context.Context, batchSize int, lockDuration time.Duration) ([]model.LineageOutbox, error) {
	lockedUntil := time.Now().Add(lockDuration)

	query := `
		UPDATE ledgerforge.lineage_outbox
		SET status = $1, locked_until = $2
		WHERE id IN (
			SELECT id FROM ledgerforge.lineage_outbox
			WHERE status = 'pending'
			  AND (locked_until IS NULL OR locked_until < NOW())
			  AND attempts < max_attempts
			ORDER BY created_at ASC
			LIMIT $3
			FOR UPDATE SKIP LOCKED
		)
		RETURNING id, transaction_id, source_balance_id, destination_balance_id, provider, lineage_type, payload, status, attempts, max_attempts, last_error, created_at, processed_at, locked_until, inflight
	`

	rows, err := d.Conn.QueryContext(ctx, query, model.OutboxStatusProcessing, lockedUntil, batchSize)
	if err != nil {
		return nil, apierror.NewAPIError(apierror.ErrInternalServer, "Failed to claim pending outbox entries", err)
	}
	defer func() { _ = rows.Close() }()

	var entries []model.LineageOutbox
	for rows.Next() {
		var entry model.LineageOutbox
		var sourceBalanceID, destinationBalanceID, provider, lastError sql.NullString
		var processedAt, lockedUntilVal sql.NullTime

		err := rows.Scan(
			&entry.ID,
			&entry.TransactionID,
			&sourceBalanceID,
			&destinationBalanceID,
			&provider,
			&entry.LineageType,
			&entry.Payload,
			&entry.Status,
			&entry.Attempts,
			&entry.MaxAttempts,
			&lastError,
			&entry.CreatedAt,
			&processedAt,
			&lockedUntilVal,
			&entry.Inflight,
		)
		if err != nil {
			return nil, apierror.NewAPIError(apierror.ErrInternalServer, "Failed to scan outbox entry", err)
		}

		entry.SourceBalanceID = sourceBalanceID.String
		entry.DestinationBalanceID = destinationBalanceID.String
		entry.Provider = provider.String
		entry.LastError = lastError.String
		if processedAt.Valid {
			entry.ProcessedAt = &processedAt.Time
		}