Event Processing

June 8, 2026 · View on GitHub

Overview

The survey service implements NATS KV bucket event processing 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 without manual intervention.

Architecture

Components

┌─────────────────┐
│  v1 DynamoDB    │
│   (via KV)      │
└────────┬────────┘
         │
         ├─ itx-surveys:*
         └─ itx-survey-responses:*
         │
         v
┌─────────────────────────────────┐
│   NATS KV Bucket: v1-objects    │
└────────┬────────────────────────┘
         │
         v
┌─────────────────────────────────┐
│     Event Processor             │
│  (JetStream Consumer)           │
└────────┬────────────────────────┘
         │
         ├─ Transform v1 → v2
         ├─ Map IDs (v1 SFID → v2 UUID)
         │
         v
┌────────────────┬────────────────┐
│                │                │
│   Indexer      │   FGA-Sync     │
│   Service      │   Service      │
│                │                │
│ (Search Index) │ (Access Control)
└────────────────┴────────────────┘

Event Flow

  1. Watch: Event processor watches the v1-objects KV bucket, filtered server-side to only receive keys matching:

    • itx-surveys.* - Survey data ($KV.v1-objects.itx-surveys.>)
    • itx-survey-responses.* - Survey response data ($KV.v1-objects.itx-survey-responses.>)

    Note on encoding: KV entries from v1 may be encoded as either JSON or msgpack. The handler attempts JSON deserialization first; if that fails, it falls back to msgpack. Both formats are supported transparently.

  2. Transform: Converts v1 format to v2 format:

    • String fields → proper types (strings to ints)
    • v1 SFIDs → v2 UUIDs (committees, projects)
    • Preserves all data including SurveyMonkey answers

  3. Publish: Sends transformed data to two downstream services:

    • Indexer Service (lfx.index.survey, lfx.index.survey_response)

      • Enables search functionality
      • Includes parent references (committee, project)
      • Includes searchable tags (committee_uid:<uid>, project_uid:<uid> for surveys; additionally survey_uid:<uid> for responses)
      • Provides access control metadata

    • FGA-Sync Service (lfx.fga-sync.update_access, lfx.fga-sync.delete_access)

      • Updates Fine-Grained Authorization (FGA) tuples
      • Grants survey response owner to the respondent's LFX username
      • Links surveys to committees and projects

  4. Track: Records processed events in v1-mappings KV bucket for deduplication

Configuration

Environment Variables

VariableDefaultDescription
EVENT_PROCESSING_ENABLEDtrueEnable/disable event processing
EVENT_CONSUMER_NAMEsurvey-service-kv-consumerJetStream consumer name
EVENT_STREAM_NAMEKV_v1-objectsJetStream stream name
NATS_URLnats://nats:4222NATS server URL

Consumer Configuration

  • Delivery Policy: DeliverLastPerSubjectPolicy - Processes the latest version of each key
  • Ack Policy: AckExplicitPolicy - Requires explicit acknowledgment
  • Max Deliver: 3 - Retries transient failures up to 3 times
  • Ack Wait: 30s - Timeout before message redelivery
  • Max Ack Pending: 1000 - Maximum unacknowledged messages

Data Transformation

Survey Data

v1 Format (DynamoDB/KV):

  • All numeric fields stored as strings (e.g., "nps_value": "8")
  • v1 SFIDs for committees and projects
  • Committee array with per-committee statistics

v2 Format (Transformed):

  • Proper types (integers, booleans)
  • v2 UUIDs for committees and projects
  • Mapped via IDMapper service
  • Preserved committee array structure

Example Transformation:

// v1 Input
{
  "id": "survey-123",
  "nps_value": "8",
  "total_responses": "42",
  "committees": [{
    "committee_id": "a094V00000A1BcdQAF",  // v1 SFID
    "project_id": "a094V00000A1XyzQAF",     // v1 SFID
    "nps_value": "9"
  }]
}

// v2 Output
{
  "uid": "survey-123",
  "nps_value": 8,                           // Integer
  "total_responses": 42,                    // Integer
  "committees": [{
    "committee_uid": "550e8400-e29b-41d4-a716-446655440000",  // v2 UUID
    "project_uid": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",    // v2 UUID
    "nps_value": 9
  }]
}

Survey Response Data

v1 Format: Similar string-based fields, v1 references

v2 Format:

  • Proper types
  • Mapped project/committee/survey UIDs
  • Preserved SurveyMonkey question answers (no transformation)
  • username passes through unchanged for FGA owner on survey_response

Error Handling

Transient Errors (Retry)

These errors trigger NAK (negative acknowledgment) for automatic retry:

  • NATS connection timeouts
  • IDMapper service unavailable
  • Network failures
  • Temporary downstream service outages

Action: Message redelivered up to MaxDeliver times (3 attempts)

Permanent Errors (Skip)

These errors trigger ACK to skip and move on:

  • Invalid JSON structure
  • Missing required fields (e.g., empty id)
  • No parent references (survey/response orphaned)
  • Malformed data

Action: Log warning and continue processing other messages

ID Mapping Failures

When v1→v2 ID mapping fails:

  • Log warning with v1 ID
  • Skip setting v2 UID for that reference
  • Continue processing with remaining valid data
  • Survey/response may be skipped if critical references missing

