Merge branch 'main' into jme/LFXV2-1371

Author: jordane

README.md

@@ -4,50 +4,13 @@ This repository contains tools and services for synchronizing data between LFX v
 
 ## Overview
 
-Most data entities are synced from v1 into native LFX One entities. Bidirectional sync is implemented for committees and committee members.
+This repository serves two distinct purposes:
 
-However, due to the size, complexity, and number of external interactions the LFX Meetings stack has, v1 and v2 meetings will be kept separate, though v1 meetings will be made avaliable as read-only, natively-permissioned entities within LFX One via the query service.
+1. **Real-time streaming replication.** PostgreSQL WAL events (via `wal-listener`) and DynamoDB Streams are replicated in real time—alongside periodic Meltano backfills—into a `v1-objects` NATS KV bucket. LFX One wrapper services subscribe to this bucket to drive indexing pipelines (OpenSearch via the indexer service) and access-control pipelines (OpenFGA via fga-sync), without needing to integrate directly with ITX eventing.
 
-```mermaid
-flowchart TD
-    V1[LFX v1 Meetings] --> Sync[Data Sync Process]
-    Projects --> Sync2[Data Backfill]
-    Committees --> Sync3[Data Backfill]
-    Sync --> ShadowV1[**v1 Meetings**<br/>- Synced from v1<br/>- Read-only in LFX One<br/>- Separate from native v2]
-    Sync2 --> ProjectsV2
-    Sync3 --> CommitteesV2
-
-    NativeV2[**Native v2 Meetings**<br/>- Created directly in v2<br/>- Full CRUD operations]
-    ProjectsV2[Native v2 Projects]
-    CommitteesV2[Native v2 Committees]
-
-    ShadowV1 --> LFXOne[LFX One UI]
-    NativeV2 --> LFXOne
-    ProjectsV2 & CommitteesV2 --> LFXOne
-
-    LFXOne --> Search[Search & Query<br/>Services]
-    LFXOne --> FGA[OpenFGA<br/>Access Control]
-    LFXOne --> JoinFlow[Meeting Join Flow]
-
-    subgraph "LFX One Platform"
-        Search
-        FGA
-        JoinFlow
-    end
-
-    subgraph "v1 Data"
-        V1
-        Projects
-        Committees
-    end
+2. **Bidirectional sync for "core" resources.** Projects and committees are fully synced in both directions between LFX v1 and LFX One. This gives LFX One a self-contained stack for these entity types, which simplifies developer environment stand-up by removing the dependency on the highly-interconnected LFX/Salesforce/ITX stack.
 
-    subgraph "v2 Data"
-        ShadowV1
-        NativeV2
-        ProjectsV2
-        CommitteesV2
-    end
-```
+ITX-hosted resources such as Meetings are handled by v2 "wrapper" services that sit in front of the ITX APIs and rely on the NATS KV replication above for eventing; they do **not** get their own native v2 entity storage. See the [ITX wrappers component diagram](#itx-wrappers-component-diagram) in the Architecture Diagrams section for how this fits together.
 
 ## Prerequisites
 
@@ -63,14 +26,23 @@ Please see each component for further setup instructions.
 This repository contains three main components:
 
 ### [Meltano](./meltano/README.md)
+
 Data extraction and loading pipeline that extracts data from LFX v1 sources (DynamoDB for meetings, PostgreSQL for projects/committees) and loads it into NATS KV stores for processing by the v2 platform.
 
 ### [v1-sync-helper](./cmd/lfx-v1-sync-helper/README.md)
+
 Go service that monitors NATS KV stores for replicated v1 data and synchronizes it with the LFX v2 platform APIs, handling data transformation and conflict resolution.
 
 ### [Helm charts](./charts/lfx-v1-sync-helper/README.md)
+
 Kubernetes deployment manifests for the custom app service and WAL listener component, providing scalable deployment options for production environments.
 
+## Research & guides
+
+- [Adding a new DynamoDB table](./research/adding-dynamodb-table.md) — step-by-step checklist for onboarding a new DynamoDB table into the Meltano pipeline and stream consumer, with a worked example.
+- [Updating the Meltano catalog ConfigMap](./research/updating-meltano-catalog.md) — how to regenerate and apply the schema cache when tables or columns change.
+- [Meetings v1 vs v2](./research/meetings-v1-vs-v2.md) — comparison of the v1 and v2 meetings data models.
+
 ## NATS API
 
 The v1-sync-helper service provides a NATS request/reply function for querying v1-v2 ID mappings.
@@ -86,12 +58,14 @@ The v1-sync-helper service provides a NATS request/reply function for querying v
 Send a NATS request to `lfx.lookup_v1_mapping` with the mapping key as the payload. The service will respond with the corresponding mapping value or an error.
 
 **Request Format:**
+
 ```
 Subject: lfx.lookup_v1_mapping
 Payload: <mapping_key>
 ```
 
 **Response Format:**
+
 - **Success**: The mapped value as a string
 - **Not Found**: Empty string (`""`)
 - **Error**: String prefixed with `"error: "` (e.g., `"error: connection timeout"`)
@@ -116,11 +90,78 @@ The following table shows the supported mapping key patterns and their expected
 
 ## Architecture Diagrams
 
-Regarding the following diagrams:
+Regarding the following sequence diagrams:
+
+- "Projects API" is representative of the core resources that have bidirectional sync (projects, committees). ITX-hosted resources such as Meetings are handled by wrapper services that subscribe to the NATS KV bucket instead—see the component diagram below.
 
-- The DynamoDB source (incremental or realtime) is not currently included in the diagrams.
-- The planned bidirectional sync (LFX One changes back to v1) is included in the diagrams.
-- "Projects API" is representative of most data entities. However, v1 Meetings push straight to OpenSearch and OpenFGA (via platform services)—this is not shown.
+### ITX wrappers component diagram
+
+This diagram shows how the LFX One platform, the v1-sync-helper replication pipeline, and ITX-hosted services fit together at the component level.
+
+```mermaid
+flowchart TD
+    %%{init: {'flowchart': {'defaultRenderer': 'elk' }}}%%
+
+    user[User]
+
+    subgraph lfxv2["LFX Platform (k8s)"]
+        traefik[Traefik]
+        heimdall[Heimdall]
+        subgraph fga-sync
+        fga-sync-update-access[update-access]
+        fga-sync-access-check[access-check]
+        end
+        indexer
+        query-svc[Query Service]
+        opensearch[OpenSearch]
+        openfga[OpenFGA]
+
+        xyz-wrapper@{ shape: processes, label: "Entity services (wrappers)" }
+
+        traefik -.->|calls authz middleware| heimdall
+        traefik --->|"proxies all list (search) requests to"| query-svc
+        heimdall -.->|checks relations via| openfga
+        query-svc -->|queries from| opensearch
+        query-svc -.->|checks access via NATS| fga-sync-access-check
+        indexer -.->|stores to| opensearch
+        fga-sync-update-access -.->|syncs relations to| openfga
+        fga-sync-access-check -.->|checks access via| openfga
+
+        traefik -->|proxies authorized resource create/get/put requests to| xyz-wrapper
+
+        xyz-wrapper -.->|upsert via NATS| indexer
+        xyz-wrapper -.->|push relations via NATS| fga-sync-update-access
+
+        %%wal-listener
+        v1-sync-helper
+        v1-objects[(v1 replica<br />KV bucket)]
+        %%wal-listener -.->|NATS stream| v1-sync-helper
+        v1-sync-helper -.->|NATS KV operations| v1-objects
+
+        v1-objects -.->|subscribes to bucket events via NATS| xyz-wrapper
+    end
+
+    subgraph itx-aws[ITX AWS]
+        itx-api-gw[API Gateway]
+        itx-svc-authz[Authorizer Lambda]
+        itx-service-xyz@{ shape: processes, label: "ITX services (Lambdas)"}
+        dynamodb[(DynamoDB)]
+
+        itx-api-gw -.-> itx-svc-authz
+        itx-api-gw --> itx-service-xyz
+        itx-service-xyz --> dynamodb
+    end
+
+    third-party-svcs@{ shape: processes, label: "Third-party services (Zoom, etc)"}
+    itx-service-xyz --> third-party-svcs
+
+    xyz-wrapper -->|authorized<br />create/get/put| itx-api-gw
+
+    dynamodb -.->|consumed by streams| v1-sync-helper
+
+    user -->|old| PIS[PIS or User Service] -->|authorized create/get/put/list| itx-api-gw
+    user -->|new| traefik
+```
 
 ### Data extraction/replication sequence diagram
 
@@ -129,6 +170,8 @@ sequenceDiagram
     participant lfx_v1 as LFX v1 API
     participant postgres as Platform Database<br/>(PostgreSQL)
     participant wal-listener
+    participant dynamodb as DynamoDB
+    participant dynamo-stream as dynamodb-stream-consumer
     participant meltano as Meltano<br/>(custom NATS<br/>exporter)
     participant v1_kv as "v1" NATS KV bucket
     participant v1-sync-helper
@@ -140,13 +183,20 @@ sequenceDiagram
     wal-listener-)+v1-sync-helper: notification on "wal-listener" subject
     deactivate wal-listener
     v1-sync-helper-)-v1_kv: store record (or soft-deletion) by v1 ID
