fix: use correct merge API endpoint in LFID enrichment (CM-1142)

services/apps/members_enrichment_worker/src/activities.ts

@@ -19,11 +19,6 @@ import {
   updateMemberUsingSquashedPayload,
 } from './activities/enrichment'
 import { refreshToken } from './activities/lf-auth0/authenticateLFAuth0'
-import {
-  getIdentitiesExistInOtherMembers,
-  mergeMembers,
-  updateMemberWithEnrichmentData,
-} from './activities/lf-auth0/enrichLFAuth0'
 import { getEnrichmentLFAuth0 } from './activities/lf-auth0/getEnrichmentLFAuth0'
 import { getLFIDEnrichableMembers } from './activities/lf-auth0/getLFIDEnrichableMembers'
 import {
@@ -39,9 +34,12 @@ import {
 } from './activities/llm'
 import {
   getEnrichableMembers,
+  getIdentitiesExistInOtherMembers,
   getMemberById,
+  mergeMembers,
   syncMembersToOpensearch,
   syncOrganizationsToOpensearch,
+  updateMemberWithEnrichmentData,
 } from './activities/member'
 
 export {

services/apps/members_enrichment_worker/src/activities/member.ts

@@ -1,9 +1,21 @@
-import { MemberField, findMemberById, pgpQx } from '@crowd/data-access-layer'
-import { fetchMembersForEnrichment } from '@crowd/data-access-layer/src/old/apps/members_enrichment_worker'
+import {
+  MemberField,
+  PgPromiseQueryExecutor,
+  createMemberIdentity,
+  findMemberById,
+  pgpQx,
+} from '@crowd/data-access-layer'
+import {
+  fetchMembersForEnrichment,
+  getIdentitiesExistInOtherMembers as getIdentitiesExistInOthers,
+  updateMemberAttributes,
+} from '@crowd/data-access-layer/src/old/apps/members_enrichment_worker'
 import { MemberSyncService, OrganizationSyncService } from '@crowd/opensearch'
 import {
+  IAttributes,
   IEnrichableMember,
   IEnrichmentSourceQueryInput,
+  IMemberIdentity,
   MemberEnrichmentSource,
 } from '@crowd/types'
 
@@ -60,3 +72,71 @@
 export async function syncOrganizationsToOpensearch(input: string[]): Promise<void> {
   await organizationSyncService.syncOrganizations(input)
 }
+
+export async function getIdentitiesExistInOtherMembers(
+  excludeMemberId: string,
+  identities: IMemberIdentity[],
+): Promise<IMemberIdentity[]> {
+  let rows: IMemberIdentity[] = []
+
+  try {
+    const db = svc.postgres.reader
+    rows = await getIdentitiesExistInOthers(db, excludeMemberId, identities)
+  } catch (err) {
+    throw err
+  }
+
+  return rows
+}
+
+export async function updateMemberWithEnrichmentData(
+  memberId: string,
+  identities: IMemberIdentity[],
+  attributes?: IAttributes,
+): Promise<void> {
+  try {
+    await svc.postgres.writer.connection().tx(async (tx) => {
+      for (const identity of identities) {
+        await createMemberIdentity(new PgPromiseQueryExecutor(tx), {
+          memberId,
+          platform: identity.platform,
+          value: identity.value,
+          type: identity.type,
+          verified: identity.verified || false,
+          source: 'enrichment',
+        })
+      }
+      if (attributes) {
+        await updateMemberAttributes(tx, memberId, attributes)
+      }
+    })
+  } catch (err) {
+    throw err
+  }
+}
+
+export async function mergeMembers(
+  primaryMemberId: string,
+  secondaryMemberId: string,
+): Promise<void> {
+  const res = await fetch(
+    `${process.env['CROWD_API_SERVICE_URL']}/member/${primaryMemberId}/merge`,
+    {
+      method: 'PUT',
+      headers: {
+        Authorization: `Bearer ${process.env['CROWD_API_SERVICE_USER_TOKEN']}`,
+        'Content-Type': 'application/json',
+      },
+      body: JSON.stringify({
+        memberToMerge: secondaryMemberId,
+      }),
+    },
+  )
+
+  if (res.status !== 200) {
+    const body = await res.text()
+    throw new Error(
+      `Failed to merge member ${primaryMemberId} with ${secondaryMemberId}! Status: ${res.status}, Response: ${body}`,
+    )
+  }
+}