Operations

Starting the Service

Event processing starts automatically when the service starts:

# Default (event processing enabled)
./survey-api

# Explicitly enable
EVENT_PROCESSING_ENABLED=true ./survey-api

# Disable event processing
EVENT_PROCESSING_ENABLED=false ./survey-api

Monitoring

Log Messages:

INFO  Event processing is ENABLED - initializing event processor
INFO  Event processor started in background
INFO  processing survey update key=itx-surveys:survey-123
INFO  successfully sent survey indexer and access messages survey_id=survey-123

Consumer Status:

# Check consumer status
nats consumer info KV_v1-objects survey-service-kv-consumer

Lifecycle

  1. Startup: Event processor initializes after IDMapper
  2. Running: Processes events in background goroutine
  3. Shutdown: Graceful shutdown sequence:
    • Context cancellation stops the consumer
    • Consumer drains pending messages
    • NATS connection closed
    • HTTP server shutdown

Troubleshooting

No events processing:

  • Check EVENT_PROCESSING_ENABLED=true
  • Verify NATS connection: NATS_URL
  • Check consumer exists: nats consumer ls KV_v1-objects

Events failing repeatedly:

  • Check logs for permanent errors
  • Verify IDMapper service is running
  • Confirm indexer and FGA-sync services are available

Duplicate processing:

  • Check v1-mappings KV bucket for tracking entries (live = 1, deleted = !del)
  • Verify consumer name is unique per instance

ID mapping failures:

  • Ensure IDMapper service has v1↔v2 mappings populated
  • Check project/committee references exist in v1 system

Deduplication

The service uses the v1-mappings KV bucket to track processed events:

Key Pattern:

  • Surveys: survey.<uid>
  • Responses: survey_response.<uid>

Value:

  • 1 — resource has been synced at least once (live mapping)
  • !del — resource was deleted (tombstone marker)

Logic:

  • If mapping missing or tombstoned → CREATE operation
  • If mapping exists (live) → UPDATE operation
  • After successful sync → store/update mapping with value 1
  • After successful delete → overwrite mapping with !del (tombstone)

Tombstone deduplication:

When a delete event is processed, the mapping key is set to !del rather than being removed. On any redelivery of the same delete message, the tombstone is detected and the event is safely skipped without re-publishing a delete to downstream services.

Soft delete support:

KV PUT operations that include a non-empty _sdc_deleted_at field are treated as soft deletes and follow the same delete path as hard KV DEL/PURGE operations.

Orphan prevention:

When processing a survey response, the handler checks that the parent survey mapping exists and is not tombstoned. If the parent mapping is missing or tombstoned (i.e., the survey was deleted), the response event is NAKed for retry rather than being synced to downstream services.

Performance Considerations

Concurrency:

  • Single consumer per service instance
  • Messages processed sequentially per consumer
  • Multiple service instances = parallel processing

Throughput:

  • MaxAckPending=1000 allows up to 1000 in-flight messages
  • Adjust based on processing speed and resource availability

Backpressure:

  • Consumer automatically pauses when MaxAckPending reached
  • Resumes when pending count drops

Resource Usage:

  • Event processor runs in background goroutine (low overhead)
  • NATS connection shared with IDMapper
  • Memory footprint minimal (streaming model)

IDMapper Service

  • Maps v1 SFIDs ↔ v2 UUIDs
  • Required for event processing
  • Queries via NATS request-reply pattern

Indexer Service

  • Receives transformed survey/response data
  • Indexes in OpenSearch for search functionality
  • Handles ActionCreated, ActionUpdated, ActionDeleted

FGA-Sync Service

  • Receives access control updates
  • Manages OpenFGA authorization tuples
  • Sets survey_response:{uid}:owner to the respondent's LFX username
  • Links resources to parent entities (committees, projects, surveys)

Development

Testing Event Processing

  1. Disable in local development:

    export EVENT_PROCESSING_ENABLED=false

  2. Watch consumer activity:

    nats consumer next KV_v1-objects survey-service-kv-consumer --count 10

  3. Trigger test event:

    # Put test survey in v1-objects KV
nats kv put v1-objects itx-surveys:test-123 '{"id":"test-123",...}'

  4. Check processing logs:

    # Look for processing messages
grep "processing survey update" logs/survey-api.log

Code Structure

cmd/survey-api/eventing/
├── event_processor.go           # Lifecycle management
├── kv_handler.go                # Event routing by key prefix
├── survey_event_handler.go      # Survey transformation logic
└── survey_response_event_handler.go  # Response transformation logic

internal/domain/
├── event_models.go              # v2 data models
└── event_publisher.go           # Publisher interface

internal/infrastructure/eventing/
├── event_config.go              # Configuration structs
└── nats_publisher.go            # NATS publishing implementation

Adding New Event Types

To add processing for new entity types:

  1. Create handler in cmd/survey-api/eventing/{entity}_event_handler.go
  2. Add routing logic in kv_handler.go
  3. Define v2 model in internal/domain/event_models.go
  4. Add publisher method in internal/infrastructure/eventing/nats_publisher.go
  5. Update documentation

References