chore: lint & format

Author: mbani01

services/apps/cron_service/src/jobs/snowflakeS3Export.job.ts

@@ -1,10 +1,6 @@
 import CronTime from 'cron-time-generator'
 
-import {
-  TEMPORAL_CONFIG,
-  WorkflowIdReusePolicy,
-  getTemporalClient,
-} from '@crowd/temporal'
+import { TEMPORAL_CONFIG, WorkflowIdReusePolicy, getTemporalClient } from '@crowd/temporal'
 
 import { IJobDefinition } from '../types'
 

services/apps/snowflake_connectors/src/activities/exportActivity.ts

@@ -4,14 +4,14 @@
  * This activity is invoked by the exportWorkflow and performs
  * the actual Snowflake export and metadata bookkeeping.
  */
-
-import { PlatformType } from '@crowd/types'
+import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/database'
 import { getServiceChildLogger } from '@crowd/logging'
-import { getDbConnection, WRITE_DB_CONFIG } from '@crowd/database'
+import { PlatformType } from '@crowd/types'
 
-import { SnowflakeExporter } from '../core/snowflakeExporter'
 import { MetadataStore } from '../core/metadataStore'
+import { SnowflakeExporter } from '../core/snowflakeExporter'
 import { getPlatform } from '../integrations'
+
 export { getEnabledPlatforms } from '../integrations'
 
 const log = getServiceChildLogger('exportActivity')
@@ -21,7 +21,10 @@ function buildS3FilenamePrefix(platform: string): string {
   const year = now.getFullYear()
   const month = String(now.getMonth() + 1).padStart(2, '0')
   const day = String(now.getDate()).padStart(2, '0')
-  const s3BucketPath = process.env.CROWD_SNOWFLAKE_S3_BUCKET_PATH!
+  const s3BucketPath = process.env.CROWD_SNOWFLAKE_S3_BUCKET_PATH
+  if (!s3BucketPath) {
+    throw new Error('Missing required env var CROWD_SNOWFLAKE_S3_BUCKET_PATH')
+  }
   return `${s3BucketPath}/${platform}/${year}/${month}/${day}`
 }
 
@@ -53,6 +56,8 @@ export async function executeExport(platform: PlatformType): Promise<void> {
     log.error({ platform, err }, 'Export failed')
     throw err
   } finally {
-    await exporter.destroy().catch((err) => log.warn({ err }, 'Failed to close Snowflake connection'))
+    await exporter
+      .destroy()
+      .catch((err) => log.warn({ err }, 'Failed to close Snowflake connection'))
   }
 }

services/apps/snowflake_connectors/src/consumer/transformerConsumer.ts

@@ -4,18 +4,17 @@
  * Continuously polls the metadata store for pending jobs
  * that need transformation, then runs the appropriate transformer.
  */
-
-import { PlatformType } from '@crowd/types'
-import { getServiceChildLogger } from '@crowd/logging'
-import { getDbConnection, WRITE_DB_CONFIG } from '@crowd/database'
-import { QueueFactory, QUEUE_CONFIG } from '@crowd/queue'
-import { getRedisClient, REDIS_CONFIG, RedisCache } from '@crowd/redis'
 import { DataSinkWorkerEmitter } from '@crowd/common_services'
+import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/database'
+import { getServiceChildLogger } from '@crowd/logging'
+import { QUEUE_CONFIG, QueueFactory } from '@crowd/queue'
+import { REDIS_CONFIG, RedisCache, getRedisClient } from '@crowd/redis'
+import { PlatformType } from '@crowd/types'
 
+import { IntegrationResolver } from '../core/integrationResolver'
 import { MetadataStore, SnowflakeExportJob } from '../core/metadataStore'
 import { S3Consumer } from '../core/s3Consumer'
-import { IntegrationResolver } from '../core/integrationResolver'
-import { getPlatform, getEnabledPlatforms } from '../integrations'
+import { getEnabledPlatforms, getPlatform } from '../integrations'
 
 const log = getServiceChildLogger('transformerConsumer')
 
@@ -56,9 +55,15 @@ export class TransformerConsumer {
         continue
       }
 
-      log.info({ currentPollingIntervalMs: this.currentPollingIntervalMs }, 'No pending jobs, backing off')
+      log.info(
+        { currentPollingIntervalMs: this.currentPollingIntervalMs },
+        'No pending jobs, backing off',
+      )
       await this.sleep(this.currentPollingIntervalMs)
-      this.currentPollingIntervalMs = Math.min(this.currentPollingIntervalMs * 2, MAX_POLLING_INTERVAL_MS)
+      this.currentPollingIntervalMs = Math.min(
+        this.currentPollingIntervalMs * 2,
+        MAX_POLLING_INTERVAL_MS,
+      )
     }
 
     log.info('Transformer consumer stopped')
@@ -97,7 +102,11 @@ export class TransformerConsumer {
           continue
         }
 
