perf: skip unchanged organization writes in data-sink member updates

services/libs/data-access-layer/src/organizations/attributes.ts

@@ -86,6 +86,34 @@ export const upsertOrgAttributes = async (
   }
 }
 
+const isSameOrgAttribute = (
+  stored: IDbOrgAttribute,
+  attribute: { name: string; source: string; value: string },
+): boolean =>
+  stored.name === attribute.name &&
+  stored.source === attribute.source &&
+  stored.value === attribute.value
+
+// Must match the conflict key of upsertOrgAttributes: organizationId, name, source, md5(value)
+export function hasAllOrgAttributes(
+  stored: IDbOrgAttribute[],
+  attributes: IDbOrgAttributeInput[],
+): boolean {
+  return attributes.every((attribute) => stored.some((s) => isSameOrgAttribute(s, attribute)))
+}
+
+// Must match the flags markOrgAttributeDefault sets
+export function isOrgAttributeDefault(
+  stored: IDbOrgAttribute[],
+  attribute: { name: string; source: string; value: string },
+): boolean {
+  const sameName = stored.filter((s) => s.name === attribute.name)
+  return (
+    sameName.some((s) => isSameOrgAttribute(s, attribute)) &&
+    sameName.every((s) => s.default === isSameOrgAttribute(s, attribute))
+  )
+}
+
 export async function markOrgAttributeDefault(
   qx: QueryExecutor,
   organizationId: string,

services/libs/data-access-layer/src/organizations/base.ts

@@ -22,9 +22,23 @@ import { QueryExecutor } from '../queryExecutor'
 import { findManyLfSegmentsByNames } from '../segments'
 import { QueryOptions, QueryResult, prepareBulkInsert, queryTable, queryTableById } from '../utils'
 import { prepareSelectColumns } from '../utils'
-import { findOrgAttributes, markOrgAttributeDefault, upsertOrgAttributes } from './attributes'
-import { insertOrganizationIdentities, upsertOrgIdentities } from './identities'
-import { IDbOrganization, IDbOrganizationInput, IFindOrCreateOrganizationResult } from './types'
+import {
+  hasAllOrgAttributes,
+  isOrgAttributeDefault,
+  markOrgAttributeDefault,
+  upsertOrgAttributes,
+} from './attributes'
+import {
+  hasAllOrgIdentities,
+  insertOrganizationIdentities,
+  upsertOrgIdentities,
+} from './identities'
+import {
+  IDbOrgAttribute,
+  IDbOrganization,
+  IDbOrganizationInput,
+  IFindOrCreateOrganizationResult,
+} from './types'
 import { prepareOrganizationData } from './utils'
 
 const log = getServiceChildLogger('data-access-layer/organizations')
@@ -530,6 +544,45 @@ async function findOrgForUpdate(qe: QueryExecutor, id: string): Promise<IDbOrgan
   )
 }
 
+interface IStoredOrg {
+  attributes: IDbOrgAttribute[]
+  identities: IOrganizationIdentity[]
+}
+
+async function findOrgAttributesAndIdentities(qe: QueryExecutor, id: string): Promise<IStoredOrg> {
+  return qe.selectOne(
+    `
+      select
+        coalesce((select jsonb_agg(oa) from "orgAttributes" oa where oa."organizationId" = $(id)), '[]') as attributes,
+        coalesce((select jsonb_agg(oi) from "organizationIdentities" oi where oi."organizationId" = $(id)), '[]') as identities
+    `,
+    { id },
+  )
+}
+
+function planOrgUpdate(
+  data: IOrganization,
+  source: string,
+  existing: IDbOrganization,
+  stored: IStoredOrg,
+) {
+  // prepareOrganizationData rewrites the default flags of the attributes it gets, so it gets copies
+  const processed = prepareOrganizationData(
+    data,
+    source,
+    existing,
+    stored.attributes.map((a) => ({ ...a })),
+  )
+  return {
+    processed,
+    changesNothing:
+      Object.keys(processed.organization).length === 0 &&
+      hasAllOrgIdentities(stored.identities, data.identities) &&
+      hasAllOrgAttributes(stored.attributes, processed.attributes) &&
+      processed.attributes.every((a) => !a.default || isOrgAttributeDefault(stored.attributes, a)),
+  }
+}
+
 // lockOrganization: a caller that runs this concurrently without transactions passes a lock, so
 // calls that reach the same organization through different identities update it one at a time
 export async function findOrCreateOrganization(
@@ -618,21 +671,28 @@ export async function findOrCreateOrganization(
     if (existing) {
       log.trace(`Found existing organization, organization will be updated!`)
 
+      let stored = await logExecutionTimeV2(
+        async () => findOrgAttributesAndIdentities(qe, existing.id),
+        log,
+        'organizationService -> findOrCreateOrganization -> findOrgAttributesAndIdentities',
+      )
+      let plan = planOrgUpdate(data, source, existing, stored)
+      if (plan.changesNothing) {
+        return { id: existing.id, created: false }
+      }
+
       const release = await lockOrganization?.(existing.id)
       try {
         if (release) {
           // read it again: the call that held the lock may have changed it
           existing = (await findOrgForUpdate(qe, existing.id)) ?? existing
+          stored = await findOrgAttributesAndIdentities(qe, existing.id)
+          plan = planOrgUpdate(data, source, existing, stored)
         }
 
-        const existingAttributes = await logExecutionTimeV2(
-          async () => findOrgAttributes(qe, existing.id),
-          log,
-          'organizationService -> findOrCreateOrganization -> findOrgAttributes',
-        )
-
-        const processed = prepareOrganizationData(data, source, existing, existingAttributes)
-
+        // Once anything changes, every write runs as before: a decision taken from the earlier read
+        // could be stale by now for callers without the lock
+        const { processed } = plan
         log.trace({ updateData: processed.organization }, `Updating organization!`)
 
         if (Object.keys(processed.organization).length > 0) {

services/libs/data-access-layer/src/organizations/identities.ts

@@ -121,6 +121,26 @@ export async function insertOrganizationIdentities(
   return qx.result(query)
 }
 
+function findStoredOrgIdentity(
+  stored: IOrganizationIdentity[],
+  identity: Partial<IOrganizationIdentity>,
+): IOrganizationIdentity | undefined {
+  return stored.find(
+    (s) =>
+      s.value === identity.value && s.platform === identity.platform && s.type === identity.type,
+  )
+}
+
+export function hasAllOrgIdentities(
+  stored: IOrganizationIdentity[],
+  identities: Partial<IOrganizationIdentity>[],
+): boolean {
+  return identities.every((identity) => {
+    const existing = findStoredOrgIdentity(stored, identity)
+    return existing && (identity.verified === undefined || existing.verified === identity.verified)
+  })
+}
+
 export async function upsertOrgIdentities(
   qe: QueryExecutor,
   organizationId: string,
@@ -132,9 +152,7 @@ export async function upsertOrgIdentities(
   const toUpdate: Partial<IOrganizationIdentity>[] = []
 
   for (const i of identities) {
-    const existing = existingIdentities.find(
-      (ei) => ei.value === i.value && ei.platform === i.platform && ei.type === i.type,
-    )
+    const existing = findStoredOrgIdentity(existingIdentities, i)
     if (!existing) {
       toCreate.push({
         organizationId,