Merge branch 'main' into andrest50/unmarshal-json

Author: andrest50

cmd/dynamodb-stream-consumer/publisher.go

@@ -18,11 +18,11 @@ import (
 
 // DynamoDBStreamEvent is the event payload published to NATS for each DynamoDB stream record.
 type DynamoDBStreamEvent struct {
-	EventID                 string                 `json:"event_id"`
-	EventName               string                 `json:"event_name"` // INSERT, MODIFY, REMOVE
-	TableName               string                 `json:"table_name"`
-	SequenceNumber          string                 `json:"sequence_number"`
-	ApproximateCreationTime time.Time              `json:"approximate_creation_time"`
+	EventID                 string    `json:"event_id"`
+	EventName               string    `json:"event_name"` // INSERT, MODIFY, REMOVE
+	TableName               string    `json:"table_name"`
+	SequenceNumber          string    `json:"sequence_number"`
+	ApproximateCreationTime time.Time `json:"approximate_creation_time"`
 	// Keys contains only the primary key attribute(s) of the item (partition key +
 	// optional sort key). Consumers can use this to construct a stable record
 	// identifier without needing to know the full item schema.

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

@@ -130,9 +130,9 @@ func LoadConfig() (*Config, error) {
 		Auth0ClientID:   os.Getenv("AUTH0_CLIENT_ID"),
 		Auth0PrivateKey: os.Getenv("AUTH0_PRIVATE_KEY"),
 		// Other configuration
-		NATSURL:    os.Getenv("NATS_URL"),
-		Port:       os.Getenv("PORT"),
-		Bind:       os.Getenv("BIND"),
+		NATSURL:               os.Getenv("NATS_URL"),
+		Port:                  os.Getenv("PORT"),
+		Bind:                  os.Getenv("BIND"),
 		Debug:                 parseBooleanEnv("DEBUG"),
 		HTTPDebug:             parseBooleanEnv("HTTP_DEBUG"),
 		UseMsgpack:            parseBooleanEnv("USE_MSGPACK"),

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

@@ -416,18 +416,15 @@ func handleZoomMeetingMappingUpdate(ctx context.Context, key string, v1Data map[
 		return
 	}
 
-	// Fetch the meeting object from v1-objects KV bucket
+	// Fetch and parse the meeting data.
 	meetingKey := fmt.Sprintf("itx-zoom-meetings-v2.%s", meetingID)
-	meetingEntry, err := v1KV.Get(ctx, meetingKey)
+	meetingData, exists, err := getV1ObjectData(ctx, meetingKey)
 	if err != nil {
-		funcLogger.With(errKey, err, "meeting_id", meetingID).WarnContext(ctx, "failed to fetch meeting from KV bucket, cannot trigger re-index")
+		funcLogger.With(errKey, err, "meeting_id", meetingID).ErrorContext(ctx, "failed to get meeting data from KV bucket")
 		return
 	}
-
-	// Parse the meeting data
-	var meetingData map[string]any
-	if err := json.Unmarshal(meetingEntry.Value(), &meetingData); err != nil {
-		funcLogger.With(errKey, err, "meeting_id", meetingID).ErrorContext(ctx, "failed to unmarshal meeting data")
+	if !exists {
+		funcLogger.With("meeting_id", meetingID).WarnContext(ctx, "meeting data not found or deleted in KV bucket")
 		return
 	}
 
@@ -1073,18 +1070,15 @@ func handleZoomPastMeetingMappingUpdate(ctx context.Context, key string, v1Data
 		return
 	}
 
-	// Fetch the past meeting object from v1-objects KV bucket
+	// Fetch and parse the past meeting data.
 	pastMeetingKey := fmt.Sprintf("itx-zoom-past-meetings.%s", meetingAndOccurrenceID)
-	pastMeetingEntry, err := v1KV.Get(ctx, pastMeetingKey)
+	pastMeetingData, exists, err := getV1ObjectData(ctx, pastMeetingKey)
 	if err != nil {
-		funcLogger.With(errKey, err).WarnContext(ctx, "failed to fetch past meeting from KV bucket, cannot trigger re-index")
+		funcLogger.With(errKey, err).ErrorContext(ctx, "failed to get past meeting data from KV bucket")
 		return
 	}
-
-	// Parse the past meeting data
-	var pastMeetingData map[string]any
-	if err := json.Unmarshal(pastMeetingEntry.Value(), &pastMeetingData); err != nil {
-		funcLogger.With(errKey, err).ErrorContext(ctx, "failed to unmarshal past meeting data")
+	if !exists {
+		funcLogger.WarnContext(ctx, "past meeting data not found or deleted in KV bucket")
 		return
 	}
 
@@ -1262,22 +1256,16 @@ func handleZoomPastMeetingInviteeUpdate(ctx context.Context, key string, v1Data
 	isHost := false
 	registrantID := invitee.RegistrantID
 	if registrantID != "" {
-		// Look up the registrant in the v1-objects KV bucket
+		// Look up the registrant in the v1-objects KV bucket.
 		registrantKey := fmt.Sprintf("itx-zoom-meetings-registrants-v2.%s", registrantID)
-		registrantEntry, err := v1KV.Get(ctx, registrantKey)
-		if err == nil && registrantEntry != nil {
-			// Parse the registrant data
-			var registrantData map[string]any
-			if err := json.Unmarshal(registrantEntry.Value(), &registrantData); err == nil {
-				// Check if the registrant has the host field set to true
-				if hostValue, ok := registrantData["host"].(bool); ok {
-					isHost = hostValue
-				}
-			} else {
-				funcLogger.With(errKey, err, "registrant_id", registrantID).WarnContext(ctx, "failed to unmarshal registrant data")
+		registrantData, exists, err := getV1ObjectData(ctx, registrantKey)
+		if err != nil {
+			funcLogger.With(errKey, err, "registrant_id", registrantID).WarnContext(ctx, "failed to get registrant data")
+		} else if exists {
+			// Check if the registrant has the host field set to true
+			if hostValue, ok := registrantData["host"].(bool); ok {
+				isHost = hostValue
 			}
-		} else {
-			funcLogger.With(errKey, err, "registrant_id", registrantID).WarnContext(ctx, "failed to fetch registrant from KV bucket")
 		}
 	}
 
@@ -1457,23 +1445,17 @@ func handleZoomPastMeetingAttendeeUpdate(ctx context.Context, key string, v1Data
 	isRegistrant := false
 	registrantID := attendee.RegistrantID
 	if registrantID != "" {
-		// Look up the registrant in the v1-objects KV bucket
+		// Look up the registrant in the v1-objects KV bucket.
 		registrantKey := fmt.Sprintf("itx-zoom-meetings-registrants-v2.%s", registrantID)
-		registrantEntry, err := v1KV.Get(ctx, registrantKey)
-		if err == nil && registrantEntry != nil {
+		registrantData, exists, err := getV1ObjectData(ctx, registrantKey)
+		if err != nil {
+			funcLogger.With(errKey, err, "registrant_id", registrantID).WarnContext(ctx, "failed to get registrant data")
+		} else if exists {
 			isRegistrant = true
-			// Parse the registrant data
-			var registrantData map[string]any
-			if err := json.Unmarshal(registrantEntry.Value(), &registrantData); err == nil {
-				// Check if the registrant has the host field set to true
-				if hostValue, ok := registrantData["host"].(bool); ok {
-					isHost = hostValue
-				}
-			} else {
-				funcLogger.With(errKey, err, "registrant_id", registrantID).WarnContext(ctx, "failed to unmarshal registrant data")
+			// Check if the registrant has the host field set to true
+			if hostValue, ok := registrantData["host"].(bool); ok {
+				isHost = hostValue
 			}
-		} else {
-			funcLogger.With(errKey, err, "registrant_id", registrantID).WarnContext(ctx, "failed to fetch registrant from KV bucket")
 		}
 	}
 
@@ -2011,18 +1993,13 @@ func handleZoomPastMeetingSummaryUpdate(ctx context.Context, key string, v1Data
 	aiSummaryAccess := ""
 	if summaryInput.PastMeetingUID != "" {
 		pastMeetingKey := fmt.Sprintf("itx-zoom-past-meetings.%s", summaryInput.PastMeetingUID)
-		pastMeetingEntry, err := v1KV.Get(ctx, pastMeetingKey)
-		if err == nil && pastMeetingEntry != nil {
-			var pastMeetingData map[string]any
-			if err := json.Unmarshal(pastMeetingEntry.Value(), &pastMeetingData); err == nil {
-				if aiSummaryAccessValue, ok := pastMeetingData["ai_summary_access"].(string); ok && aiSummaryAccessValue != "" {
-					aiSummaryAccess = aiSummaryAccessValue
-				}
-			} else {
-				funcLogger.With(errKey, err).WarnContext(ctx, "failed to unmarshal past meeting data")
+		pastMeetingData, exists, err := getV1ObjectData(ctx, pastMeetingKey)
+		if err != nil {
+			funcLogger.With(errKey, err).WarnContext(ctx, "failed to get past meeting data")
+		} else if exists {
+			if aiSummaryAccessValue, ok := pastMeetingData["ai_summary_access"].(string); ok && aiSummaryAccessValue != "" {
+				aiSummaryAccess = aiSummaryAccessValue
 			}
-		} else {
-			funcLogger.With(errKey, err).WarnContext(ctx, "failed to fetch past meeting from KV bucket")
 		}
 	}
 

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

@@ -138,33 +138,12 @@ func lookupV1User(ctx context.Context, platformID string) (*V1User, error) {
 	// Look up user in the salesforce-merged_user table via v1-objects KV bucket
 	userKey := fmt.Sprintf("salesforce-merged_user.%s", platformID)
 
-	entry, err := v1KV.Get(ctx, userKey)
+	userData, exists, err := getV1ObjectData(ctx, userKey)
 	if err != nil {
-		if err == jetstream.ErrKeyNotFound {
-			return nil, fmt.Errorf("user %s not found in v1-objects KV bucket", platformID)
-		}
-		return nil, fmt.Errorf("failed to get user from v1-objects KV bucket: %w", err)
-	}
-
-	// Parse the merged_user data
-	var userData map[string]any
-	if err := json.Unmarshal(entry.Value(), &userData); err != nil {
-		// Try msgpack if JSON fails.
-		if msgpackErr := msgpack.Unmarshal(entry.Value(), &userData); msgpackErr != nil {
-			return nil, fmt.Errorf("failed to unmarshal user data (json: %w, msgpack: %w)", err, msgpackErr)
-		}
+		return nil, fmt.Errorf("failed to get user data: %w", err)
 	}
-
-	// Check if the user record is deleted
-	if isDeleted, ok := userData["isdeleted"].(bool); ok && isDeleted {
-		return nil, fmt.Errorf("user %s is deleted (isdeleted)", platformID)
-	}
-
-	// Also treat WAL-based soft deletes (indicated by _sdc_deleted_at) as deleted.
-	if deletedAt, ok := userData["_sdc_deleted_at"]; ok {
-		if s, okStr := deletedAt.(string); (okStr && strings.TrimSpace(s) != "") || (!okStr && deletedAt != nil) {
-			return nil, fmt.Errorf("user %s is deleted (_sdc_deleted_at))", platformID)
-		}
+	if !exists {
+		return nil, fmt.Errorf("user %s not found or is deleted in v1-objects KV bucket", platformID)
 	}
 
 	// Extract user fields from the merged_user record
@@ -264,42 +243,15 @@ func getPrimaryEmailForUser(ctx context.Context, userSfid string) (string, error
 func getAlternateEmailDetails(ctx context.Context, emailSfid string) (email string, isPrimary bool, isTombstoned bool, err error) {
 	emailKey := fmt.Sprintf("salesforce-alternate_email__c.%s", emailSfid)
 
-	entry, err := v1KV.Get(ctx, emailKey)
+	// Parse the alternate email record.
+	emailData, exists, err := getV1ObjectData(ctx, emailKey)
 	if err != nil {
-		if err == jetstream.ErrKeyNotFound || err == jetstream.ErrKeyDeleted {
-			// Key not found or deleted could mean it was tombstoned/deleted
-			return "", false, true, nil
-		}
-		return "", false, false, fmt.Errorf("failed to get email record %s from v1-objects: %w", emailSfid, err)
-	}
-
-	// Check if this is a tombstone marker
-	if isTombstonedMapping(entry.Value()) {
-		return "", false, true, nil
-	}
-
-	// Parse the alternate email record
-	var emailData map[string]any
-	if err := json.Unmarshal(entry.Value(), &emailData); err != nil {
-		// Try msgpack if JSON fails.
-		if msgpackErr := msgpack.Unmarshal(entry.Value(), &emailData); msgpackErr != nil {
-			return "", false, false, fmt.Errorf("failed to unmarshal email data (json: %w, msgpack: %w)", err, msgpackErr)
-		}
+		return "", false, false, fmt.Errorf("failed to get email data: %w", err)
 	}
-
-	// Check if the email record is deleted.
-	if isDeleted, ok := emailData["isdeleted"].(bool); ok && isDeleted {
+	if !exists {
 		return "", false, true, nil
 	}
 
-	// Also check for WAL-based soft deletes (indicated by _sdc_deleted_at).
-	// This is expected when soft-deleted email records are preserved in v1-objects KV bucket.
-	if deletedAt, ok := emailData["_sdc_deleted_at"]; ok {
-		if s, okStr := deletedAt.(string); (okStr && strings.TrimSpace(s) != "") || (!okStr && deletedAt != nil) {
-			return "", false, true, nil
-		}
-	}
-
 	// Also check if the email is inactive (active__c is not true).
 	if isActive, ok := emailData["active__c"].(bool); !ok || !isActive {
 		return "", false, true, nil
@@ -563,3 +515,45 @@ func parseWebsiteURL(website string) string {
 
 	return ""
 }
+
+// getV1ObjectData retrieves and unmarshals data from the v1-objects KV bucket with dual-format support.
+// It attempts JSON decoding first, then falls back to msgpack if JSON fails.
+// Returns (data, exists, error) where exists indicates if the record exists and is not deleted/tombstoned.
+// This abstraction should be used for all v1-objects bucket reads to ensure consistent
+// dual-format handling across the codebase.
+func getV1ObjectData(ctx context.Context, key string) (map[string]any, bool, error) {
+	entry, err := v1KV.Get(ctx, key)
+	if err != nil {
+		if err == jetstream.ErrKeyNotFound || err == jetstream.ErrKeyDeleted {
+			return nil, false, nil
+		}
+		return nil, false, fmt.Errorf("failed to get data from v1-objects KV bucket: %w", err)
+	}
+
+	// Check if this is a tombstone marker.
+	if isTombstonedMapping(entry.Value()) {
+		return nil, false, nil
+	}
+
+	var data map[string]any
+	if err := json.Unmarshal(entry.Value(), &data); err != nil {
+		// Try msgpack if JSON fails.
+		if msgpackErr := msgpack.Unmarshal(entry.Value(), &data); msgpackErr != nil {
+			return nil, false, fmt.Errorf("failed to unmarshal data (json: %w, msgpack: %w)", err, msgpackErr)
+		}
+	}
+
+	// Check if the record is deleted.
+	if isDeleted, ok := data["isdeleted"].(bool); ok && isDeleted {
+		return nil, false, nil
+	}
+
+	// Also check for WAL-based soft deletes (indicated by _sdc_deleted_at).
+	if deletedAt, ok := data["_sdc_deleted_at"]; ok {
+		if s, okStr := deletedAt.(string); (okStr && strings.TrimSpace(s) != "") || (!okStr && deletedAt != nil) {
+			return nil, false, nil
+		}
+	}
+
+	return data, true, nil
+}

go.mod

@@ -6,24 +6,24 @@ go 1.25.6
 
 require (
 	github.com/akamensky/base58 v0.0.0-20210829145138-ce8bf8802e8f
-	github.com/auth0/go-auth0 v1.32.1
+	github.com/auth0/go-auth0 v1.34.0
 	github.com/aws/aws-sdk-go-v2 v1.41.1
 	github.com/aws/aws-sdk-go-v2/config v1.32.9
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.9
 	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.55.0
 	github.com/aws/aws-sdk-go-v2/service/dynamodbstreams v1.32.10
 	github.com/aws/aws-sdk-go-v2/service/sts v1.41.6
-	github.com/golang-jwt/jwt/v5 v5.3.0
+	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/google/uuid v1.6.0
 	github.com/linuxfoundation/lfx-v2-committee-service v0.2.19
 	github.com/linuxfoundation/lfx-v2-indexer-service v0.4.14
-	github.com/linuxfoundation/lfx-v2-project-service v0.5.4
+	github.com/linuxfoundation/lfx-v2-project-service v0.5.6
 	github.com/nats-io/nats.go v1.48.0
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/teambition/rrule-go v1.8.2
 	github.com/vmihailenco/msgpack/v5 v5.4.1
-	goa.design/goa/v3 v3.24.1
-	golang.org/x/oauth2 v0.34.0
+	goa.design/goa/v3 v3.25.3
+	golang.org/x/oauth2 v0.35.0
 )
 
 require (
@@ -41,22 +41,22 @@ require (
 	github.com/aws/smithy-go v1.24.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
-	github.com/go-chi/chi/v5 v5.2.4 // indirect
+	github.com/go-chi/chi/v5 v5.2.5 // indirect
 	github.com/goccy/go-json v0.10.5 // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
-	github.com/klauspost/compress v1.18.3 // indirect
+	github.com/klauspost/compress v1.18.4 // indirect
 	github.com/lestrrat-go/blackmagic v1.0.4 // indirect
 	github.com/lestrrat-go/httpcc v1.0.1 // indirect
 	github.com/lestrrat-go/httprc v1.0.6 // indirect
 	github.com/lestrrat-go/iter v1.0.2 // indirect
 	github.com/lestrrat-go/jwx/v2 v2.1.6 // indirect
 	github.com/lestrrat-go/option v1.0.1 // indirect
-	github.com/nats-io/nkeys v0.4.14 // indirect
+	github.com/nats-io/nkeys v0.4.15 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/segmentio/asm v1.2.1 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.devnw.com/structs v1.0.0 // indirect
-	golang.org/x/crypto v0.47.0 // indirect
-	golang.org/x/sys v0.40.0 // indirect
+	golang.org/x/crypto v0.48.0 // indirect
+	golang.org/x/sys v0.41.0 // indirect
 )