Member enrichment null byte sanitization, retry policy updates

services/apps/premium/members_enrichment_worker/src/schedules/getMembersToEnrich.ts

@@ -20,7 +20,11 @@ export const scheduleMembersEnrichment = async () => {
         type: 'startWorkflow',
         workflowType: getMembersToEnrich,
         taskQueue: 'members-enrichment',
-        workflowExecutionTimeout: '5 minutes',
+        retry: {
+          initialInterval: '15 seconds',
+          backoffCoefficient: 2,
+          maximumAttempts: 3,
+        },
         args: [
           {
             afterId: null,

services/apps/premium/members_enrichment_worker/src/types.ts

@@ -23,7 +23,11 @@ export type IMemberEnrichmentData = IMemberEnrichmentDataProgAI | IMemberEnrichm
 
 export interface IEnrichmentService {
   source: MemberEnrichmentSource
+
+  // cache rows with older updatedAt than this will be considered obsolete and will be re-enriched
   cacheObsoleteAfterSeconds: number
+
+  // can the source enrich using this input
   isEnrichableBySource(input: IEnrichmentSourceInput): boolean
 
   // what kind of identities can this source use as input

services/apps/premium/members_enrichment_worker/src/workflows/enrichMember.ts

@@ -16,6 +16,12 @@ const {
   normalizeEnrichmentData,
 } = proxyActivities<typeof activities>({
   startToCloseTimeout: '10 seconds',
+  retry: {
+    initialInterval: '5s',
+    backoffCoefficient: 2.0,
+    maximumInterval: '30s',
+    maximumAttempts: 4,
+  },
 })
 
 export async function enrichMember(

services/apps/premium/members_enrichment_worker/src/workflows/getMembersToEnrich.ts

@@ -20,7 +20,7 @@ const { getMembers } = proxyActivities<typeof activities>({
 })
 
 export async function getMembersToEnrich(args: IGetMembersForEnrichmentArgs): Promise<void> {
-  const MEMBER_ENRICHMENT_PER_RUN = 300
+  const MEMBER_ENRICHMENT_PER_RUN = 100
   const afterId = args?.afterId || null
   const sources = [MemberEnrichmentSource.PROGAI, MemberEnrichmentSource.CLEARBIT]
 
@@ -34,8 +34,8 @@ export async function getMembersToEnrich(args: IGetMembersForEnrichmentArgs): Pr
     members.map((member) => {
       return executeChild(enrichMember, {
         workflowId: 'member-enrichment/' + member.tenantId + '/' + member.id,
-        cancellationType: ChildWorkflowCancellationType.ABANDON,
-        parentClosePolicy: ParentClosePolicy.PARENT_CLOSE_POLICY_ABANDON,
+        cancellationType: ChildWorkflowCancellationType.WAIT_CANCELLATION_COMPLETED,
+        parentClosePolicy: ParentClosePolicy.PARENT_CLOSE_POLICY_REQUEST_CANCEL,
         workflowExecutionTimeout: '15 minutes',
         retry: {
           backoffCoefficient: 2,

services/libs/common/src/utils.ts

@@ -69,6 +69,9 @@ export class BatchProcessor<T> {
 export const escapeNullByte = (str: string | null | undefined): string =>
   str ? str.replace(/\0/g, 'u0000') : str
 
+export const redactNullByte = (str: string | null | undefined): string =>
+  str ? str.replace(/\\u0000|\0/g, '[NULL]') : ''
+
 export const dateEqualityChecker = (a, b) => {
   if (a instanceof Date) {
     a = a.toISOString()

services/libs/data-access-layer/src/old/apps/premium/members_enrichment_worker/index.ts

@@ -1,4 +1,4 @@
-import { generateUUIDv4 } from '@crowd/common'
+import { generateUUIDv4, redactNullByte } from '@crowd/common'
 import { DbConnOrTx, DbStore, DbTransaction } from '@crowd/database'
 import {
   IAttributes,
@@ -475,7 +475,7 @@ export async function insertMemberEnrichmentCacheDb<T>(
   memberId: string,
   source: MemberEnrichmentSource,
 ) {
-  const dataSanitized = data ? JSON.stringify(data) : null
+  const dataSanitized = data ? redactNullByte(JSON.stringify(data)) : null
   return tx.query(
     `INSERT INTO "memberEnrichmentCache" ("memberId", "data", "createdAt", "updatedAt", "source")
       VALUES ($1, $2, NOW(), NOW(), $3);`,
@@ -489,7 +489,7 @@ export async function updateMemberEnrichmentCacheDb<T>(
   memberId: string,
   source: MemberEnrichmentSource,
 ) {
-  const dataSanitized = data ? JSON.stringify(data) : null
+  const dataSanitized = data ? redactNullByte(JSON.stringify(data)) : null
   return tx.query(
     `UPDATE "memberEnrichmentCache"
       SET