Skip to content

Commit f676cde

Browse files
skwowetskwowet
committed
chore: add script to clean same-member case-variant identities
Signed-off-by: Yeganathan S <63534555+skwowet@users.noreply.github.com>
1 parent e8081d8 commit f676cde

2 files changed

Lines changed: 277 additions & 0 deletions

File tree

‎backend/package.json‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
"script:fix-duplicate-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/fix-duplicate-members.ts",
3434
"script:fix-members-activities-after-unaffilation": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/fix-members-activities-after-unaffilation.ts",
3535
"script:process-bot-members": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/process-bot-members.ts",
36+
"script:cleanup-same-member-case-variant-identities": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/cleanup-same-member-case-variant-identities.ts",
3637
"script:backfill-email-domain-member-organization-dates": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/backfill-email-domain-member-organization-dates.ts",
3738
"script:onboard-default-tenant": "SERVICE=script TS_NODE_TRANSPILE_ONLY=true tsx src/bin/scripts/onboard-default-tenant.ts",
3839
"script:onboard-default-tenant:local": "set -a && . ./.env.dist.local && . ./.env.override.local && set +a && pnpm run script:onboard-default-tenant",

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

Lines changed: 276 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,276 @@
1+
import commandLineArgs from 'command-line-args'
2+
3+
import { pgpQx } from '@crowd/data-access-layer'
4+
import { getDbConnection } from '@crowd/data-access-layer/src/database'
5+
import { QueryExecutor } from '@crowd/data-access-layer/src/queryExecutor'
6+
import { getServiceLogger } from '@crowd/logging'
7+
8+
import { DB_CONFIG } from '@/conf'
9+
10+
const log = getServiceLogger()
11+
12+
const options = [
13+
{
14+
name: 'testRun',
15+
alias: 't',
16+
type: Boolean,
17+
},
18+
]
19+
20+
const parameters = commandLineArgs(options)
21+
22+
type CaseVariantGroup = {
23+
memberId: string
24+
platform: string
25+
type: string
26+
lv: string
27+
}
28+
29+
type IdentityRow = {
30+
id: string
31+
value: string
32+
verified: boolean
33+
verifiedBy: string | null
34+
updatedAt: Date
35+
}
36+
37+
const AR_UPDATE_BATCH_SIZE = 5000
38+
39+
function pickKeeper(rows: IdentityRow[]): IdentityRow {
40+
return [...rows].sort((a, b) => {
41+
if (a.verified !== b.verified) return a.verified ? -1 : 1
42+
if (Boolean(a.verifiedBy) !== Boolean(b.verifiedBy)) return a.verifiedBy ? -1 : 1
43+
const aUpdated = new Date(a.updatedAt).getTime()
44+
const bUpdated = new Date(b.updatedAt).getTime()
45+
if (aUpdated !== bUpdated) return bUpdated - aUpdated
46+
return a.id < b.id ? -1 : 1
47+
})[0]
48+
}
49+
50+
async function findDuplicateCaseVariantGroups(
51+
qx: QueryExecutor,
52+
limit?: number,
53+
): Promise<CaseVariantGroup[]> {
54+
const baseQuery = `
55+
select
56+
"memberId",
57+
platform,
58+
type,
59+
lower(value) as lv
60+
from "memberIdentities"
61+
where "deletedAt" is null
62+
group by "memberId", platform, type, lower(value)
63+
having count(distinct value) > 1
64+
order by "memberId", platform, type, lower(value)
65+
`
66+
67+
if (limit != null) {
68+
return qx.select(`${baseQuery} limit $(limit)`, { limit })
69+
}
70+
71+
return qx.select(baseQuery)
72+
}
73+
74+
async function fetchGroupIdentities(
75+
qx: QueryExecutor,
76+
group: CaseVariantGroup,
77+
): Promise<IdentityRow[]> {
78+
return qx.select(
79+
`
80+
select id, value, verified, "verifiedBy", "updatedAt"
81+
from "memberIdentities"
82+
where "memberId" = $(memberId)
83+
and platform = $(platform)
84+
and type = $(type)
85+
and lower(value) = $(lv)
86+
and "deletedAt" is null
87+
`,
88+
group,
89+
)
90+
}
91+
92+
async function softDeleteIdentities(qx: QueryExecutor, ids: string[]): Promise<number> {
93+
if (ids.length === 0) return 0
94+
95+
return qx.result(
96+
`
97+
update "memberIdentities"
98+
set
99+
"deletedAt" = now(),
100+
"updatedAt" = now()
101+
where id in ($(ids:csv))
102+
and "deletedAt" is null
103+
`,
104+
{ ids },
105+
)
106+
}
107+
108+
async function rewriteActivityRelationUsernames(
109+
qx: QueryExecutor,
110+
memberId: string,
111+
platform: string,
112+
keeperUsername: string,
113+
deletedUsernames: string[],
114+
): Promise<{ usernameRows: number; objectMemberUsernameRows: number }> {
115+
const values = deletedUsernames.filter((v) => v !== keeperUsername)
116+
if (values.length === 0) {
117+
return { usernameRows: 0, objectMemberUsernameRows: 0 }
118+
}
119+
120+
let usernameRows = 0
121+
let objectMemberUsernameRows = 0
122+
let updated: number
123+
124+
do {
125+
updated = await qx.result(
126+
`
127+
update "activityRelations"
128+
set
129+
username = $(keeperUsername),
130+
"updatedAt" = now()
131+
where "activityId" in (
132+
select "activityId"
133+
from "activityRelations"
134+
where "memberId" = $(memberId)
135+
and platform = $(platform)
136+
and username in ($(values:csv))
137+
limit $(batchSize)
138+
)
139+
`,
140+
{
141+
memberId,
142+
platform,
143+
keeperUsername,
144+
values,
145+
batchSize: AR_UPDATE_BATCH_SIZE,
146+
},
147+
)
148+
usernameRows += updated
149+
} while (updated === AR_UPDATE_BATCH_SIZE)
150+
151+
do {
152+
updated = await qx.result(
153+
`
154+
update "activityRelations"
155+
set
156+
"objectMemberUsername" = $(keeperUsername),
157+
"updatedAt" = now()
158+
where "activityId" in (
159+
select "activityId"
160+
from "activityRelations"
161+
where "objectMemberId" = $(memberId)
162+
and platform = $(platform)
163+
and "objectMemberUsername" in ($(values:csv))
164+
limit $(batchSize)
165+
)
166+
`,
167+
{
168+
memberId,
169+
platform,
170+
keeperUsername,
171+
values,
172+
batchSize: AR_UPDATE_BATCH_SIZE,
173+
},
174+
)
175+
objectMemberUsernameRows += updated
176+
} while (updated === AR_UPDATE_BATCH_SIZE)
177+
178+
return { usernameRows, objectMemberUsernameRows }
179+
}
180+
181+
setImmediate(async () => {
182+
const testRun = parameters.testRun ?? false
183+
const PROCESS_BATCH_LOG_EVERY = testRun ? 1 : 200
184+
185+
const db = await getDbConnection({
186+
host: DB_CONFIG.writeHost,
187+
port: DB_CONFIG.port,
188+
database: DB_CONFIG.database,
189+
user: DB_CONFIG.username,
190+
password: DB_CONFIG.password,
191+
})
192+
193+
const qx = pgpQx(db)
194+
195+
log.info({ testRun }, 'Running script with the following parameters!')
196+
197+
const groups = await findDuplicateCaseVariantGroups(qx, testRun ? 10 : undefined)
198+
log.info({ groupCount: groups.length }, 'Loaded same-member case-variant groups!')
199+
200+
let processed = 0
201+
let softDeleted = 0
202+
let skipped = 0
203+
let arUsernameUpdated = 0
204+
let arObjectUsernameUpdated = 0
205+
206+
for (const group of groups) {
207+
const rows = await fetchGroupIdentities(qx, group)
208+
209+
if (rows.length < 2) {
210+
skipped += 1
211+
} else {
212+
const keeper = pickKeeper(rows)
213+
const toDeleteRows = rows.filter((r) => r.id !== keeper.id)
214+
const toDeleteIds = toDeleteRows.map((r) => r.id)
215+
const deletedValues = toDeleteRows.map((r) => r.value)
216+
217+
if (testRun) {
218+
log.info(
219+
{
220+
memberId: group.memberId,
221+
platform: group.platform,
222+
type: group.type,
223+
keep: keeper.value,
224+
softDelete: deletedValues,
225+
},
226+
'Soft-deleting case variants!',
227+
)
228+
}
229+
230+
const { deletedCount, ar } = await qx.tx(async (tx) => {
231+
const deletedCount = await softDeleteIdentities(tx, toDeleteIds)
232+
const ar = await rewriteActivityRelationUsernames(
233+
tx,
234+
group.memberId,
235+
group.platform,
236+
keeper.value,
237+
deletedValues,
238+
)
239+
return { deletedCount, ar }
240+
})
241+
242+
softDeleted += deletedCount
243+
arUsernameUpdated += ar.usernameRows
244+
arObjectUsernameUpdated += ar.objectMemberUsernameRows
245+
246+
processed += 1
247+
248+
if (processed % PROCESS_BATCH_LOG_EVERY === 0) {
249+
log.info(
250+
{
251+
processed,
252+
total: groups.length,
253+
softDeleted,
254+
skipped,
255+
arUsernameUpdated,
256+
arObjectUsernameUpdated,
257+
},
258+
'Progress!',
259+
)
260+
}
261+
}
262+
}
263+
264+
log.info(
265+
{
266+
processed,
267+
softDeleted,
268+
skipped,
269+
arUsernameUpdated,
270+
arObjectUsernameUpdated,
271+
},
272+
'Done!',
273+
)
274+
275+
process.exit(0)
276+
})

0 commit comments

Comments
 (0)