[themarolt]

Summary

When one data-sink batch carried two different payloads for the same organization, only the first one's data was written. The batch's organization cache was keyed by verified identities only, so later payloads got the cached organization id and their attributes were dropped. In meetings data this happens when attendees of one meeting are linked to two Salesforce accounts with the same website (for example "Croxel Inc." and "Croxel"): one has a logo and a different industry, and only one of them was ever stored.

This PR applies every distinct payload once, and makes the payload from the newest activity set the defaults.

Changes

• The batch's organization cache is keyed by identities, find-or-create source and the payload content, with domains normalized the way findOrCreateOrganization normalizes them. Identical payloads still share one call. A payload with different content runs its own find-or-create. The existing identity locks make it wait for the earlier one, and it goes to the organization that one resolved to, without a lookup, and adds what it carries. No duplicate organizations.
• Before member processing, the batch records for each organization payload the newest activity time and how many activities carry it (once per activity, even when both the member and the object member carry it). After the member updates, for each organization and source that got more than one distinct payload, all of them are applied again, to the organization they went to, from the lowest ranked to the highest: newest activity first, then the payload more activities carry, then a fixed order. Each attribute's default for that source ends on the highest-ranked payload that has it, wherever that source is the organization's top source for the attribute. Plain organization columns are only filled when empty, as before. A failure in this step is logged per payload and doesn't fail the batch's activities.
• addOrgsToSegments sorts the organization and segment ids, so concurrent inserts of the same new rows always use the same order and can't deadlock. Its callers keep their behavior.

Local end-to-end runs against main, on 92.5k real github/git results and 39,659 real meetings activities seeded read-only from prod:

• github/git: identical to main in every table, and in Kafka activity messages, search-sync messages and Redis keys.
• meetings: identical to main except 2 added non-default attributes for croxel.com (the meetings logo and industry). The two runs of this branch are identical to each other, and there are no duplicate organizations.
• Speed: unchanged within run-to-run noise (meetings 0.159–0.207 s per batch against 0.177–0.192 on main, github/git 0.101–0.103 against 0.106). Find-or-create calls are 1.4% more, and the final step takes 7–10 s per run in total.

Type of change

• Bug fix
• New feature
• Refactor / cleanup
• Performance improvement
• Chore / dependency update
• Documentation

Related issue

Closes linuxfoundation/lfx-cdp-insights-ops#972

[cursor]

PR Summary

Medium Risk
Changes organization merge, caching, and default-selection logic in the hot data-sink path; incorrect ranking could skew org attributes, though re-apply failures are isolated from activity success.

Overview
Fixes a bug where a data-sink batch could carry multiple different organization payloads for the same company (same verified identities) but only the first payload’s attributes were persisted, because the batch cache was keyed by identities alone.

Batch org deduplication and merging now keys cache entries by verified identities plus member link source plus normalized payload content (orgPayloadKey). Identical payloads still share one findOrCreate; different content gets its own call. Shared-identity locks and a per-batch map of resolved org IDs keep concurrent updates on one organization without duplicate rows. findOrCreateOrganization accepts an optional organizationId so follow-up applies target the org from an earlier payload in the batch (including when a later payload unverified the lookup identity).

Default / winner selection: Before member work, registerOrgPayloadActivities records newest activity time and activity count per payload. After updateMembers, applyNewestOrgPayloads re-applies every distinct payload for each org+source (lowest rank first, highest last) so per-source attribute defaults reflect the payload tied to the newest activity (then count, then stable key). Errors in that step are logged only and do not fail the batch’s activities.

Deadlock guard: addOrgsToSegments sorts segment and organization IDs before insert so concurrent batches inserting the same junction rows use a consistent lock order.

^{Reviewed by Cursor Bugbot for commit be35606. Bugbot is set up for automated code reviews on this repo. Configure here.}

[Copilot]

🟡 Changes recommended

Organization defaults remain processing-order dependent for omitted attributes, and duplicate payloads within a single activity can inflate counts and select the wrong tie-break winner.

2 open issues

• Deduplicate organization votes within each activity · New
• Replay distinct payloads from lowest to highest precedence · New

What changed in this PR

This PR fixes organization updates being dropped when a data-sink batch contains different payloads for the same organization, while preserving deduplication for identical payloads.

Changes:

• Keys organization caching by normalized payload content and source.
• Tracks activity timestamps and counts, then reapplies the winning payload to set source defaults.
• Sorts organization and segment IDs before inserts to prevent conflicting insertion orders.

File	Description
services/​libs/​data-access-layer/​src/​organizations/​base.ts	Sorts IDs before inserting organization-segment links.
services/​apps/​data_sink_worker/​src/​service/​member.service.ts	Adds payload-aware caching, activity ranking, and winner replay.
services/​apps/​data_sink_worker/​src/​service/​activity.service.ts	Registers organization payload activity data and invokes replay after member updates.

