diff --git a/services/libs/data-access-layer/src/organizations/attributes.ts b/services/libs/data-access-layer/src/organizations/attributes.ts index 280e5f8937..03c0a0e4b8 100644 --- a/services/libs/data-access-layer/src/organizations/attributes.ts +++ b/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, diff --git a/services/libs/data-access-layer/src/organizations/base.ts b/services/libs/data-access-layer/src/organizations/base.ts index dc4579f622..38b8ad886e 100644 --- a/services/libs/data-access-layer/src/organizations/base.ts +++ b/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 { + 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) { diff --git a/services/libs/data-access-layer/src/organizations/identities.ts b/services/libs/data-access-layer/src/organizations/identities.ts index df7d9d73b7..2606fc7915 100644 --- a/services/libs/data-access-layer/src/organizations/identities.ts +++ b/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 | undefined { + return stored.find( + (s) => + s.value === identity.value && s.platform === identity.platform && s.type === identity.type, + ) +} + +export function hasAllOrgIdentities( + stored: IOrganizationIdentity[], + identities: Partial[], +): 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[] = [] 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,