Update pubsub consumer

[tmosleyIII]

This pull request introduces support for publishing Stellar payment events to Google Pub/Sub using a new V2 message format, designed for improved downstream integration and message filtering. The main changes include the implementation of the PublishToGooglePubSubV2 consumer, comprehensive documentation for the new format, and example configuration files to guide adoption. Minor clarifications and improvements were also made to existing pipeline configurations.

Google Pub/Sub V2 Integration

• Added a new consumer implementation, PublishToGooglePubSubV2, in consumer_publish_to_google_pubsub_v2.go. This supports publishing payment events in a structured V2 format, with improved field naming, message attributes, and flexible authentication via config or environment variables.
• Updated the pipeline creation logic in main.go to support the new consumer type, enabling it to be used in pipeline configurations.

Documentation

• Added a detailed guide, docs/google-pubsub-v2-format.md, describing the V2 message format, configuration options, migration steps from V1, usage examples, and troubleshooting tips for Google Pub/Sub integration.

Configuration Examples

• Provided an example pipeline configuration, kwickbit-payments-pubsub-v2.yaml, demonstrating how to use the new V2 consumer and message format, including environment variable overrides and emulator support for local testing.

Clarifications and Minor Improvements

• Clarified the topic filter description in usdc_transfers_rpc.yaml to specify XDR-encoded filters.
• Improved documentation in rpc-streaming-mode.yaml to note that poll_interval is required.
• Removed unused time import from test_websocket_client.py for code cleanliness.

[claude]

Pull Request Review: Google Pub/Sub V2 Integration

Summary

This PR introduces a new PublishToGooglePubSubV2 consumer with an improved message format for publishing Stellar payment events to Google Pub/Sub. The implementation is well-structured with comprehensive documentation and configuration examples.

✅ Strengths

Code Quality

• Clean Architecture: The V2 consumer follows the existing codebase patterns and properly implements the processor.Processor interface
• Well-Documented: Extensive inline comments and comprehensive external documentation (315 lines in docs/google-pubsub-v2-format.md)
• Thread-Safe: Proper use of sync.RWMutex for protecting shared statistics
• Error Handling: Good error wrapping with context using fmt.Errorf with %w verb
• Flexible Authentication: Supports multiple authentication methods (credentials file, JSON string, environment variables, default credentials)

Design Decisions

• Message Format: The three-level structure (chainIdentifier, payload, details) provides good separation of concerns
• String Amounts: Using strings for amounts (line 190-191) is correct to avoid precision loss with large numbers
• Message Attributes: Publishing rich metadata as Pub/Sub attributes enables efficient filtering
• Emulator Support: Built-in support for local development via PUBSUB_EMULATOR_HOST

🔍 Issues & Concerns

🔴 Critical Issues

1. Context Lifecycle Management (High Priority)

Location: consumer_publish_to_google_pubsub_v2.go:85

ctx: context.Background(),

Issue: The consumer stores context.Background() during initialization and reuses it throughout its lifetime. This is problematic because:

• The stored context is never canceled, leading to potential resource leaks
• The Process() method receives a context parameter but the Close() method uses the stored context
• If the parent context is canceled, cleanup operations may hang indefinitely

Recommendation:

// Remove ctx field from struct
// Use the context passed to Process() for publishing
// Create a new context with timeout for Close() operations:

func (c *PublishToGooglePubSubV2) Close() error {
    // Use context with timeout for graceful shutdown
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    
    // Use ctx for cleanup operations...
}

2. Missing Helper Function Definition

Location: consumer_publish_to_google_pubsub_v2.go:80-84

The code uses getStringConfig() but this function is not defined in this file. It appears to exist in consumer_save_event_payment_to_postgresql.go:281 but that would require it to be exported or duplicated.

Recommendation: Extract common helper functions to a shared file like consumer/helpers.go or duplicate the helpers in this file:

func getStringConfig(config map[string]interface{}, key, defaultValue string) string {
    if val, ok := config[key].(string); ok {
        return val
    }
    return defaultValue
}

3. Suspicious .gitignore Entry

Location: .gitignore:72

CB23WRDQWGSP6YPMY4UV5C4OW5CBTXKYN3XEATG7KJEZCXMJBYEHOUOV:mint

