fix: code review suggestions

Author: mbani01

services/apps/snowflake_connectors/src/core/integrationResolver.ts

@@ -17,6 +17,8 @@ const log = getServiceChildLogger('integrationResolver')
 
 const CACHE_TTL_SECONDS = 24 * 60 * 60 // 24 hours
 const LOCK_TTL_SECONDS = 30
+const LOCK_WAIT_RETRIES = 5
+const LOCK_WAIT_BASE_MS = 200
 
 export interface ResolvedIntegration {
   segmentId: string
@@ -121,20 +123,32 @@ export class IntegrationResolver {
       } else {
         log.info(
           { platform, segmentId: segmentRow.id },
-          'Lock held by another consumer, fetching existing',
-        )
-        integration = await this.db.oneOrNone<{ id: string }>(
-          `SELECT id
-           FROM integrations
-           WHERE platform = $1
-             AND "segmentId" = $2
-             AND "deletedAt" IS NULL
-           LIMIT 1`,
-          [platform, segmentRow.id],
+          'Lock held by another consumer, waiting for commit',
         )
+        for (let attempt = 0; attempt < LOCK_WAIT_RETRIES; attempt++) {
+          await new Promise((r) => setTimeout(r, LOCK_WAIT_BASE_MS * 2 ** attempt))
+          integration = await this.db.oneOrNone<{ id: string }>(
+            `SELECT id
+             FROM integrations
+             WHERE platform = $1
+               AND "segmentId" = $2
+               AND "deletedAt" IS NULL
+             LIMIT 1`,
+            [platform, segmentRow.id],
+          )
+          if (integration) break
+        }
       }
     }
 
+    if (!integration) {
+      log.warn(
+        { platform, segmentId: segmentRow.id },
+        'Integration not found after lock contention, skipping',
+      )
+      return null
+    }
+
     const result: ResolvedIntegration = {
       segmentId: segmentRow.id,
       integrationId: integration.id,

services/apps/snowflake_connectors/src/integrations/cvent/transformer.ts

@@ -16,7 +16,7 @@ export class CventTransformer extends TransformerBase {
 
   transformRow(row: Record<string, unknown>): TransformedActivity | null {
     const userName = (row.USERNAME as string | null)?.trim() || null
-    const lfUsername = (row.LF_USERNAME as string | null)?.trim() || null
+    const lfUsername = (row.LFID as string | null)?.trim() || null
     const fullName = (row.FULL_NAME as string | null)?.trim() || null
     const firstName = (row.FIRST_NAME as string | null)?.trim() || null
     const lastName = (row.LAST_NAME as string | null)?.trim() || null

services/apps/snowflake_connectors/src/integrations/index.ts

@@ -40,15 +40,6 @@ export function getPlatform(platform: PlatformType): PlatformDefinition {
   return supported[platform]
 }
 
-export function getEnabledPlatformDefinitions(): PlatformDefinition[] {
-  return enabled.map((platform) => {
-    if (!supported[platform]) {
-      throw new Error(`Enabled platform not supported: ${platform}`)
-    }
-    return supported[platform]
-  })
-}
-
 export function getEnabledPlatforms(): PlatformType[] {
   return enabled
 }

services/apps/snowflake_connectors/src/workflows/exportWorkflow.ts

@@ -1,4 +1,5 @@
 import { executeChild, proxyActivities } from '@temporalio/workflow'
+import { PlatformType } from '@crowd/types'
 
 import type * as activities from '../activities/exportActivity'
 
@@ -35,6 +36,6 @@ export async function snowflakeS3ExportScheduler(): Promise<void> {
   await Promise.allSettled(childWorkflows)
 }
 
-export async function exportSnowflakeData(platform: string): Promise<void> {
+export async function exportSnowflakeData(platform: PlatformType): Promise<void> {
   await executeExport(platform)
 }

services/libs/types/src/enums/organizations.ts

@@ -12,6 +12,7 @@ export enum OrganizationSource {
   ENRICHMENT_CRUSTDATA = 'enrichment-crustdata',
   GITHUB = 'github',
   UI = 'ui',
+  CVENT = 'cvent',
 }
 
 export enum OrganizationMergeSuggestionType {