_{Review effort: Balanced}

[Copilot]

🟡 Changes recommended

Replaying organization resolution can create duplicates and send updates to the wrong organization, while malformed organization entries can cause an entire batch to fail.

1 open issue

• Replay can create a duplicate organization after identity changes · New

1 previously missed issue

In code that hasn't changed since last review

Malformed organization entry aborts the entire batch

services/​apps/​data_sink_worker/​src/​service/​activity.service.ts:1475

Organization entries are not validated by prepareMemberData, so an entry such as organizations: [null] throws during this pre-pass, outside the per-member error handlers. This makes the whole batch retry and eventually fail, including valid activities. Catch registration errors per activity and leave malformed organizations to the existing member-level handling.

2 resolved since last review

• Replay distinct payloads from lowest to highest precedence
• Deduplicate organization votes within each activity

_{Review effort: Balanced}

[Copilot]

🔵 Needs a closer look

The initial pass can still create duplicate organizations when a domain is unverified between payload variants, splitting member links in a way that replaying recorded IDs cannot repair.

No open issues 🎉

1 previously missed issue

In code that hasn't changed since last review

Preserve identity resolution across payload variants during initial pass

services/​apps/​data_sink_worker/​src/​service/​member.service.ts:297

The initial pass can still create duplicate organizations: A resolves Acme by a verified domain without a display name, B finds Acme by name and unverifies that domain, then C supplies A's domain with different attributes and no display name. C misses the payload-specific cache and both verified-identity lookups, creating a second organization; pinning replay to recorded IDs cannot repair the split member links. Preserve identity-to-organization resolution separately from payload deduplication and reuse the resolved ID for later variants during the initial pass too.

1 resolved since last review

• Replay can create a duplicate organization after identity changes

_{Review effort: Balanced}

[themarolt]

Copilot review on 3560fec, previously missed: Preserve identity resolution across payload variants during initial pass (member.service.ts:297). Fixed in c1d99d8. Each batch keeps the organization id that each identity key resolved to. Later payloads with the same verified identities, or the same name when they have none, go to that organization without a lookup, so a payload that unverifies a domain earlier in the batch can't send a later variant to a new organization. Local meetings and github/git runs give the same data as before this change.

[Copilot]

🔵 Needs a closer look

Human assessment is needed because identity resolution and verification updates can interact to split one organization into duplicates that replay cannot repair.

No open issues 🎉

_{Review effort: Balanced}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit c1d99d8. Configure here.}

[Copilot]

🟡 Changes recommended

The platform-independent organization cache key can route email payloads to a GitHub organization sharing the same domain, writing data and affiliations to the wrong organization.

1 open issue

• Include platform in organization cache keys · New

_{Review effort: Balanced}

[Copilot]

🔵 Needs a closer look

Human assessment is needed because interactions between username updates, domain verification changes, and cache-key differences can cause an existing organization to be duplicated.

No open issues 🎉

1 resolved since last review

• Include platform in organization cache keys

_{Review effort: Balanced}

[themarolt]

Copilot review on 2a85ae8, Needs a closer look: duplicates from username updates, domain verification changes and cache-key differences.

Checked against main. This PR adds no new path to a duplicate:

• Payloads with the same verified identities (platform, type, normalized value), or the same name when they have none, go to the organization the first one resolved to. Main's identity-keyed cache returned that same result, so this matches main and adds the later payload's data.
• Payloads with different identity sets each run the normal lookup, as on main. The identity locks make them wait for each other, so the later one sees what the earlier one wrote.
• Replays go to the recorded organization id and never create one.

The duplicate in that summary comes from upsertOrgIdentities overwriting a stored verified flag with an incoming false. That applies to usernames as well as domains: GitHub sends an organization's website and Twitter username unverified. Once the only verified identity is cleared, a later payload with a different identity set misses every lookup and creates a new organization. That has been the behavior since #2465, across batches as well as within one, and is tracked separately in linuxfoundation/lfx-cdp-insights-ops#978.

[themarolt]

Copilot review on 85cd9db, previously missed: Malformed organization entry aborts the entire batch (activity.service.ts:1475). Fixed in be35606. The registration pass runs before the per-member error handling, so it now skips a malformed entry (for example organizations: [null], or a non-array organizations) and an invalid timestamp instead of throwing. The member-level handling reports the malformed organization as before, and the other activities of the batch go through. Well-formed payloads take the same path as before.

Copilot review on c1d99d8, Needs a closer look (identity resolution and verification updates splitting one organization): the same answer as #5171 (comment). The PR adds no new duplicate path. The split comes from incoming unverified identities clearing a stored verified flag, which predates this PR and is tracked in linuxfoundation/lfx-cdp-insights-ops#978.

[Copilot]

🟢 Approval recommended

No active review issues are identified for the organization payload handling changes.

No open issues 🎉

_{Review effort: Balanced}