chore: clean up script for the email identity shape

Signed-off-by: Yeganathan S <63534555+skwowet@users.noreply.github.com>

Author: skwowet

backend/src/bin/scripts/cleanup-same-member-case-variant-identities.ts

@@ -1,5 +1,6 @@
 import commandLineArgs from 'command-line-args'
 
+import { isEmail } from '@crowd/common'
 import { pgpQx } from '@crowd/data-access-layer'
 import { getDbConnection } from '@crowd/data-access-layer/src/database'
 import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
@@ -19,104 +20,58 @@ const options = [
 
 const parameters = commandLineArgs(options)
 
-type CaseVariantGroup = {
+type MixedCaseEmailIdentity = {
+  id: string
   memberId: string
   platform: string
   type: string
-  lv: string
-}
-
-type IdentityRow = {
-  id: string
   value: string
-  verified: boolean
-  verifiedBy: string | null
-  updatedAt: Date
 }
 
 const AR_UPDATE_BATCH_SIZE = 5000
+const LOAD_BATCH_SIZE = 500
 
-function pickKeeper(rows: IdentityRow[]): IdentityRow {
-  return [...rows].sort((a, b) => {
-    if (a.verified !== b.verified) return a.verified ? -1 : 1
-    if (Boolean(a.verifiedBy) !== Boolean(b.verifiedBy)) return a.verifiedBy ? -1 : 1
-    const aUpdated = new Date(a.updatedAt).getTime()
-    const bUpdated = new Date(b.updatedAt).getTime()
-    if (aUpdated !== bUpdated) return bUpdated - aUpdated
-    return a.id < b.id ? -1 : 1
-  })[0]
-}
-
-async function findDuplicateCaseVariantGroups(
-  qx: QueryExecutor,
-  limit?: number,
-): Promise<CaseVariantGroup[]> {
-  const baseQuery = `
-      select
-        "memberId",
-        platform,
-        type,
-        lower(value) as lv
-      from "memberIdentities"
-      where "deletedAt" is null
-      group by "memberId", platform, type, lower(value)
-      having count(distinct value) > 1
-      order by "memberId", platform, type, lower(value)
-  `
-
-  if (limit != null) {
-    return qx.select(`${baseQuery} limit $(limit)`, { limit })
-  }
-
-  return qx.select(baseQuery)
-}
-
-async function fetchGroupIdentities(
+async function findMixedCaseEmailIdentities(
   qx: QueryExecutor,
-  group: CaseVariantGroup,
-): Promise<IdentityRow[]> {
+  afterId: string | null,
+  limit: number,
+): Promise<MixedCaseEmailIdentity[]> {
   return qx.select(
     `
-      select id, value, verified, "verifiedBy", "updatedAt"
+      select id, "memberId", platform, type, value
       from "memberIdentities"
-      where "memberId" = $(memberId)
-        and platform = $(platform)
-        and type = $(type)
-        and lower(value) = $(lv)
-        and "deletedAt" is null
+      where "deletedAt" is null
+        and value <> lower(value)
+        and position('@' in value) > 0
+        ${afterId ? 'and id > $(afterId)' : ''}
+      order by id
+      limit $(limit)
     `,
-    group,
+    { afterId, limit },
   )
 }
 
-async function softDeleteIdentities(qx: QueryExecutor, ids: string[]): Promise<number> {
-  if (ids.length === 0) return 0
-
+async function lowercaseIdentityValue(qx: QueryExecutor, id: string, value: string): Promise<number> {
   return qx.result(
     `
       update "memberIdentities"
       set
-        "deletedAt" = now(),
+        value = $(value),
         "updatedAt" = now()
-      where id in ($(ids:csv))
+      where id = $(id)
         and "deletedAt" is null
+        and value <> $(value)
     `,
-    { ids },
+    { id, value },
   )
 }
 
-async function rewriteActivityRelationUsernames(
+async function rewriteActivityRelationUsernamesToLower(
   qx: QueryExecutor,
   memberId: string,
   platform: string,
-  keeperUsername: string,
-  deletedUsernames: string[],
+  lowerValue: string,
 ): Promise<{ usernameRows: number; objectMemberUsernameRows: number }> {
-  const values = deletedUsernames.filter((v) => v !== keeperUsername)
-  if (values.length === 0) {
-    return { usernameRows: 0, objectMemberUsernameRows: 0 }
-  }
-
   let usernameRows = 0
   let objectMemberUsernameRows = 0
   let updated: number
@@ -126,22 +81,22 @@ async function rewriteActivityRelationUsernames(
       `
         update "activityRelations"
         set
-          username = $(keeperUsername),
+          username = $(lowerValue),
           "updatedAt" = now()
         where "activityId" in (
           select "activityId"
           from "activityRelations"
           where "memberId" = $(memberId)
             and platform = $(platform)
-            and username in ($(values:csv))
+            and lower(username) = $(lowerValue)
+            and username <> $(lowerValue)
           limit $(batchSize)
         )
       `,
       {
         memberId,
         platform,
-        keeperUsername,
-        values,
+        lowerValue,
         batchSize: AR_UPDATE_BATCH_SIZE,
       },
     )
@@ -153,22 +108,22 @@ async function rewriteActivityRelationUsernames(
       `
         update "activityRelations"
         set
-          "objectMemberUsername" = $(keeperUsername),
+          "objectMemberUsername" = $(lowerValue),
           "updatedAt" = now()
         where "activityId" in (
           select "activityId"
           from "activityRelations"
           where "objectMemberId" = $(memberId)
             and platform = $(platform)
-            and "objectMemberUsername" in ($(values:csv))
+            and lower("objectMemberUsername") = $(lowerValue)
+            and "objectMemberUsername" <> $(lowerValue)
           limit $(batchSize)
         )
       `,
       {
         memberId,
         platform,
-        keeperUsername,
-        values,
+        lowerValue,
         batchSize: AR_UPDATE_BATCH_SIZE,
       },
     )
@@ -181,6 +136,7 @@ async function rewriteActivityRelationUsernames(
 setImmediate(async () => {
   const testRun = parameters.testRun ?? false
   const PROCESS_BATCH_LOG_EVERY = testRun ? 1 : 200
+  const batchSize = testRun ? 10 : LOAD_BATCH_SIZE
 
   const db = await getDbConnection({
     host: DB_CONFIG.writeHost,
@@ -192,79 +148,83 @@ setImmediate(async () => {
 
   const qx = pgpQx(db)
 
-  log.info({ testRun }, 'Running script with the following parameters!')
-
-  const groups = await findDuplicateCaseVariantGroups(qx, testRun ? 10 : undefined)
-  log.info({ groupCount: groups.length }, 'Loaded same-member case-variant groups!')
+  log.info({ testRun, batchSize }, 'Lowercasing email-shaped identity values!')
 
+  let afterId: string | null = null
   let processed = 0
-  let softDeleted = 0
+  let identitiesUpdated = 0
   let skipped = 0
   let arUsernameUpdated = 0
   let arObjectUsernameUpdated = 0
 
-  for (const group of groups) {
-    const rows = await fetchGroupIdentities(qx, group)
-
-    if (rows.length < 2) {
-      skipped += 1
-    } else {
-      const keeper = pickKeeper(rows)
-      const toDeleteRows = rows.filter((r) => r.id !== keeper.id)
-      const toDeleteIds = toDeleteRows.map((r) => r.id)
-      const deletedValues = toDeleteRows.map((r) => r.value)
+  for (;;) {
+    const candidates = await findMixedCaseEmailIdentities(qx, afterId, batchSize)
+    if (candidates.length === 0) {
+      break
+    }
 
-      if (testRun) {
-        log.info(
-          {
-            memberId: group.memberId,
-            platform: group.platform,
-            type: group.type,
-            keep: keeper.value,
-            softDelete: deletedValues,
-          },
-          'Soft-deleting case variants!',
-        )
+    afterId = candidates[candidates.length - 1].id
+
+    for (const row of candidates) {
+      const lowerValue = row.value.trim().toLowerCase()
+
+      if (isEmail(lowerValue) && lowerValue !== row.value) {
+        if (testRun) {
+          log.info(
+            {
+              memberId: row.memberId,
+              platform: row.platform,
+              type: row.type,
+              from: row.value,
+              to: lowerValue,
+            },
+            'Lowercasing email-shaped identity!',
+          )
+        }
+
+        const { updatedCount, ar } = await qx.tx(async (tx) => {
+          const updatedCount = await lowercaseIdentityValue(tx, row.id, lowerValue)
+          const ar = await rewriteActivityRelationUsernamesToLower(
+            tx,
+            row.memberId,
+            row.platform,
+            lowerValue,
+          )
+          return { updatedCount, ar }
+        })
+
+        identitiesUpdated += updatedCount
+        arUsernameUpdated += ar.usernameRows
+        arObjectUsernameUpdated += ar.objectMemberUsernameRows
+        processed += 1
+
+        if (processed % PROCESS_BATCH_LOG_EVERY === 0) {
+          log.info(
+            {
+              processed,
+              identitiesUpdated,
+              skipped,
+              arUsernameUpdated,
+              arObjectUsernameUpdated,
+            },
+            'Progress!',
+          )
+        }
+      } else {
+        skipped += 1
       }
+    }
 
-      const { deletedCount, ar } = await qx.tx(async (tx) => {
-        const deletedCount = await softDeleteIdentities(tx, toDeleteIds)
-        const ar = await rewriteActivityRelationUsernames(
-          tx,
-          group.memberId,
-          group.platform,
-          keeper.value,
-          deletedValues,
-        )
-        return { deletedCount, ar }
-      })
-
-      softDeleted += deletedCount
-      arUsernameUpdated += ar.usernameRows
-      arObjectUsernameUpdated += ar.objectMemberUsernameRows
-
-      processed += 1
-
-      if (processed % PROCESS_BATCH_LOG_EVERY === 0) {
-        log.info(
-          {
-            processed,
-            total: groups.length,
-            softDeleted,
-            skipped,
-            arUsernameUpdated,
-            arObjectUsernameUpdated,
-          },
-          'Progress!',
-        )
-      }
+    if (testRun) {
+      log.info('Test run - stopping after first batch!')
+      break
     }
   }
 
   log.info(
     {
       processed,
-      softDeleted,
+      identitiesUpdated,
       skipped,
       arUsernameUpdated,
       arObjectUsernameUpdated,