Issue: This appears to be a Stellar account address followed by :mint. This looks like it might be:

• Accidentally committed sensitive data or test data
• A filename that was mistakenly added
• Related to local testing that shouldn't be in the global gitignore

Recommendation: Remove this line from .gitignore unless there's a specific reason for it. If it's test-related, document why it needs to be ignored.

⚠️ Medium Priority Issues

4. Synchronous Publishing May Block Pipeline

Location: consumer_publish_to_google_pubsub_v2.go:224

messageID, err := result.Get(ctx)

Issue: The code blocks waiting for each publish to complete. For high-throughput pipelines, this could create a bottleneck. The Google Pub/Sub client supports async publishing with batching.

Recommendation: Consider using async publishing with periodic flushing:

// Don't block on result.Get() in the hot path
// Let Pub/Sub client batch messages automatically
// Check errors periodically or log them asynchronously

5. Stats Locking Granularity

Location: Multiple instances (lines 180-182, 205-207, 225-228, 231-234)

Issue: The code locks the mutex multiple times per message processing. This is correct but could be optimized.

Current:

c.mu.Lock()
c.stats.TotalProcessed++
c.mu.Unlock()
// ... processing ...
c.mu.Lock()
c.stats.SuccessfulPublishes++
c.mu.Unlock()

Recommendation: Use atomic operations for simple counters:

import "sync/atomic"

type PublishToGooglePubSubV2Stats struct {
    TotalProcessed      atomic.Uint64
    SuccessfulPublishes atomic.Uint64
    FailedPublishes     atomic.Uint64
}

// Then use:
c.stats.TotalProcessed.Add(1)

6. Close() Method Lock Duration

Location: consumer_publish_to_google_pubsub_v2.go:256-257