+    lfx_v1 ->> dynamodb: create/update/delete (via ITX API)
+    dynamodb-)+dynamo-stream: DynamoDB Streams event
+    dynamo-stream-)+v1-sync-helper: notification on "dynamodb_streams" subject
+    deactivate dynamo-stream
+    v1-sync-helper-)-v1_kv: store record (or soft-deletion) by v1 ID
 
     Note over lfx_v1,v1_kv: Data backfill (full sync & incremental gap-fill)
     meltano->>meltano: scheduled task invoke (weekly/monthly)
     activate meltano
     meltano->>meltano: load state from S3<br/>(incremental state bookmark)
     meltano->>+postgres: query records >= LAST_SYNC<br/>(full re-sync also supported)
     postgres--)-meltano: results
+    meltano->>+dynamodb: Scan tables >= LAST_MONTH<br/>(full re-scan also supported)
+    dynamodb--)-meltano: results
     loop for each record
     meltano->>+v1_kv: fetch KV item by v1 ID
     v1_kv--)-meltano: KV item, soft-deletion, or empty
@@ -274,6 +324,8 @@ sequenceDiagram
     participant lfx_v1 as LFX v1 API
     participant postgres as Platform Database<br/>(PostgreSQL)
     participant wal-listener
+    participant dynamodb as DynamoDB
+    participant dynamo-stream as dynamodb-stream-consumer
     participant meltano as Meltano<br/>(custom NATS<br/>exporter)
     participant v1_kv as "v1" NATS KV bucket
     participant v1-sync-helper
