-
Notifications
You must be signed in to change notification settings - Fork 1
Standardize v1-objects data decoding with unified abstraction #41
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Closed
Closed
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -536,18 +536,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 using unified abstraction | ||||||
| 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 | ||||||
| } | ||||||
|
|
||||||
|
|
@@ -1251,18 +1248,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 using unified abstraction | ||||||
| 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") | ||||||
|
||||||
| funcLogger.With(errKey, err).ErrorContext(ctx, "failed to get past meeting data from KV bucket") | |
| funcLogger.With(errKey, err).WarnContext(ctx, "failed to get past meeting data from KV bucket") |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -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) | ||||||
| } | ||||||
|
Comment on lines
+141
to
147
|
||||||
|
|
||||||
| // 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 using the unified abstraction | ||||||
| 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) | ||||||
|
||||||
| return "", false, false, fmt.Errorf("failed to get email data: %w", err) | |
| return "", false, false, fmt.Errorf("failed to get email data for emailSfid %s (key %s): %w", emailSfid, emailKey, err) |
Copilot
AI
Feb 19, 2026
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When KV.Get fails with a non-notfound error, this message doesn’t include the KV key being fetched. Including the key in the returned error would significantly improve debuggability (especially since many call sites only wrap with a generic "failed to get ... data").
Suggested change
| return nil, false, fmt.Errorf("failed to get data from v1-objects KV bucket: %w", err) | |
| return nil, false, fmt.Errorf("failed to get data from v1-objects KV bucket for key %q: %w", key, err) |
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This failure is handled by logging and returning (i.e., it’s a recoverable condition that just prevents re-index). Logging it at Error level may create noisy alerts; consider using Warn level here to match other recoverable KV/mapping misses in this service.