func (c *PublishToGooglePubSubV2) Close() error {
    c.mu.Lock()
    defer c.mu.Unlock()

Issue: The entire Close() method holds a write lock, including potentially slow I/O operations (topic stop, client close). This prevents any concurrent reads of stats.

Recommendation: Only lock for reading stats, not for cleanup operations:

func (c *PublishToGooglePubSubV2) Close() error {
    // Read stats under lock
    c.mu.RLock()
    stats := c.stats
    c.mu.RUnlock()
    
    log.Printf("PublishToGooglePubSubV2 Stats: Processed %d events...", 
        stats.TotalProcessed)
    
    // Perform I/O without holding lock
    if c.topic != nil {
        c.topic.Stop()
    }
    // ...
}

7. Error Handling in Close() Could Be Improved

Location: consumer_publish_to_google_pubsub_v2.go:254-275

Issue: If topic.Stop() encounters issues, there's no error handling. While this is likely fine for cleanup, it should be considered.

Recommendation: Add graceful error handling:

if c.topic != nil {
    log.Printf("PublishToGooglePubSubV2: Stopping topic...")
    c.topic.Stop()
}

💡 Low Priority / Style Issues

8. Inconsistent Naming Convention

Location: consumer_publish_to_google_pubsub_v2.go:80-84

The config keys use snake_case (project_id, topic_id) while the environment variables use SCREAMING_SNAKE_CASE. This is fine but be aware some configs use camelCase in the codebase.

9. Magic String Duplication

Location: Lines 215-220

The string "payment" appears in multiple places. Consider defining as a constant:

const (
    EventTypePayment = "payment"
    MessageVersionV2 = "v2"
)

10. Verbose Logging

Location: consumer_publish_to_google_pubsub_v2.go:236-237

Logging every published message could be overwhelming in production. Consider:

• Making it debug-level
• Adding a sample rate
• Using structured logging

if c.stats.SuccessfulPublishes.Load() % 100 == 0 {
    log.Printf("PublishToGooglePubSubV2: Published 100 messages, latest: payment_id=%s", 
        eventPayment.PaymentID)
}

🔒 Security Review

✅ Good Security Practices

• Credentials are loaded from config/env, not hardcoded
• Support for service account JSON follows GCP best practices
• Documentation includes security guidance for minimum permissions

Concerns

• Credentials Logging: Lines 130, 134 log credential file paths which is okay, but be careful not to log credential contents
• No Credential Validation: The code doesn't validate that credentials are properly formatted before attempting to use them

🧪 Test Coverage

Critical Gap: No test files exist for this consumer.

Recommendations:

1. Add unit tests for message transformation
2. Add integration tests with Pub/Sub emulator
3. Test credential loading from different sources
4. Test error conditions (network failures, malformed messages, etc.)

Example test structure:

func TestPublishToGooglePubSubV2_Process(t *testing.T) {
    // Test with emulator
    os.Setenv("PUBSUB_EMULATOR_HOST", "localhost:8085")
    
    consumer, err := NewPublishToGooglePubSubV2(config)
    require.NoError(t, err)
    defer consumer.Close()
    
    // Test publishing EventPayment
    msg := processor.Message{
        Payload: &processor.EventPayment{
            PaymentID: "0x123",
            // ...
        },
    }
    
    err = consumer.Process(context.Background(), msg)
    assert.NoError(t, err)
    
    stats := consumer.GetStats()
    assert.Equal(t, uint64(1), stats.SuccessfulPublishes)
}

📚 Documentation Quality

Excellent: The docs/google-pubsub-v2-format.md is comprehensive with:

• Clear format specification
• Configuration examples
• Migration guide from V1
• Code samples in multiple languages
• Troubleshooting section

Minor Suggestions:

• Add performance benchmarks or throughput expectations
• Document retry behavior and error handling
• Include monitoring/observability recommendations

🎯 Performance Considerations

1. Throughput: Synchronous publishing (result.Get()) limits throughput. For high-volume pipelines, consider async mode
2. Batching: The Google Pub/Sub client has built-in batching. Consider exposing batch settings in config
3. Memory: Each message is marshaled to JSON in memory. For large payloads, this is acceptable
4. Network: Each publish is a separate RPC. The client batches these, but connection pool sizing may need tuning

📋 Configuration Review

The example config kwickbit-payments-pubsub-v2.yaml is well-structured with helpful comments. One suggestion:

# Add these optional performance tuning parameters:
consumers:
  - type: PublishToGooglePubSubV2
    config:
      # ... existing config ...
      
      # Optional: Pub/Sub client settings
      # publish_timeout: 60s
      # max_outstanding_messages: 1000

✅ Code Follows Project Standards

Based on CLAUDE.md:

• ✅ Follows factory pattern for component instantiation
• ✅ Implements Processor interface correctly
• ✅ Uses YAML configuration style consistent with other consumers
• ✅ Follows error handling patterns (fmt.Errorf with %w)
• ✅ Includes comprehensive documentation
• ❌ Missing test coverage (consistent with project, but still a gap)

🎬 Recommendations Summary

Must Fix Before Merge

1. Fix context lifecycle management - Remove stored context, use passed contexts properly
2. Add or export getStringConfig helper function - Code won't compile without it
3. Investigate and fix .gitignore suspicious entry - Potential data leak or mistake

Should Fix

4. Consider async publishing for better throughput
5. Use atomic operations for stats counters
6. Reduce lock duration in Close() method
7. Add basic unit tests (at least for message transformation)

Nice to Have

8. Add constants for magic strings
9. Reduce logging verbosity for production
10. Add performance tuning config options

🏆 Overall Assessment

Rating: 7.5/10

This is a well-crafted PR with excellent documentation and clean code structure. The V2 message format is well-designed and the implementation follows good practices. However, there are critical issues around context management and missing helper functions that must be addressed before merging. The lack of tests is concerning but consistent with the project's current state.

The PR demonstrates strong understanding of:

• Google Cloud Pub/Sub SDK
• Go concurrency patterns
• Configuration management
• Documentation best practices

With the critical fixes applied, this would be a solid addition to the codebase.

---

Files Reviewed: 8 files, 701 additions, 3 deletions
Review Completed: 2025-11-19

Reviewed using repository guidance from CLAUDE.md

[Copilot]

Pull Request Overview

This pull request introduces a new V2 format for publishing Stellar payment events to Google Pub/Sub, with improved message structure and enhanced downstream integration capabilities. The implementation adds a new consumer type, comprehensive documentation, and example configurations to guide adoption.

Key Changes:

• Added PublishToGooglePubSubV2 consumer with structured three-level message format (chainIdentifier, payload, details) and improved field naming
• Comprehensive documentation covering V2 format, migration from V1, local testing with emulator, and code examples in TypeScript and Python
• Example configuration demonstrating V2 consumer usage with environment variable support and emulator testing

Reviewed Changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
main.go	Registers new PublishToGooglePubSubV2 consumer type in the consumer factory
consumer/consumer_publish_to_google_pubsub_v2.go	Implements V2 consumer with structured message format, flexible authentication, and message attributes for filtering
docs/google-pubsub-v2-format.md	Comprehensive documentation of V2 format, configuration, migration guide, and usage examples
config/examples/kwickbit-payments-pubsub-v2.yaml	Example pipeline configuration demonstrating V2 consumer with environment variable overrides
config/examples/rpc-streaming-mode.yaml	Clarified that poll_interval is required in streaming mode
config/base/usdc_transfers_rpc.yaml	Improved topic filter documentation to specify XDR-encoded format
config/base/test_websocket_client.py	Removed unused time import
.gitignore	Added contract address file pattern to ignore list

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The getStringConfig helper function is used but not defined in this file. This will cause a compilation error. The function needs to be added to this file, similar to how it's defined in consumer_save_event_payment_to_postgresql.go:

func getStringConfig(config map[string]interface{}, key, defaultValue string) string {
	if val, ok := config[key].(string); ok {
		return val
	}
	return defaultValue
}

[claude]

Pull Request Review - PublishToGooglePubSubV2 Consumer

I've reviewed PR #27 which introduces a new V2 message format for Google Pub/Sub integration. Here's my detailed feedback:

---

✅ Strengths

1. Excellent Documentation: The docs/google-pubsub-v2-format.md is comprehensive and well-structured, including:

   • Clear message format specifications
   • Migration guide from V1
   • Local testing instructions with emulator
   • Example usage in multiple languages (TypeScript, Python)

2. Clean Code Architecture: The V2 consumer follows the established pattern from V1 and maintains consistency with the codebase style

3. Good Configuration Flexibility: Multiple authentication methods supported (file, JSON string, environment variables)

4. Proper Error Handling: Good use of error wrapping and descriptive error messages

5. Thread-Safe Statistics: Using atomic operations for stats is a good improvement over V1's mutex-based approach

---

🔧 Issues & Recommendations

1. Code Duplication - getStringConfig Helper Function

Issue: The getStringConfig helper is not defined in this file. It's defined in consumer_save_event_payment_to_postgresql.go:281, which creates a hidden dependency.

Impact:

• Compilation will fail if that file is removed
• Not immediately obvious where the function comes from
• Violates DRY principle across the consumer package

Recommendation: Create a shared consumer/helpers.go file with common helper functions:

// consumer/helpers.go
package consumer

func getStringConfig(config map[string]interface{}, key, defaultValue string) string {
    if val, ok := config[key].(string); ok {
        return val
    }
    return defaultValue
}

func getIntConfig(config map[string]interface{}, key string, defaultValue int) int {
    if val, ok := config[key].(int); ok {
        return val
    }
    if val, ok := config[key].(float64); ok {
        return int(val)
    }
    return defaultValue
}

2. Inconsistent Mutex Usage Between V1 and V2

Issue:

• V1 uses sync.RWMutex for statistics (lines 26, 130-132)
• V2 uses sync/atomic operations (lines 80-84, 191, 214, 233, 237)

Recommendation:

• The atomic approach in V2 is more performant and correct - consider backporting to V1
• Document why the approaches differ, or standardize on atomic operations
• Current V2 approach is better for high-throughput scenarios

3. Context Handling Inconsistency

Issue:

• V1 stores ctx context.Context as a field (line 27) and uses it during initialization
• V2 creates a background context on-the-fly in NewPublishToGooglePubSubV2 (line 149)
• Both use the context passed to Process() for publishing

Recommendation: V2's approach is better (don't store context as field). Consider updating V1 to match.