@@ -290,13 +342,20 @@ sequenceDiagram
     wal-listener-)+v1-sync-helper: notification on "wal-listener" subject
     deactivate wal-listener
     v1-sync-helper-)-v1_kv: store record (or soft-deletion) by v1 ID
+    lfx_v1 ->> dynamodb: create/update/delete (via ITX API)
+    dynamodb-)+dynamo-stream: DynamoDB Streams event
+    dynamo-stream-)+v1-sync-helper: notification on "dynamodb_streams" subject
+    deactivate dynamo-stream
+    v1-sync-helper-)-v1_kv: store record (or soft-deletion) by v1 ID
 
     Note over lfx_v1,v1_kv: Data backfill (full sync & incremental gap-fill)
     meltano->>meltano: scheduled task invoke (weekly/monthly)
     activate meltano
     meltano->>meltano: load state from S3<br/>(incremental state bookmark)
     meltano->>+postgres: query records >= LAST_SYNC<br/>(full re-sync also supported)
     postgres--)-meltano: results
+    meltano->>+dynamodb: Scan tables >= LAST_MONTH<br/>(full re-scan also supported)
+    dynamodb--)-meltano: results
     loop for each record
     meltano->>+v1_kv: fetch KV item by v1 ID
     v1_kv--)-meltano: KV item, soft-deletion, or empty

charts/lfx-v1-sync-helper/values.yaml

@@ -252,7 +252,7 @@ dynamodbStreamConsumer:
     # DYNAMODB_TABLES is a comma-separated list of DynamoDB table names to consume.
     # Defaults to the full set of tables used by the tap-dynamodb Meltano extractor.
     DYNAMODB_TABLES:
-      value: "itx-poll,itx-poll-vote,itx-surveys,itx-survey-responses,itx-zoom-meetings-mappings-v2,itx-zoom-meetings-v2,itx-zoom-past-meetings-mappings,itx-zoom-past-meetings,itx-zoom-past-meetings-attendees,itx-zoom-past-meetings-invitees,itx-zoom-past-meetings-recordings,itx-zoom-past-meetings-summaries,itx-zoom-meetings-registrants-v2,itx-zoom-meetings-invite-responses-v2,itx-zoom-meetings-attachments-v2,itx-zoom-past-meetings-attachments,itx-groupsio-v2-service,itx-groupsio-v2-subgroup,itx-groupsio-v2-member"
+      value: "itx-poll,itx-poll-vote,itx-surveys,itx-survey-responses,surveymonkey-surveys,itx-zoom-meetings-mappings-v2,itx-zoom-meetings-v2,itx-zoom-past-meetings-mappings,itx-zoom-past-meetings,itx-zoom-past-meetings-attendees,itx-zoom-past-meetings-invitees,itx-zoom-past-meetings-recordings,itx-zoom-past-meetings-summaries,itx-zoom-meetings-registrants-v2,itx-zoom-meetings-invite-responses-v2,itx-zoom-meetings-attachments-v2,itx-zoom-past-meetings-attachments,itx-groupsio-v2-service,itx-groupsio-v2-subgroup,itx-groupsio-v2-member"
     # START_FROM_LATEST controls the iterator start position for new shards with no checkpoint.
     # Set to "true" to only receive new records; "false" (default) replays all available records.
     START_FROM_LATEST:

cmd/lfx-v1-sync-helper/handlers.go