-        await this.emitter.createAndProcessActivityResult(resolved.segmentId, resolved.integrationId, result.activity)
+        await this.emitter.createAndProcessActivityResult(
+          resolved.segmentId,
+          resolved.integrationId,
+          result.activity,
+        )
         transformedCount++
       }
 
@@ -115,7 +124,9 @@ export class TransformerConsumer {
       log.error({ jobId: job.id, err }, 'Job failed')
 
       try {
-        await this.metadataStore.markFailed(job.id, errorMessage, { processingDurationMs: Date.now() - startTime })
+        await this.metadataStore.markFailed(job.id, errorMessage, {
+          processingDurationMs: Date.now() - startTime,
+        })
       } catch (updateErr) {
         log.error({ jobId: job.id, updateErr }, 'Failed to mark job as failed')
       }

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

@@ -5,12 +5,11 @@
  * integration for the given platform + segment pair. Results are cached
  * in Redis to avoid repeated DB queries across consumer restarts.
  */
-
 import { DEFAULT_TENANT_ID, generateUUIDv4 } from '@crowd/common'
 import type { DbConnection } from '@crowd/database'
-import { PlatformType, IntegrationState } from '@crowd/types'
 import { getServiceChildLogger } from '@crowd/logging'
 import { RedisCache } from '@crowd/redis'
+import { IntegrationState, PlatformType } from '@crowd/types'
 
 import type { SegmentRef } from './transformerBase'
 
@@ -120,7 +119,10 @@ export class IntegrationResolver {
           await this.cache.delete(lockKey)
         }
       } else {
-        log.info({ platform, segmentId: segmentRow.id }, 'Lock held by another consumer, fetching existing')
+        log.info(
+          { platform, segmentId: segmentRow.id },
+          'Lock held by another consumer, fetching existing',
+        )
         integration = await this.db.oneOrNone<{ id: string }>(
           `SELECT id
            FROM integrations

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

@@ -4,7 +4,6 @@
  * Tracks export batches, processing status, and timestamps
  * to enable incremental exports and consumer polling.
  */
-
 import type { DbConnection } from '@crowd/database'
 
 export interface JobMetrics {
@@ -88,7 +87,6 @@ export class MetadataStore {
     return row ? mapRowToJob(row) : null
   }
 
-  
   // TODO: Add a cleanup workflow that deletes S3 files for completed/failed jobs
   // and sets "cleanedAt" to reclaim storage.
 
@@ -125,7 +123,6 @@ export class MetadataStore {
     )
     return row?.max ?? null
   }
-
 }
 
 function mapRowToJob(row: {

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

@@ -4,7 +4,6 @@
  * Responsible for reading exported files from S3
  * (e.g., Parquet manifests or raw data) for downstream transformation.
  */
-
 import { GetObjectCommand, S3Client } from '@aws-sdk/client-s3'
 import { ParquetReader } from '@dsnp/parquetjs'
 
@@ -18,19 +17,27 @@ export class S3Consumer {
   private readonly s3: S3Client
 
   constructor() {
+    const accessKeyId = process.env.CROWD_SNOWFLAKE_S3_ACCESS_KEY_ID
+    const secretAccessKey = process.env.CROWD_SNOWFLAKE_S3_SECRET_ACCESS_KEY
+    if (!accessKeyId || !secretAccessKey) {
+      throw new Error(
+        'Missing required env vars CROWD_SNOWFLAKE_S3_ACCESS_KEY_ID / CROWD_SNOWFLAKE_S3_SECRET_ACCESS_KEY',
+      )
+    }
+
     this.s3 = new S3Client({
       region: process.env.CROWD_SNOWFLAKE_S3_REGION,
-      credentials: {
-        accessKeyId: process.env.CROWD_SNOWFLAKE_S3_ACCESS_KEY_ID!,
-        secretAccessKey: process.env.CROWD_SNOWFLAKE_S3_SECRET_ACCESS_KEY!,
-      },
+      credentials: { accessKeyId, secretAccessKey },
     })
   }
 
   async downloadFile(s3Uri: string): Promise<Buffer> {
     const { bucket, key } = this.parseS3Uri(s3Uri)
     const response = await this.s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }))
-    const byteArray = await response.Body!.transformToByteArray()
+    if (!response.Body) {
+      throw new Error(`Empty response body for ${s3Uri}`)
+    }
+    const byteArray = await response.Body.transformToByteArray()
     return Buffer.from(byteArray)
   }
 
@@ -39,8 +46,8 @@ export class S3Consumer {
     const reader = await ParquetReader.openBuffer(buffer)
     const cursor = reader.getCursor()
     const rows: Record<string, unknown>[] = []
-    let row: Record<string, unknown> | null
-    while ((row = await cursor.next()) !== null) {
+    let row: Record<string, unknown> | null = null
+    while ((row = (await cursor.next()) as Record<string, unknown> | null) !== null) {
       rows.push(row)
     }
     await reader.close()

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

@@ -4,15 +4,18 @@
  * Responsible for executing COPY INTO queries against Snowflake
  * to export data into S3 as Parquet files.
  */
-
-import { SnowflakeClient } from '@crowd/snowflake'
 import { getServiceChildLogger } from '@crowd/logging'
+import { SnowflakeClient } from '@crowd/snowflake'
 
 const log = getServiceChildLogger('snowflakeExporter')
 
 const DEFAULT_BATCH_SIZE = 10_000
 
-export type OnBatchComplete = (s3Path: string, totalRows: number, totalBytes: number) => Promise<void>
+export type OnBatchComplete = (
+  s3Path: string,
+  totalRows: number,
+  totalBytes: number,
+) => Promise<void>
 
 interface CopyIntoRow {
   rows_unloaded: number
@@ -37,12 +40,16 @@ export class SnowflakeExporter {
     s3FilenamePrefix: string,
     onBatchComplete?: OnBatchComplete,
   ): Promise<void> {
-    const storageIntegration = process.env.CROWD_SNOWFLAKE_STORAGE_INTEGRATION!
+    const storageIntegration = process.env.CROWD_SNOWFLAKE_STORAGE_INTEGRATION
+    if (!storageIntegration) {
+      throw new Error('Missing required env var CROWD_SNOWFLAKE_STORAGE_INTEGRATION')
+    }
     const limit = DEFAULT_BATCH_SIZE
     let offset = 0
     let batch = 0
+    let hasMoreRows = true
 
-    while (true) {
+    while (hasMoreRows) {
       const filename = `batch_${batch}.parquet`
       const s3Path = `${s3FilenamePrefix}/${filename}`
 
@@ -58,13 +65,17 @@ export class SnowflakeExporter {
         OVERWRITE = TRUE
       `
 
-      log.info({ s3FilenamePrefix, storageIntegration, batch, offset, limit }, 'Executing COPY INTO batch')
+      log.info(
+        { s3FilenamePrefix, storageIntegration, batch, offset, limit },
+        'Executing COPY INTO batch',
+      )
 
       const results = await this.snowflake.run<CopyIntoRow>(copyQuery)
 
       if (results.length === 0) {
         log.info({ batch, totalRowsExported: offset }, 'No more rows to export')
-        break
+        hasMoreRows = false
+        continue
       }
 
       const batchRows = results.reduce((sum, r) => sum + r.rows_unloaded, 0)
@@ -76,8 +87,12 @@ export class SnowflakeExporter {
         await onBatchComplete(s3Path, batchRows, batchBytes)
       }
       if (batchRows < limit) {
-        log.info({ totalRowsExported: offset + batchRows }, 'Export finished (last batch was partial)')
-        break
+        log.info(
+          { totalRowsExported: offset + batchRows },
+          'Export finished (last batch was partial)',
+        )
+        hasMoreRows = false
+        continue
       }
 
       offset += limit

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

@@ -4,9 +4,8 @@
  * Each integration must implement this base class to define
  * how raw exported data is transformed into activities.
  */
-
-import { IActivityData, PlatformType } from '@crowd/types'
 import { getServiceChildLogger } from '@crowd/logging'
+import { IActivityData, PlatformType } from '@crowd/types'
 
 const log = getServiceChildLogger('transformer')
 
@@ -28,12 +27,12 @@ export abstract class TransformerBase {
    * Transform a single raw row from the S3 export into an activity
    * along with routing metadata. Returns null if the row should be skipped.
    */
-  abstract transformRow(row: Record<string, any>): TransformedActivity | null
+  abstract transformRow(row: Record<string, unknown>): TransformedActivity | null
 
   /**
    * Safe wrapper around transformRow that catches errors and returns null.
    */
-  safeTransformRow(row: Record<string, any>): TransformedActivity | null {
+  safeTransformRow(row: Record<string, unknown>): TransformedActivity | null {
     try {
       return this.transformRow(row)
     } catch (err) {

services/apps/snowflake_connectors/src/index.ts

@@ -1,11 +1,10 @@
 /**
  * Entry point: Start Temporal worker + transformer consumer loop.
  */
-
 import { getServiceChildLogger } from '@crowd/logging'
 
-import { svc } from './main'
 import { createTransformerConsumer } from './consumer/transformerConsumer'
+import { svc } from './main'
 
 const log = getServiceChildLogger('main')
 

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

@@ -30,9 +30,9 @@ const CDP_MATCHED_SEGMENTS = `
 
 export const buildSourceQuery = (sinceTimestamp?: string): string => {
   const excludeClause = TIMESTAMP_TZ_COLUMNS.join(', ')
-  const castClauses = TIMESTAMP_TZ_COLUMNS.map(
-    (col) => `er.${col}::TIMESTAMP_NTZ AS ${col}`,
-  ).join(', ')
+  const castClauses = TIMESTAMP_TZ_COLUMNS.map((col) => `er.${col}::TIMESTAMP_NTZ AS ${col}`).join(
+    ', ',
+  )
 
   const select = `
   SELECT