Function insertPendingBatchTx

internal/webhookpub/worker.go:378–405

insertPendingBatchTx writes one webhook_subscriber_deliveries row per matched webhook in a single multi-row INSERT. Per-row ON CONFLICT (event_id, webhook_id) WHERE event_id IS NOT NULL AND replay_id IS NULL DO NOTHING swallows duplicate inserts that come from a multi-replica lease race (worker A in

(ctx context.Context, tx pgx.Tx, eventID string, webhookIDs []string, eventType string, messageID *string, envelope []byte)

// in migration 028 — Postgres requires exact predicate matching for
// ON CONFLICT to bind to a partial index.
func insertPendingBatchTx(ctx context.Context, tx pgx.Tx, eventID string, webhookIDs []string, eventType string, messageID *string, envelope []byte) error {
	if len(webhookIDs) == 0 {
		return nil
	}

	// Build multi-row VALUES list. pgx supports parameterized arrays,
	// but the simplest portable shape is to fan out the parameters.
	args := make([]any, 0, 5+len(webhookIDs)*2)
	values := make([]string, 0, len(webhookIDs))
	args = append(args, eventID, eventType, messageID, envelope)
	for i, whID := range webhookIDs {
		// id, webhook_id placeholders. The other 4 fields are shared
		// across all rows so reference them by their fixed indexes.
		base := 4 + i*2
		values = append(values, fmt.Sprintf("($%d, $%d, $1, $2, $4, $3, 'pending')", base+1, base+2))
		args = append(args, generateDeliveryID(), whID)
	}

	sql := `INSERT INTO webhook_subscriber_deliveries
	            (id, webhook_id, event_id, event_type, event_payload, message_id, status)
	        VALUES ` + strings.Join(values, ", ") + `
	        ON CONFLICT (event_id, webhook_id)
	            WHERE event_id IS NOT NULL AND replay_id IS NULL
	            DO NOTHING`

	_, err := tx.Exec(ctx, sql, args...)
	return err
}

// generateDeliveryID mirrors the format used by the legacy publisher's
// dbInserter — whd_<32-hex>. The relay's blocker-fix commit moved