@@ -107,7 +107,7 @@ func handleKVPut(ctx context.Context, entry jetstream.KeyValueEntry) bool {
 		// Voting records are handled by lfx-v2-voting-service.
 		logger.With("key", key).DebugContext(ctx, "voting record, handled by lfx-v2-voting-service")
 		return false
-	case "itx-surveys", "itx-survey-responses":
+	case "itx-surveys", "itx-survey-responses", "surveymonkey-surveys":
 		// Survey records are handled by lfx-v2-survey-service.
 		logger.With("key", key).DebugContext(ctx, "survey record, handled by lfx-v2-survey-service")
 		return false
@@ -254,7 +254,7 @@ func handleResourceDelete(ctx context.Context, key string, v1Principal string, v
 		// Voting records are handled by lfx-v2-voting-service.
 		logger.With("key", key).DebugContext(ctx, "voting record deleted, handled by lfx-v2-voting-service")
 		return false
-	case "itx-surveys", "itx-survey-responses":
+	case "itx-surveys", "itx-survey-responses", "surveymonkey-surveys":
 		// Survey records are handled by lfx-v2-survey-service.
 		logger.With("key", key).DebugContext(ctx, "survey record deleted, handled by lfx-v2-survey-service")
 		return false

cmd/lfx-v1-sync-helper/handlers_meetings.go

@@ -1285,12 +1285,6 @@ func handleZoomPastMeetingUpdate(ctx context.Context, key string, v1Data map[str
 		indexerAction = MessageActionUpdated
 	}
 
-	tags := getPastMeetingTags(pastMeeting)
-	if err := sendIndexerMessage(ctx, IndexV1PastMeetingSubject, indexerAction, pastMeeting, tags); err != nil {
-		funcLogger.With(errKey, err).ErrorContext(ctx, "failed to send past meeting indexer message")
-		return
-	}
-
 	// Try to get committee mappings from the index first
 	var committees []string
 	pastMeeting.Committees = []Committee{}
@@ -1339,6 +1333,12 @@ func handleZoomPastMeetingUpdate(ctx context.Context, key string, v1Data map[str
 		}
 	}
 
+	tags := getPastMeetingTags(pastMeeting)
+	if err := sendIndexerMessage(ctx, IndexV1PastMeetingSubject, indexerAction, pastMeeting, tags); err != nil {
+		funcLogger.With(errKey, err).ErrorContext(ctx, "failed to send past meeting indexer message")
+		return
+	}
+
 	accessMsg := PastMeetingAccessMessage{
 		UID:        uid,
 		MeetingUID: pastMeeting.MeetingID,

cmd/lfx-v1-sync-helper/ingest_dynamodb.go

@@ -265,20 +265,17 @@ func dynamodbKVKey(tableName string, keys map[string]interface{}) string {
 	return tableName + "." + strings.Join(parts, "#")
 }
 
-// shouldDynamoDBUpdate returns true when the incoming new image should overwrite
-// the existing KV entry. It compares modified_at timestamps when both are present,
-// falling back to last_modified_at for tables that use that field instead (e.g.
-// Groups.io tables). If either timestamp is missing or unparseable the write
-// proceeds (stream events are treated as authoritative).
+// knownTimestampFields is the ordered list of timestamp field names checked when
+// comparing records. Add new field names here when onboarding tables that use a
+// different timestamp field.
+var knownTimestampFields = []string{"modified_at", "last_modified_at", "date_modified"}
+
+// shouldDynamoDBUpdate returns true when the new image should overwrite the
+// existing KV entry. If either record has no recognisable timestamp the write
+// proceeds; stream events are treated as authoritative.
 func shouldDynamoDBUpdate(ctx context.Context, newData, existingData map[string]interface{}, key string) bool {
-	newModifiedAt := getTimestampString(newData, "modified_at")
-	if newModifiedAt == "" {
-		newModifiedAt = getTimestampString(newData, "last_modified_at")
-	}
-	existingModifiedAt := getTimestampString(existingData, "modified_at")
-	if existingModifiedAt == "" {
-		existingModifiedAt = getTimestampString(existingData, "last_modified_at")
-	}
+	newModifiedAt := firstTimestamp(newData, knownTimestampFields...)
+	existingModifiedAt := firstTimestamp(existingData, knownTimestampFields...)
 
 	if newModifiedAt == "" || existingModifiedAt == "" {
 		return true
@@ -295,3 +292,14 @@ func shouldDynamoDBUpdate(ctx context.Context, newData, existingData map[string]
 
 	return newTime.After(existingTime)
 }
+
+// firstTimestamp returns the first non-empty timestamp string found in data
+// for the given field names, or "" if none are present.
+func firstTimestamp(data map[string]interface{}, fields ...string) string {
+	for _, f := range fields {
+		if v := getTimestampString(data, f); v != "" {
+			return v
+		}
+	}
+	return ""
+}