4. Missing Resource Cleanup Verification

Issue: The Close() method doesn't verify if the topic was stopped successfully or handle potential errors from topic.Stop()

Current Code (line 265):

c.topic.Stop()

Recommendation: While Stop() doesn't return an error, add a comment or timeout verification:

// Stop the topic (flush any pending messages)
// Note: topic.Stop() blocks until all pending publishes complete
if c.topic != nil {
    log.Printf("PublishToGooglePubSubV2: Stopping topic to flush pending messages...")
    c.topic.Stop()
    log.Printf("PublishToGooglePubSubV2: Topic stopped successfully")
}

5. Type Conversion Potential Issue

Issue: At line 199-200, converting uint64 amounts to strings:

Amount:          strconv.FormatUint(eventPayment.Amount, 10),
RoyaltyFee:      strconv.FormatUint(eventPayment.RoyaltyAmount, 10),

This is correct and matches the documented V2 format. However, the documentation could clarify the precision preservation rationale.

Recommendation: Ensure downstream consumers are aware these are string-encoded integers, not floating point. The docs do mention this, which is good.

6. Configuration Validation - Missing Topic Creation Permission Check

Minor Issue: The code attempts to create topics if they don't exist (lines 165-171), but doesn't validate if the service account has pubsub.topics.create permission beforehand.

