Skip to content

Commit 2352f05

Browse files
themaroltthemarolt
authored
fix: member attributes should have non empty values and a filled default (#3449)
Signed-off-by: Uroš Marolt <uros@marolt.me>
1 parent e5fb896 commit 2352f05

5 files changed

Lines changed: 321 additions & 18 deletions

File tree

‎services/apps/data_sink_worker/src/bin/fix-member-attributes.ts‎

Lines changed: 293 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,293 @@
1+
import isEqual from 'lodash.isequal'
2+
import mergeWith from 'lodash.mergewith'
3+
4+
import { connQx, updateMember } from '@crowd/data-access-layer'
5+
import {
6+
DbConnOrTx,
7+
DbStore,
8+
WRITE_DB_CONFIG,
9+
getDbConnection,
10+
} from '@crowd/data-access-layer/src/database'
11+
import { getServiceLogger } from '@crowd/logging'
12+
import { REDIS_CONFIG, getRedisClient } from '@crowd/redis'
13+
14+
import MemberAttributeService from '../service/memberAttribute.service'
15+
16+
/* eslint-disable @typescript-eslint/no-explicit-any */
17+
18+
const BATCH_SIZE = process.env.TEST_RUN ? 1 : 5000
19+
20+
const log = getServiceLogger()
21+
22+
async function getMemberIds(
23+
db: DbConnOrTx,
24+
lastId?: string,
25+
): Promise<{ id: string; attributes: any; manuallyChangedFields: any }[]> {
26+
try {
27+
log.debug({ lastId }, 'Querying for members with attribute issues')
28+
29+
const results = await db.any(
30+
`
31+
with relevant_members as (with member_with_attributes as (select id,
32+
"createdAt",
33+
jsonb_object_keys(attributes) as attr_key,
34+
attributes -> jsonb_object_keys(attributes) as attr_value
35+
from members
36+
where "deletedAt" is null
37+
and attributes is not null
38+
and attributes != 'null'::jsonb
39+
and attributes != '{}'::jsonb)
40+
select distinct id
41+
from member_with_attributes
42+
where jsonb_typeof(attr_value) = 'object'
43+
and coalesce(attr_value ->> 'default', '') = ''
44+
and exists (select 1
45+
from jsonb_each_text(attr_value) as kv
46+
where kv.key != 'default'
47+
and coalesce(kv.value, '') != '')
48+
${lastId ? `and id < '${lastId}'` : ''}
49+
order by id desc
50+
limit ${BATCH_SIZE})
51+
select m.id, m.attributes, m."manuallyChangedFields"
52+
from members m
53+
inner join
54+
relevant_members rm on rm.id = m.id
55+
order by m.id desc;
56+
`,
57+
{ lastId },
58+
)
59+
60+
log.debug(
61+
{
62+
resultCount: results.length,
63+
lastId,
64+
firstId: results.length > 0 ? results[0].id : null,
65+
lastResultId: results.length > 0 ? results[results.length - 1].id : null,
66+
},
67+
'Query completed',
68+
)
69+
70+
return results
71+
} catch (error) {
72+
log.error(
73+
{
74+
error: error.message,
75+
lastId,
76+
stack: error.stack,
77+
},
78+
'Failed to query member IDs',
79+
)
80+
throw error
81+
}
82+
}
83+
84+
setImmediate(async () => {
85+
let dbClient: DbConnOrTx | undefined
86+
let redisClient: any | undefined
87+
let exitCode = 0
88+
89+
try {
90+
log.info('Starting member attributes fix script')
91+
92+
// Initialize connections
93+
log.info('Connecting to database...')
94+
dbClient = await getDbConnection(WRITE_DB_CONFIG())
95+
log.info('Database connection established')
96+
97+
log.info('Connecting to Redis...')
98+
redisClient = await getRedisClient(REDIS_CONFIG())
99+
log.info('Redis connection established')
100+
101+
const pgQx = connQx(dbClient)
102+
const mas = new MemberAttributeService(redisClient, new DbStore(log, dbClient), log)
103+
104+
let totalProcessed = 0
105+
let totalUpdated = 0
106+
let batchNumber = 1
107+
108+
log.info('Starting to process members with attribute issues')
109+
let membersToFix = await getMemberIds(dbClient)
110+
log.info({ count: membersToFix.length }, 'Found members to process in first batch')
111+
112+
while (membersToFix.length > 0) {
113+
log.info({ batchNumber, batchSize: membersToFix.length }, 'Processing batch')
114+
let batchUpdated = 0
115+
116+
for (const data of membersToFix) {
117+
try {
118+
if (data.attributes) {
119+
log.debug(
120+
{ memberId: data.id, oldAttributes: data.attributes },
121+
'Processing member attributes',
122+
)
123+
124+
// check if any has default empty but other are full
125+
let toProcess = false
126+
for (const attName of Object.keys(data.attributes)) {
127+
const defValue = data.attributes[attName].default
128+
129+
if (defValue === undefined || defValue === null || String(defValue) === '') {
130+
log.debug(
131+
{
132+
memberId: data.id,
133+
attribute: data.attributes[attName],
134+
attName,
135+
defValue: defValue ? String(defValue) : 'undefined',
136+
},
137+
'Attribute has default empty',
138+
)
139+
for (const platform of Object.keys(data.attributes[attName]).filter(
140+
(p) => p !== 'default',
141+
)) {
142+
const value = data.attributes[attName][platform]
143+
144+
if (value !== undefined && value !== null && String(value) !== '') {
145+
log.debug(
146+
{ memberId: data.id, attName, platform, value },
147+
'Found value for attribute',
148+
)
149+
toProcess = true
150+
break
151+
}
152+
}
153+
154+
if (toProcess) {
155+
break
156+
}
157+
}
158+
}
159+
160+
if (toProcess) {
161+
const oldAttributes = JSON.parse(JSON.stringify(data.attributes)) // Deep copy
162+
data.attributes = await mas.setAttributesDefaultValues(data.attributes)
163+
164+
let attributes: Record<string, unknown> | undefined
165+
const temp = mergeWith({}, oldAttributes, data.attributes)
166+
const manuallyChangedFields: string[] = data.manuallyChangedFields || []
167+
168+
if (manuallyChangedFields.length > 0) {
169+
log.warn(
170+
{
171+
memberId: data.id,
172+
manuallyChangedFieldsCount: manuallyChangedFields.length,
173+
},
174+
'Member has manually changed fields',
175+
)
176+
177+
const prefix = 'attributes.'
178+
const manuallyChangedAttributes = [
179+
...new Set(
180+
manuallyChangedFields
181+
.filter((f) => f.startsWith(prefix))
182+
.map((f) => f.slice(prefix.length)),
183+
),
184+
]
185+
186+
log.warn(
187+
{
188+
memberId: data.id,
189+
manuallyChangedAttributes,
190+
},
191+
'Preserving manually changed attributes',
192+
)
193+
194+
// Preserve manually changed attributes
195+
for (const key of manuallyChangedAttributes) {
196+
if (oldAttributes?.[key] !== undefined) {
197+
temp[key] = oldAttributes[key] // Fixed: removed .attributes
198+
}
199+
}
200+
}
201+
202+
if (!isEqual(temp, oldAttributes)) {
203+
attributes = temp
204+
log.info({ memberId: data.id }, 'Attributes changed, will update')
205+
} else {
206+
log.debug(
207+
{ memberId: data.id, newAttributes: temp, oldAttributes },
208+
'No changes needed for attributes',
209+
)
210+
}
211+
212+
if (attributes) {
213+
log.info({ memberId: data.id }, 'Updating member attributes')
214+
215+
if (!process.env.TEST_RUN) {
216+
await updateMember(pgQx, data.id, { attributes } as any)
217+
}
218+
219+
batchUpdated++
220+
totalUpdated++
221+
log.debug({ memberId: data.id }, 'Member attributes updated successfully')
222+
}
223+
} else {
224+
log.debug(
225+
{ memberId: data.id, attributes: data.attributes },
226+
'No changes needed for attributes',
227+
)
228+
}
229+
} else {
230+
log.debug({ memberId: data.id }, 'Member has no attributes to process')
231+
}
232+
233+
totalProcessed++
234+
} catch (error) {
235+
log.error(
236+
{
237+
error: error.message,
238+
memberId: data.id,
239+
stack: error.stack,
240+
},
241+
'Failed to process member',
242+
)
243+
// Continue processing other members
244+
}
245+
}
246+
247+
log.info(
248+
{
249+
batchNumber,
250+
batchProcessed: membersToFix.length,
251+
batchUpdated,
252+
totalProcessed,
253+
totalUpdated,
254+
},
255+
'Completed batch processing',
256+
)
257+
258+
// Get next batch
259+
const lastId = membersToFix[membersToFix.length - 1].id
260+
log.debug({ lastId }, 'Fetching next batch starting from last ID')
261+
262+
if (process.env.TEST_RUN) {
263+
break
264+
}
265+
266+
membersToFix = await getMemberIds(dbClient, lastId)
267+
log.info({ count: membersToFix.length }, 'Found members for next batch')
268+
269+
batchNumber++
270+
}
271+
272+
log.info(
273+
{
274+
totalProcessed,
275+
totalUpdated,
276+
totalBatches: batchNumber - 1,
277+
},
278+
'Member attributes fix completed successfully',
279+
)
280+
} catch (error) {
281+
log.error(
282+
{
283+
error: error.message,
284+
stack: error.stack,
285+
},
286+
'Fatal error in member attributes fix script',
287+
)
288+
exitCode = 1
289+
} finally {
290+
log.info('Script execution completed')
291+
process.exit(exitCode)
292+
}
293+
})

‎services/apps/data_sink_worker/src/service/activity.service.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1068,7 +1068,7 @@ export default class ActivityService extends LoggerBase {
10681068
segmentIds: Set<string>
10691069
resultIds: Set<string>
10701070
integrationId: string
1071-
platform: string
1071+
platform: PlatformType
10721072
username: string
10731073
timestamp: string
10741074
}

‎services/apps/data_sink_worker/src/service/member.service.ts‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ export default class MemberService extends LoggerBase {
6565
segmentIds: string[],
6666
integrationId: string,
6767
data: IMemberCreateData,
68-
source: string,
68+
platform: PlatformType,
6969
releaseMemberLock?: () => Promise<void>,
7070
): Promise<string> {
7171
return logExecutionTimeV2(
@@ -89,8 +89,7 @@ export default class MemberService extends LoggerBase {
8989
let attributes: Record<string, unknown> = {}
9090
if (data.attributes) {
9191
attributes = await logExecutionTimeV2(
92-
() =>
93-
memberAttributeService.validateAttributes(data.attributes, source as PlatformType),
92+
() => memberAttributeService.validateAttributes(platform, data.attributes),
9493
this.log,
9594
'memberService -> create -> validateAttributes',
9695
)
@@ -190,7 +189,7 @@ export default class MemberService extends LoggerBase {
190189
if (data.organizations) {
191190
for (const org of data.organizations) {
192191
const id = await logExecutionTimeV2(
193-
() => orgService.findOrCreate(source, integrationId, org),
192+
() => orgService.findOrCreate(platform, integrationId, org),
194193
this.log,
195194
'memberService -> create -> findOrCreateOrg',
196195
)
@@ -265,7 +264,7 @@ export default class MemberService extends LoggerBase {
265264
data: IMemberUpdateData,
266265
original: IDbMember,
267266
originalIdentities: IMemberIdentity[],
268-
source: string,
267+
platform: PlatformType,
269268
releaseMemberLock?: () => Promise<void>,
270269
): Promise<void> {
271270
await logExecutionTimeV2(
@@ -282,8 +281,7 @@ export default class MemberService extends LoggerBase {
282281
if (data.attributes) {
283282
this.log.trace({ memberId: id }, 'Validating member attributes!')
284283
data.attributes = await logExecutionTimeV2(
285-
() =>
286-
memberAttributeService.validateAttributes(data.attributes, source as PlatformType),
284+
() => memberAttributeService.validateAttributes(platform, data.attributes),
287285
this.log,
288286
'memberService -> update -> validateAttributes',
289287
)
@@ -403,7 +401,7 @@ export default class MemberService extends LoggerBase {
403401
this.log.trace({ memberId: id }, 'Finding or creating organization!')
404402

405403
const orgId = await logExecutionTimeV2(
406-
() => orgService.findOrCreate(source, integrationId, org),
404+
() => orgService.findOrCreate(platform, integrationId, org),
407405
this.log,
408406
'memberService -> update -> findOrCreateOrg',
409407
)

‎services/apps/data_sink_worker/src/service/memberAttribute.service.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,8 +26,8 @@ export default class MemberAttributeService extends LoggerBase {
2626
}
2727

2828
public async validateAttributes(
29-
attributes: Record<string, unknown>,
3029
platform: PlatformType,
30+
attributes: Record<string, unknown>,
3131
): Promise<Record<string, unknown>> {
3232
const settings = await getMemberAttributeSettings(dbStoreQx(this.store), this.redis)
3333
const memberAttributeSettings = settings.reduce((acc, attribute) => {

0 commit comments

Comments
 (0)