Bug: Re-enrichment duplicates work experiences creating the same employments without dates (#1185)

Author: sausage-todd

backend/src/database/migrations/U1690797541__fix-duplicate-work-experiences.sql

@@ -0,0 +1,53 @@
+DO $$
+    DECLARE
+        _record RECORD;
+        _experience JSONB;
+        _to_cleanup UUID[];
+        _with_dates_exist BOOLEAN;
+    BEGIN
+        -- iterate over every work experience, grouped by member and organization
+        FOR _record IN SELECT
+                           "memberId",
+                           "organizationId",
+                           COUNT(*) AS count,
+                           JSONB_AGG(
+                               JSON_BUILD_OBJECT(
+                                   'id', "id",
+                                   'dateStart', TO_CHAR("dateStart", 'YYYY-MM-DD'),
+                                   'dateEnd', TO_CHAR("dateEnd", 'YYYY-MM-DD'),
+                                   'title', "title"
+                                   )
+                               ) AS experiences
+                       FROM "memberOrganizations"
+                       GROUP BY "memberId", "organizationId"
+                       ORDER BY 3 DESC
+            LOOP
+                _with_dates_exist := FALSE;
+                _to_cleanup := ARRAY[]::UUID[];
+
+                -- ignore those that don't have duplicates
+                IF _record.count <=1 THEN
+                    CONTINUE;
+                END IF;
+
+                -- iterate over every experience
+                FOR _experience IN SELECT JSONB_ARRAY_ELEMENTS(_record.experiences) LOOP
+                    RAISE NOTICE 'Processing record %, experience: %', _record."memberId", _experience;
+
+                    -- if there is no start date, mark it for deletion
+                    -- but also check if there are any experiences with dates
+                    IF _experience->>'dateStart' IS NULL THEN
+                        _to_cleanup := _to_cleanup || (_experience->>'id')::UUID;
+                    ELSE
+                        _with_dates_exist := TRUE;
+                    END IF;
+                END LOOP;
+
+                -- and then only delete duplicates if there are experiences with dates
+                IF _with_dates_exist THEN
+                    RAISE NOTICE 'Deleting experiences: %', _to_cleanup;
+                    DELETE FROM "memberOrganizations" WHERE "id" = ANY (_to_cleanup);
+                END IF;
+            END LOOP;
+    END;
+$$ LANGUAGE PLPGSQL

backend/src/database/migrations/V1690797541__fix-duplicate-work-experiences.sql

@@ -1,6 +1,7 @@
 import { QueryTypes } from 'sequelize'
 import { EnrichmentCache } from '../../services/premium/enrichment/types/memberEnrichmentTypes'
 import { IRepositoryOptions } from './IRepositoryOptions'
+import SequelizeRepository from './sequelizeRepository'
 
 class MemberEnrichmentCacheRepository {
   /**
@@ -18,6 +19,7 @@ class MemberEnrichmentCacheRepository {
     options: IRepositoryOptions,
   ): Promise<EnrichmentCache> {
     if (data && Object.keys(data).length > 0) {
+      const transaction = SequelizeRepository.getTransaction(options)
       await options.database.sequelize.query(
         `INSERT INTO "memberEnrichmentCache" ("createdAt", "updatedAt", "memberId", "data")
           VALUES
@@ -31,6 +33,7 @@ class MemberEnrichmentCacheRepository {
             data: JSON.stringify(data),
           },
           type: QueryTypes.UPSERT,
+          transaction,
         },
       )
     }
@@ -50,6 +53,7 @@ class MemberEnrichmentCacheRepository {
     memberId: string,
     options: IRepositoryOptions,
   ): Promise<EnrichmentCache> {
+    const transaction = SequelizeRepository.getTransaction(options)
     const records = await options.database.sequelize.query(
       `select *
        from "memberEnrichmentCache"
@@ -60,6 +64,7 @@ class MemberEnrichmentCacheRepository {
           memberId,
         },
         type: QueryTypes.SELECT,
+        transaction,
       },
     )
 

backend/src/database/repositories/memberEnrichmentCacheRepository.ts

@@ -547,7 +547,23 @@ class MemberRepository {
                             group by "memberId", platform) mi
                       group by mi."memberId"),
         member_organizations as (
-          select "memberId", array_agg("organizationId") as orgs 
+          select
+            "memberId",
+            JSONB_AGG(
+                DISTINCT JSONB_BUILD_OBJECT(
+                  'id', "organizationId",
+                  'memberOrganizations',
+                  JSONB_BUILD_OBJECT(
+                    'memberId', "memberId",
+                    'organizationId', "organizationId",
+                    'dateStart', "dateStart",
+                    'dateEnd', "dateEnd",
+                    'createdAt', "createdAt",
+                    'updatedAt', "updatedAt",
+                    'title', title
+                  )
+                )
+            ) AS orgs
           from "memberOrganizations"
           where "memberId" = :memberId
           group by "memberId"
@@ -571,7 +587,7 @@ class MemberRepository {
               m."updatedById",
               i.username,
               si."segmentIds" as segments,
-              coalesce(mo.orgs, array []::uuid[]) as "organizations"
+              coalesce(mo.orgs, '[]'::JSONB) as "organizations"
         from members m
                 inner join identities i on i."memberId" = m.id
                 inner join segment_ids si on si."memberId" = m.id
@@ -3242,23 +3258,41 @@ class MemberRepository {
     const seq = SequelizeRepository.getSequelize(options)
     const transaction = SequelizeRepository.getTransaction(options)
 
-    const query = `
-      INSERT INTO "memberOrganizations" ("memberId", "organizationId", "createdAt", "updatedAt", "title", "dateStart", "dateEnd")
-      VALUES (:memberId, :organizationId, NOW(), NOW(), :title, :dateStart, :dateEnd)
-      ON CONFLICT ("memberId", "organizationId", "dateStart", "dateEnd") DO NOTHING
-    `
+    await seq.query(
+      `
+        DELETE FROM "memberOrganizations"
+        WHERE "memberId" = :memberId
+        AND "organizationId" = :organizationId
+        AND "dateEnd" IS NULL
+      `,
+      {
+        replacements: {
+          memberId,
+          organizationId,
+        },
+        type: QueryTypes.DELETE,
+        transaction,
+      },
+    )
 
-    await seq.query(query, {
-      replacements: {
-        memberId,
-        organizationId,
-        title: title || null,
-        dateStart: dateStart || null,
-        dateEnd: dateEnd || null,
+    await seq.query(
+      `
+        INSERT INTO "memberOrganizations" ("memberId", "organizationId", "createdAt", "updatedAt", "title", "dateStart", "dateEnd")
+        VALUES (:memberId, :organizationId, NOW(), NOW(), :title, :dateStart, :dateEnd)
+        ON CONFLICT ("memberId", "organizationId", "dateStart", "dateEnd") DO NOTHING
+      `,
+      {
+        replacements: {
+          memberId,
+          organizationId,
+          title: title || null,
+          dateStart: dateStart || null,
+          dateEnd: dateEnd || null,
+        },
+        type: QueryTypes.INSERT,
+        transaction,
       },
-      type: QueryTypes.INSERT,
-      transaction,
-    })
+    )
   }
 
   static sortOrganizations(organizations) {

backend/src/database/repositories/memberRepository.ts

@@ -326,7 +326,7 @@ export default class MemberService extends LoggerBase {
             }
             // We findOrCreate the organization and add it to the list of IDs
             const organizationRecord = await organizationService.findOrCreate(data)
-            organizations.push(organizationRecord.id)
+            organizations.push({ id: organizationRecord.id })
           }
         }
 
@@ -349,14 +349,14 @@ export default class MemberService extends LoggerBase {
             if (domain) {
               const organizationRecord = await organizationService.findByUrl(domain)
               if (organizationRecord) {
-                organizations.push(organizationRecord.id)
+                organizations.push({ id: organizationRecord.id })
               }
             }
           }
         }
 
         // Remove dups
-        data.organizations = [...new Set(organizations)]
+        data.organizations = lodash.uniqBy(organizations, 'id')
       }
 
       const fillRelations = false

backend/src/services/memberService.ts

@@ -33,6 +33,7 @@ import {
 import OrganizationService from '../../organizationService'
 import MemberRepository from '../../../database/repositories/memberRepository'
 import OrganizationRepository from '../../../database/repositories/organizationRepository'
+import SequelizeRepository from '@/database/repositories/sequelizeRepository'
 
 export default class MemberEnrichmentService extends LoggerBase {
   options: IServiceOptions
@@ -219,37 +220,48 @@ export default class MemberEnrichmentService extends LoggerBase {
    * @returns a promise that resolves to the enrichment data for the member
    */
   async enrichOne(memberId) {
-    // If the attributes have not been fetched yet, fetch them
-    if (!this.attributes) {
-      await this.getAttributes()
+    const transaction = await SequelizeRepository.createTransaction(this.options)
+    const options = {
+      ...this.options,
+      transaction,
     }
 
-    // Create an instance of the MemberService and use it to look up the member
-    const memberService = new MemberService(this.options)
-    const member = await memberService.findById(memberId, false, false)
+    try {
+      // If the attributes have not been fetched yet, fetch them
+      if (!this.attributes) {
+        await this.getAttributes()
+      }
 
-    // If the member's GitHub handle or email address is not available, throw an error
-    if (!member.username[PlatformType.GITHUB] && member.emails.length === 0) {
-      throw new Error400(this.options.language, 'enrichment.errors.noGithubHandleOrEmail')
-    }
+      // Create an instance of the MemberService and use it to look up the member
+      const memberService = new MemberService(options)
+      const member = await memberService.findById(memberId, false, false)
 
-    let enrichedFrom = ''
-    let enrichmentData: EnrichmentAPIMember
-    // If the member has a GitHub handle, use it to make a request to the Enrichment API
-    if (member.username[PlatformType.GITHUB]) {
-      enrichedFrom = 'github'
-      enrichmentData = await this.getEnrichmentByGithubHandle(
-        member.username[PlatformType.GITHUB][0],
-      )
-    } else if (member.emails.length > 0) {
-      enrichedFrom = 'email'
-      // If the member has an email address, use it to make a request to the Enrichment API
-      enrichmentData = await this.getEnrichmentByEmail(member.emails[0])
-    }
+      // If the member's GitHub handle or email address is not available, throw an error
+      if (!member.username[PlatformType.GITHUB] && member.emails.length === 0) {
+        throw new Error400(this.options.language, 'enrichment.errors.noGithubHandleOrEmail')
+      }
+
+      let enrichedFrom = ''
+      let enrichmentData: EnrichmentAPIMember
+      // If the member has a GitHub handle, use it to make a request to the Enrichment API
+      if (member.username[PlatformType.GITHUB]) {
+        enrichedFrom = 'github'
+        enrichmentData = await this.getEnrichmentByGithubHandle(
+          member.username[PlatformType.GITHUB][0],
+        )
+      } else if (member.emails.length > 0) {
+        enrichedFrom = 'email'
+        // If the member has an email address, use it to make a request to the Enrichment API
+        enrichmentData = await this.getEnrichmentByEmail(member.emails[0])
+      }
+
+      if (!enrichmentData) {
+        await SequelizeRepository.commitTransaction(transaction)
+        return null
+      }
 
-    if (enrichmentData) {
       // save raw data to cache
-      await MemberEnrichmentCacheRepository.upsert(memberId, enrichmentData, this.options)
+      await MemberEnrichmentCacheRepository.upsert(memberId, enrichmentData, options)
 
       const normalized = await this.normalize(member, enrichmentData)
 
@@ -270,18 +282,18 @@ export default class MemberEnrichmentService extends LoggerBase {
           memberId: member.id,
           enrichedFrom,
         },
-        this.options,
+        options,
       )
 
-      const result = await memberService.upsert({
+      let result = await memberService.upsert({
         ...normalized,
         platform: Object.keys(member.username)[0],
       })
 
       // for every work experience in `enrichmentData`
       //   - upsert organization
       //   - upsert `memberOrganization` relation
-      const organizationService = new OrganizationService(this.options)
+      const organizationService = new OrganizationService(options)
       if (enrichmentData.work_experiences) {
         for (const workExperience of enrichmentData.work_experiences) {
           const org = await organizationService.findOrCreate({
@@ -299,14 +311,18 @@ export default class MemberEnrichmentService extends LoggerBase {
             dateStart: workExperience.startDate,
             dateEnd,
           }
-          await MemberRepository.createOrUpdateWorkExperience(data, this.options)
-          await OrganizationRepository.includeOrganizationToSegments(org.id, this.options)
+          await MemberRepository.createOrUpdateWorkExperience(data, options)
+          await OrganizationRepository.includeOrganizationToSegments(org.id, options)
         }
       }
 
+      result = await memberService.findById(result.id, true, false)
+      await SequelizeRepository.commitTransaction(transaction)
       return result
+    } catch (error) {
+      await SequelizeRepository.rollbackTransaction(transaction)
+      throw error
     }
-    return null
   }
 
   async normalize(member: Member, enrichmentData: EnrichmentAPIMember) {