Recommendation: Add a note in the documentation about required permissions, or make topic auto-creation optional via config:

auto_create_topic: false  # Default true for backward compatibility

---

🧪 Test Coverage Concerns

Critical Issue: No test files exist for this consumer (consumer_publish_to_google_pubsub_v2_test.go)

Recommendation: Add unit tests covering:

• Message formatting correctness
• Environment variable override behavior
• Error handling (invalid chain identifier, missing config, etc.)
• Statistics tracking accuracy
• Mock Pub/Sub client for testing without real GCP resources

Example test structure:

func TestNewPublishToGooglePubSubV2_InvalidChainIdentifier(t *testing.T) {
    config := map[string]interface{}{
        "project_id": "test-project",
        "topic_id": "test-topic",
        "chain_identifier": "InvalidChain",
    }
    _, err := NewPublishToGooglePubSubV2(config)
    if err == nil {
        t.Error("Expected error for invalid chain identifier")
    }
}

---

🔒 Security Considerations

1. ✅ Good: Credentials are loaded from environment variables or files (not hardcoded)
2. ✅ Good: Warning when no credentials provided (line 145)
3. ⚠️ Minor: Consider logging sanitization - ensure credentials aren't accidentally logged if config is printed

---

📊 Performance

1. ✅ Good: Atomic operations for stats are more efficient than mutexes
2. ✅ Good: Blocking on result.Get(ctx) ensures message delivery confirmation
3. ⚠️ Consider: For very high throughput, you might want async publishing with batch confirmations, but current approach prioritizes reliability

---

📝 Minor Code Quality Issues

1. Line 239: Log message could include more context for troubleshooting:

log.Printf("PublishToGooglePubSubV2: Published payment_id=%s, chain=%s, block=%d, tx=%s, messageID=%s",
    eventPayment.PaymentID, c.ChainIdentifier, eventPayment.BlockHeight, eventPayment.TxHash[:8], messageID)

2. Missing godoc comments: Constants at lines 25-34 could use documentation comments

---

📋 Config File Review

kwickbit-payments-pubsub-v2.yaml:

• ✅ Well-documented with inline comments
• ✅ Shows multiple auth options
• ⚠️ Contains placeholder YOUR_KWICKBIT_CONTRACT_ID - ensure users know to replace this

---

🎯 Summary

Overall Assessment: This is a well-implemented feature with excellent documentation. The code quality is good and follows established patterns in the codebase.

Must Fix Before Merge:

1. ❗ Add consumer/helpers.go to eliminate hidden dependency on getStringConfig
2. ❗ Add basic unit tests (at minimum, config validation tests)

Should Fix:
3. Document the atomic vs mutex statistics approach difference between V1/V2
4. Consider making topic auto-creation configurable

Nice to Have:
5. Add integration test with Pub/Sub emulator
6. Enhance logging with transaction hash truncation
7. Add godoc comments to constants

Migration Compatibility: The V2 format is well-designed for coexistence with V1. The message_version attribute is a good practice.

---

Recommendation: Request changes to address items #1 and #2, then approve after fixes.