[LFXV2-1084] Add NATS KV Event Processing to Survey Service

[andrest50]

Summary

Implements NATS KV bucket event processing for the survey service to automatically sync survey and survey response data from the v1 system to the v2 system. This enables real-time data synchronization, search indexing, and access control updates.

Ticket

https://linuxfoundation.atlassian.net/browse/LFXV2-1084

Implementation Details

Architecture

• Watches the v1-objects KV bucket for keys matching itx-surveys:* and itx-survey-responses:*
• Uses JetStream consumer with DeliverLastPerSubjectPolicy for latest version processing
• Runs in the same binary as the HTTP API (background goroutine)
• Event processing enabled by default, configurable via EVENT_PROCESSING_ENABLED

Data Transformation

• Transforms v1 DynamoDB format (string fields) → v2 format (proper types)
• Converts string integers to actual integers (nps_value, total_responses, etc.)
• Maps v1 SFIDs → v2 UUIDs for committees and projects via IDMapper service
• Preserves SurveyMonkey answers without transformation

Publishing

Publishes transformed data to two downstream services:

1. Indexer Service (lfx.index.survey, lfx.index.survey_response)
   • Enables search functionality with parent references and access control metadata
2. FGA-Sync Service (lfx.fga-sync.update_access, lfx.fga-sync.delete_access)
   • Updates Fine-Grained Authorization tuples and manages permissions

Deduplication

• Tracks processed events in v1-mappings KV bucket
• Distinguishes between CREATE and UPDATE operations

Error Handling

• Transient errors (NAK for retry): NATS timeouts, IDMapper unavailable, network failures
• Permanent errors (ACK and skip): Invalid JSON, missing required fields
• Retries transient failures up to 3 times with 30-second timeout

Files Changed

New Files

• cmd/survey-api/eventing/event_processor.go - Lifecycle management
• cmd/survey-api/eventing/kv_handler.go - Event routing by key prefix
• cmd/survey-api/eventing/survey_event_handler.go - Survey transformation logic
• cmd/survey-api/eventing/survey_response_event_handler.go - Response transformation logic
• internal/domain/event_models.go - v2 data models
• internal/domain/event_publisher.go - Publisher interface
• internal/infrastructure/eventing/event_config.go - Configuration
• internal/infrastructure/eventing/nats_publisher.go - NATS publishing implementation
• pkg/utils/string.go - String utility functions
• docs/event-processing.md - Comprehensive documentation
• scripts/delete_survey_documents.sh - Cleanup utility script

Modified Files

• cmd/survey-api/main.go - Event processor initialization and shutdown
• README.md - Documentation updates
• go.mod - Added indexer service dependency

Configuration

New environment variables:

• EVENT_PROCESSING_ENABLED - Enable/disable event processing (default: true)
• EVENT_CONSUMER_NAME - JetStream consumer name (default: survey-service-kv-consumer)
• EVENT_STREAM_NAME - JetStream stream name (default: KV_v1-objects)
• EVENT_FILTER_SUBJECT - NATS subject filter (default: $KV.v1-objects.>)

Reference Implementation

Follows the exact pattern from voting service PR #8: linuxfoundation/lfx-v2-voting-service#8

Uses function-based handlers (not struct-based methods) to match the voting service architecture.

Documentation

• Added docs/event-processing.md with architecture, configuration, operations, and troubleshooting
• Updated README.md with event processing feature and configuration
• Included operational procedures and monitoring guidance

Checklist

• Event processor starts and connects to NATS KV bucket
• Survey events transformed correctly from v1 → v2 format
• Committee/project IDs mapped via IDMapper
• String integers converted to proper ints
• Events published to both indexer and FGA-sync
• Deduplication working via v1-mappings bucket
• Error handling matches voting service pattern
• Graceful shutdown without data loss
• Documentation complete and accurate

[Copilot]

Pull request overview

This pull request implements NATS KV bucket event processing for the survey service to enable real-time synchronization of survey and survey response data from v1 (DynamoDB) to v2 (indexer and FGA services). The implementation follows the architectural pattern established in the voting service PR #8.

Changes:

• Added event processing infrastructure with NATS JetStream consumer for KV bucket watching
• Implemented data transformation logic from v1 (string-based DynamoDB format) to v2 (properly typed format)
• Added ID mapping integration for converting v1 SFIDs to v2 UUIDs
• Integrated with indexer service for search functionality and FGA-sync service for access control
• Added comprehensive documentation and operational utilities

Reviewed changes

Copilot reviewed 14 out of 18 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
cmd/survey-api/eventing/event_processor.go	Event processor lifecycle management with NATS connection and JetStream consumer
cmd/survey-api/eventing/kv_handler.go	Routes KV events to appropriate handlers based on key prefix and operation
cmd/survey-api/eventing/survey_event_handler.go	Transforms v1 survey data to v2 format, handles ID mapping and validation
cmd/survey-api/eventing/survey_response_event_handler.go	Transforms v1 survey response data with similar conversion logic
internal/domain/event_models.go	Defines v2 data models for surveys and responses with proper types
internal/domain/event_publisher.go	Publisher interface abstraction for event publishing
internal/infrastructure/eventing/event_config.go	Configuration structure for event processor
internal/infrastructure/eventing/nats_publisher.go	NATS publisher implementation for indexer and FGA-sync
cmd/survey-api/main.go	Initializes event processor on startup and handles graceful shutdown
pkg/utils/string.go	Custom string utility (reinvents stdlib)
go.mod	Adds indexer service dependency
docs/event-processing.md	Comprehensive event processing documentation
README.md	Updates with event processing feature documentation
scripts/delete_survey_documents.sh	Utility script for cleaning up test data from OpenSearch
pkg/models/itx/models.go	Whitespace/alignment formatting cleanup
internal/service/survey_service.go	Minor formatting adjustments
internal/domain/errors.go	Formatting cleanup

---

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

[Copilot]

The logic for deduplicating project UIDs (lines 118-130) and committee UIDs in access messages (lines 163-174) is duplicated. This is a maintenance burden and could lead to inconsistencies.

Consider extracting this into a helper function:

func deduplicateStrings(items []string) []string {
    seen := make(map[string]bool)
    result := []string{}
    for _, item := range items {
        if !seen[item] {
            seen[item] = true
            result = append(result, item)
        }
    }
    return result
}

[Copilot]

The custom Contains function reinvents the standard library's strings.Contains() function. This implementation is more complex and slower than the standard library version which uses optimized assembly on many platforms.

Replace utils.Contains(errStr, "timeout") calls with the standard strings.Contains(errStr, "timeout") throughout the codebase.

Suggested change

	// Contains checks if a string contains a substring
	func Contains(s, substr string) bool {
	return len(s) >= len(substr) && (s == substr || len(s) > len(substr) && (s[:len(substr)] == substr ||
	s[len(s)-len(substr):] == substr ||
	containsMiddle(s, substr)))
	}
	// containsMiddle checks if substr appears in the middle of s
	func containsMiddle(s, substr string) bool {
	for i := 0; i <= len(s)-len(substr); i++ {
	if s[i:i+len(substr)] == substr {
	return true
	}
	}
	return false
	}
	import "strings"
	// Contains checks if a string contains a substring
	func Contains(s, substr string) bool {
	return strings.Contains(s, substr)
	}