chore: replace cronjob with temporal schedule

Author: mbani01

services/apps/snowflake_connectors/package.json

@@ -9,7 +9,8 @@
     "lint": "npx eslint --ext .ts src --max-warnings=0",
     "format": "npx prettier --write \"src/**/*.ts\"",
     "format-check": "npx prettier --check .",
-    "tsc-check": "tsc --noEmit"
+    "tsc-check": "tsc --noEmit",
+    "trigger-export": "SERVICE=snowflake-connectors tsx src/scripts/triggerExport.ts"
   },
   "dependencies": {
     "@crowd/archetype-standard": "workspace:*",
@@ -20,11 +21,13 @@
     "@crowd/archetype-worker": "workspace:*",
     "@crowd/queue": "workspace:*",
     "@crowd/redis": "workspace:*",
+    "@crowd/slack": "workspace:*",
     "@crowd/snowflake": "workspace:*",
     "@crowd/temporal": "workspace:*",
     "@crowd/types": "workspace:*",
     "@aws-sdk/client-s3": "^3.700.0",
     "@dsnp/parquetjs": "^1.7.0",
+    "@temporalio/client": "~1.11.8",
     "@temporalio/workflow": "~1.11.8",
     "tsx": "^4.7.1",
     "typescript": "^5.6.3"

services/apps/snowflake_connectors/src/index.ts

@@ -5,12 +5,15 @@ import { getServiceChildLogger } from '@crowd/logging'
 
 import { createTransformerConsumer } from './consumer/transformerConsumer'
 import { svc } from './main'
+import { scheduleSnowflakeS3Export } from './schedules/snowflakeS3Export'
 
 const log = getServiceChildLogger('main')
 
 setImmediate(async () => {
   await svc.init()
 
+  await scheduleSnowflakeS3Export()
+
   const consumer = await createTransformerConsumer()
   consumer.start()
 

services/apps/snowflake_connectors/src/main.ts

@@ -20,7 +20,7 @@ const config: Config = {
     enabled: false,
   },
   temporal: {
-    enabled: false,
+    enabled: true,
   },
   redis: {
     enabled: true,

services/apps/snowflake_connectors/src/schedules/snowflakeS3Export.ts

@@ -0,0 +1,45 @@
+import { ScheduleAlreadyRunning, ScheduleOverlapPolicy } from '@temporalio/client'
+
+import { SlackChannel, SlackPersona, sendSlackNotification } from '@crowd/slack'
+
+import { svc } from '../main'
+import { snowflakeS3ExportScheduler } from '../workflows'
+
+export const scheduleSnowflakeS3Export = async () => {
+  try {
+    await svc.temporal.schedule.create({
+      scheduleId: 'snowflake-s3-export',
+      spec: {
+        cronExpressions: ['0 0 * * *'],
+      },
+      policies: {
+        overlap: ScheduleOverlapPolicy.SKIP,
+        catchupWindow: '1 minute',
+      },
+      action: {
+        type: 'startWorkflow',
+        workflowType: snowflakeS3ExportScheduler,
+        taskQueue: 'snowflakeConnectors',
+        retry: {
+          initialInterval: '15 seconds',
+          backoffCoefficient: 2,
+          maximumAttempts: 3,
+        },
+        args: [],
+      },
+    })
+  } catch (err) {
+    if (err instanceof ScheduleAlreadyRunning) {
+      svc.log.info('Schedule already registered in Temporal.')
+      svc.log.info('Configuration may have changed since. Please make sure they are in sync.')
+    } else {
+      svc.log.error({ err }, 'Failed to create snowflake-s3-export schedule')
+      sendSlackNotification(
+        SlackChannel.INTEGRATION_NOTIFICATIONS,
+        SlackPersona.ERROR_REPORTER,
+        'Snowflake S3 Export Schedule Failed',
+        `Failed to create the \`snowflake-s3-export\` Temporal schedule.\n\n*Error:* ${err.message || err}`,
+      )
+    }
+  }
+}

services/apps/snowflake_connectors/src/scripts/triggerExport.ts

@@ -0,0 +1,26 @@
+import { TEMPORAL_CONFIG, getTemporalClient } from '../config/settings'
+
+async function main() {
+  const client = await getTemporalClient(TEMPORAL_CONFIG())
+
+  const workflowId = `snowflake-export/manual/${new Date().toISOString().slice(0, 19)}`
+
+  await client.workflow.start('snowflakeS3ExportScheduler', {
+    taskQueue: 'snowflakeConnectors',
+    workflowId,
+    retry: {
+      initialInterval: '15 seconds',
+      backoffCoefficient: 2,
+      maximumAttempts: 3,
+    },
+    args: [],
+  })
+
+  console.log(`Snowflake S3 export workflow started: ${workflowId}`)
+  process.exit(0)
+}
+
+main().catch((err) => {
+  console.error('Failed to trigger workflow:', err)
+  process.exit(1)
+})