feat: snowflake export worker [CM-975] (#3844)

Author: mbani01

backend/src/database/migrations/U1771344764__addSnowflakeExportTable.sql

@@ -0,0 +1,17 @@
+CREATE TABLE integration."snowflakeExportJobs" (
+  id BIGSERIAL PRIMARY KEY,
+  platform VARCHAR(100) NOT NULL,
+  s3_path TEXT NOT NULL UNIQUE,
+  metrics JSONB,
+  "createdAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
+  "updatedAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
+  "processingStartedAt" TIMESTAMP WITH TIME ZONE, -- set when worker claims job (acts as lock)
+  "exportStartedAt" TIMESTAMP WITH TIME ZONE,
+  "completedAt" TIMESTAMP WITH TIME ZONE,
+  "cleanedAt" TIMESTAMP WITH TIME ZONE,
+  error TEXT
+);
+
+CREATE INDEX "idx_snowflakeExportJobs_platform" ON integration."snowflakeExportJobs" (platform);
+CREATE INDEX "idx_snowflakeExportJobs_pending" ON integration."snowflakeExportJobs" ("createdAt")
+  WHERE "processingStartedAt" IS NULL;

backend/src/database/migrations/U1771497876__addCventActivityTypes.sql

@@ -0,0 +1,3 @@
+INSERT INTO "activityTypes" ("activityType", platform, "isCodeContribution", "isCollaboration", description, "label") VALUES
+('registered-event', 'cvent', false, false, 'User registers to an event', 'Registered for an event'),
+('attended-event', 'cvent', false, false, 'User attends an event', 'Attended an event');

backend/src/database/migrations/V1771344764__addSnowflakeExportTable.sql

@@ -0,0 +1,25 @@
+FROM node:20-bullseye-slim AS builder
+
+RUN apt-get update && apt-get install -y python3 make g++ && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /usr/crowd/app
+RUN npm install -g corepack@latest && corepack enable pnpm && corepack prepare pnpm@9.15.0 --activate
+
+COPY ./pnpm-workspace.yaml ./pnpm-lock.yaml ./
+RUN pnpm fetch
+
+COPY ./services ./services
+RUN pnpm i --frozen-lockfile
+
+FROM node:20-bullseye-slim AS runner
+
+RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /usr/crowd/app
+RUN npm install -g corepack@latest && corepack enable pnpm && corepack prepare pnpm@9.15.0 --activate
+
+COPY --from=builder /usr/crowd/app/node_modules ./node_modules
+COPY --from=builder /usr/crowd/app/services/base.tsconfig.json ./services/base.tsconfig.json
+COPY --from=builder /usr/crowd/app/services/libs ./services/libs
+COPY --from=builder /usr/crowd/app/services/archetypes/ ./services/archetypes
+COPY --from=builder /usr/crowd/app/services/apps/snowflake_connectors/ ./services/apps/snowflake_connectors

backend/src/database/migrations/V1771497876__addCventActivityTypes.sql

@@ -0,0 +1,18 @@
+**/.git
+**/node_modules
+**/venv*
+**/.webpack
+**/.serverless
+**/.env
+**/.env.*
+**/.idea
+**/.vscode
+**/dist
+.vscode/
+.github/
+frontend/
+scripts/
+.flake8
+*.md
+Makefile
+backend/

pnpm-lock.yaml

@@ -0,0 +1,54 @@
+version: '3.1'
+
+x-env-args: &env-args
+  DOCKER_BUILDKIT: 1
+  NODE_ENV: docker
+  SERVICE: snowflake-connectors
+  CROWD_TEMPORAL_TASKQUEUE: snowflakeConnectors
+  SHELL: /bin/sh
+
+services:
+  snowflake-connectors:
+    build:
+      context: ../../
+      dockerfile: ./scripts/services/docker/Dockerfile.snowflake_connectors
+    command: 'pnpm run start'
+    working_dir: /usr/crowd/app/services/apps/snowflake_connectors
+    env_file:
+      - ../../backend/.env.dist.local
+      - ../../backend/.env.dist.composed
+      - ../../backend/.env.override.local
+      - ../../backend/.env.override.composed
+    environment:
+      <<: *env-args
+    restart: always
+    networks:
+      - crowd-bridge
+
+  snowflake-connectors-dev:
+    build:
+      context: ../../
+      dockerfile: ./scripts/services/docker/Dockerfile.snowflake_connectors
+    command: 'pnpm run dev'
+    working_dir: /usr/crowd/app/services/apps/snowflake_connectors
+    env_file:
+      - ../../backend/.env.dist.local
+      - ../../backend/.env.dist.composed
+      - ../../backend/.env.override.local
+      - ../../backend/.env.override.composed
+    environment:
+      <<: *env-args
+    hostname: snowflake-connectors
+    networks:
+      - crowd-bridge
+    volumes:
+      - ../../services/libs/common/src:/usr/crowd/app/services/libs/common/src
+      - ../../services/libs/logging/src:/usr/crowd/app/services/libs/logging/src
+      - ../../services/libs/snowflake/src:/usr/crowd/app/services/libs/snowflake/src
+      - ../../services/libs/temporal/src:/usr/crowd/app/services/libs/temporal/src
+      - ../../services/libs/types/src:/usr/crowd/app/services/libs/types/src
+      - ../../services/apps/snowflake_connectors/src:/usr/crowd/app/services/apps/snowflake_connectors/src
+
+networks:
+  crowd-bridge:
+    external: true

scripts/services/docker/Dockerfile.snowflake_connectors

@@ -0,0 +1,39 @@
+{
+  "name": "@crowd/snowflake-connectors",
+  "scripts": {
+    "start": "CROWD_TEMPORAL_TASKQUEUE=snowflakeConnectors SERVICE=snowflake-connectors tsx src/index.ts",
+    "start:debug": "CROWD_TEMPORAL_TASKQUEUE=snowflakeConnectors SERVICE=snowflake-connectors LOG_LEVEL=debug tsx src/index.ts",
+    "start:debug:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=snowflakeConnectors SERVICE=snowflake-connectors LOG_LEVEL=debug tsx src/index.ts",
+    "dev": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug",
+    "dev:local": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug:local",
+    "lint": "npx eslint --ext .ts src --max-warnings=0",
+    "format": "npx prettier --write \"src/**/*.ts\"",
+    "format-check": "npx prettier --check .",
+    "tsc-check": "tsc --noEmit",
+    "trigger-export": "SERVICE=snowflake-connectors tsx src/scripts/triggerExport.ts"
+  },
+  "dependencies": {
+    "@crowd/archetype-standard": "workspace:*",
+    "@crowd/common": "workspace:*",
+    "@crowd/integrations": "workspace:*",
+    "@crowd/common_services": "workspace:*",
+    "@crowd/database": "workspace:*",
+    "@crowd/logging": "workspace:*",
+    "@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"
+  },
+  "devDependencies": {
+    "nodemon": "^3.0.1"
+  }
+}

scripts/services/docker/Dockerfile.snowflake_connectors.dockerignore

@@ -0,0 +1,67 @@
+/**
+ * Export activity: Execute COPY INTO + write metadata.
+ *
+ * This activity is invoked by the exportWorkflow and performs
+ * the actual Snowflake export and metadata bookkeeping.
+ */
+import { WRITE_DB_CONFIG, getDbConnection } from '@crowd/database'
+import { getServiceChildLogger } from '@crowd/logging'
+import { PlatformType } from '@crowd/types'
+
+import { MetadataStore } from '../core/metadataStore'
+import { SnowflakeExporter } from '../core/snowflakeExporter'
+import { getEnabledPlatforms as _getEnabledPlatforms, getPlatform } from '../integrations'
+
+export async function getEnabledPlatforms(): Promise<PlatformType[]> {
+  return _getEnabledPlatforms()
+}
+
+const log = getServiceChildLogger('exportActivity')
+
+function buildS3FilenamePrefix(platform: string): string {
+  const now = new Date()
+  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
+  if (!s3BucketPath) {
+    throw new Error('Missing required env var CROWD_SNOWFLAKE_S3_BUCKET_PATH')
+  }
+  return `${s3BucketPath}/${platform}/${year}/${month}/${day}`
+}
+
+export async function executeExport(platform: PlatformType): Promise<void> {
+  log.info({ platform }, 'Starting export')
+
+  const exporter = new SnowflakeExporter()
+  const db = await getDbConnection(WRITE_DB_CONFIG())
+
+  try {
+    const metadataStore = new MetadataStore(db)
+    const platformDef = getPlatform(platform)
+
+    const lastSuccessfulExportTimestamp = await metadataStore.getLatestExportStartedAt(platform)
+    const sinceTimestamp = lastSuccessfulExportTimestamp
+      ? new Date(lastSuccessfulExportTimestamp).toISOString()
+      : undefined
+    const sourceQuery = platformDef.buildSourceQuery(sinceTimestamp)
+    const s3FilenamePrefix = buildS3FilenamePrefix(platform)
+
+    const exportStartedAt = new Date()
+
+    const onBatchComplete = async (s3Path: string, totalRows: number, totalBytes: number) => {
+      await metadataStore.insertExportJob(platform, s3Path, totalRows, totalBytes, exportStartedAt)
+    }
+
+    await exporter.executeBatchedCopyInto(sourceQuery, s3FilenamePrefix, onBatchComplete)
+
+    log.info({ platform }, 'Export completed')
+  } catch (err) {
+    log.error({ platform, err }, 'Export failed')
+    throw err
+  } finally {
+    await exporter
+      .destroy()
+      .catch((err) => log.warn({ err }, 'Failed to close Snowflake connection'))
+  }
+}