feat(snowflake-connectors): meetings implementation [CM-1034] (#3998)

Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mbani@contractor.linuxfoundation.org>

Author: mbani01

.claude/skills/scaffold-snowflake-connector/SKILL.md

@@ -196,6 +196,8 @@ After Step 1 and/or Step 2, build a column registry per table:
 
 **Store this as the canonical column reference. Every column name used in generated code must appear in this registry. Never assume or invent a column name.**
 
+**Flag non-VARCHAR column types** (e.g., `DATE`, `TIME`, `TIMESTAMP_TZ`, `BOOLEAN`, `NUMBER`) — these arrive as native JS types from the Parquet reader, not strings (see touch point 9 rules).
+
 For each JOIN table, check whether any existing transformer in `services/apps/snowflake_connectors/src/integrations/` queries from the same table. If yes, inherit its column mappings; if no, treat every column as unknown and derive it from sample data in the Pre-Analysis step below.
 
 ### Step 3 — Sample data
@@ -342,7 +344,10 @@ After all identity fields are confirmed, summarize how `buildMemberIdentities()`
 
 ### 3b. Organization Mapping
 
-If Pre-Analysis determined there is no org data (no org-related columns found in any table), confirm: "I don't see any organization columns in the schema. Does this source have org/company data?" — if yes, proceed; if no, skip to 3c.
+If Pre-Analysis determined there is no org data (no org-related columns found in any table): before asking the user, first read existing transformers in `services/apps/snowflake_connectors/src/integrations/` to check whether any of them join an org table using a key that also exists in the user's tables. If a match is found, prompt the user:
+> "I don't see org columns in the tables you provided, but [EXISTING_PLATFORM] sources org data from `{ORG_TABLE}` via `{join_key}` — which also appears in your table. Did you mean to include this? (Recommended)"
+
+If no existing pattern is joinable, ask: "I don't see any org columns. Does this source have org/company data?" — if yes, ask for the table; if no, skip to 3c.
 
 If Pre-Analysis identified org columns:
 
@@ -565,6 +570,7 @@ File: `services/apps/snowflake_connectors/src/integrations/{platform}/{source}/b
 **Rules (enforced — do not deviate):**
 - Use explicit column names only. Do not use `table.*` or `table.* EXCLUDE (...)` in new implementations — existing sources (TNC, CVENT) use these patterns but new sources should list columns explicitly to avoid parquet encoding/decoding issues
 - If any TIMESTAMP_TZ columns exist in the schema, exclude and re-cast them as TIMESTAMP_NTZ (see CVENT pattern)
+- Do not concatenate or transform date/time columns in SQL — keep them as separate columns and let the transformer handle type coercion (see touch point 9 rules)
 - Follow the CTE structure:
   1. `org_accounts` CTE (if org data present)
   2. `CDP_MATCHED_SEGMENTS` CTE (always)
@@ -585,6 +591,8 @@ Show the full generated file and ask for confirmation before writing.
 File: `services/apps/snowflake_connectors/src/integrations/{platform}/{source}/transformer.ts`
 
 **Rules (enforced — do not deviate):**
+
+- **Parquet type coercion — never blindly cast `row.COLUMN as string`.** Snowflake types may arrive as native JS types after Parquet decoding (e.g., `DATE` → `Date` object, `TIME` → `number` in ms, `BOOLEAN` → `boolean`). Always check the Snowflake column type from the schema registry and handle the actual JS type the Parquet reader delivers — do not assume every column is a string.
 - All string comparisons must be case-insensitive: use `.toLowerCase()` on both sides of comparison only; preserve the original value in the output
 - No broad `else` statements — every branch must have an explicit condition
 - All column names referenced in code must exactly match the schema registry — never assumed

backend/src/database/migrations/U1775219382__addMeetingsActivityTypes.sql

@@ -0,0 +1,3 @@
+INSERT INTO "activityTypes" ("activityType", platform, "isCodeContribution", "isCollaboration", description, "label") VALUES
+('invited-meeting', 'meetings', false, false, 'User is invited to a meeting', 'Invited to a meeting'),
+('attended-meeting', 'meetings', false, false, 'User attends a meeting', 'Attended a meeting');

backend/src/database/migrations/V1775219382__addMeetingsActivityTypes.sql

@@ -89,24 +89,26 @@ export class TransformerConsumer {
       let resolveSkippedCount = 0
 
       for await (const row of this.s3Service.streamParquetRows(job.s3Path)) {
-        const result = source.transformer.safeTransformRow(row)
-        if (!result) {
+        const results = source.transformer.safeTransformRow(row)
+        if (!results) {
           transformSkippedCount++
           continue
         }
 
-        const resolved = await this.integrationResolver.resolve(platform, result.segment)
-        if (!resolved) {
-          resolveSkippedCount++
-          continue
+        for (const result of results) {
+          const resolved = await this.integrationResolver.resolve(platform, result.segment)
+          if (!resolved) {
+            resolveSkippedCount++
+            continue
+          }
+
+          await this.emitter.createAndProcessActivityResult(
+            resolved.segmentId,
+            resolved.integrationId,
+            result.activity,
+          )
+          transformedCount++
         }
-
-        await this.emitter.createAndProcessActivityResult(
-          resolved.segmentId,
-          resolved.integrationId,
-          result.activity,
-        )
-        transformedCount++
       }
 
       const skippedCount = transformSkippedCount + resolveSkippedCount

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

@@ -24,10 +24,12 @@ export abstract class TransformerBase {
   abstract readonly platform: PlatformType
 
   /**
-   * Transform a single raw row from the S3 export into an activity
+   * Transform a single raw row from the S3 export into one or more activities
    * along with routing metadata. Returns null if the row should be skipped.
    */
-  abstract transformRow(row: Record<string, unknown>): TransformedActivity | null
+  abstract transformRow(
+    row: Record<string, unknown>,
+  ): TransformedActivity | TransformedActivity[] | null
 
   private static readonly INDIVIDUAL_NO_ACCOUNT_RE = /^individual\s*(?:[-–?]|with)\s*no\s+account$/i
 
@@ -104,10 +106,16 @@ export abstract class TransformerBase {
 
   /**
    * Safe wrapper around transformRow that catches errors and returns null.
+   * Always normalizes the result to an array for consistent consumption.
    */
-  safeTransformRow(row: Record<string, unknown>): TransformedActivity | null {
+  safeTransformRow(row: Record<string, unknown>): TransformedActivity[] | null {
     try {
-      return this.transformRow(row)
+      const result = this.transformRow(row)
+      if (result === null) {
+        return null
+      }
+      const arr = Array.isArray(result) ? result : [result]
+      return arr.length > 0 ? arr : null
     } catch (err) {
       const message = err instanceof Error ? err.message : String(err)
       const stack = err instanceof Error ? err.stack : undefined

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

@@ -111,16 +111,14 @@ export class CommitteesCommitteesTransformer extends TransformerBase {
         {
           displayName,
           source: OrganizationSource.COMMITTEES,
-          identities: website
-            ? [
-                {
-                  platform: PlatformType.COMMITTEES,
-                  value: website,
-                  type: OrganizationIdentityType.PRIMARY_DOMAIN,
-                  verified: true,
-                },
-              ]
-            : [],
+          identities: [
+            {
+              platform: PlatformType.COMMITTEES,
+              value: website,
+              type: OrganizationIdentityType.PRIMARY_DOMAIN,
+              verified: true,
+            },
+          ],
         },
       ]
     }

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

@@ -10,6 +10,8 @@ import { buildSourceQuery as committeesCommitteesBuildQuery } from './committees
 import { CommitteesCommitteesTransformer } from './committees/committees/transformer'
 import { buildSourceQuery as cventBuildSourceQuery } from './cvent/event-registrations/buildSourceQuery'
 import { CventTransformer } from './cvent/event-registrations/transformer'
+import { buildSourceQuery as meetingAttendanceBuildQuery } from './meetings/meeting-attendance/buildSourceQuery'
+import { MeetingAttendanceTransformer } from './meetings/meeting-attendance/transformer'
 import { buildSourceQuery as tncCertificatesBuildQuery } from './tnc/certificates/buildSourceQuery'
 import { TncCertificatesTransformer } from './tnc/certificates/transformer'
 import { buildSourceQuery as tncCoursesBuildQuery } from './tnc/courses/buildSourceQuery'
@@ -22,6 +24,15 @@ export type { BuildSourceQuery, DataSource, PlatformDefinition } from './types'
 export { DataSourceName } from './types'
 
 const supported: Partial<Record<PlatformType, PlatformDefinition>> = {
+  [PlatformType.MEETINGS]: {
+    sources: [
+      {
+        name: DataSourceName.MEETINGS_MEETING_ATTENDANCE,
+        buildSourceQuery: meetingAttendanceBuildQuery,
+        transformer: new MeetingAttendanceTransformer(),
+      },
+    ],
+  },
   [PlatformType.COMMITTEES]: {
     sources: [
       {

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

@@ -0,0 +1,116 @@
+import { IS_PROD_ENV } from '@crowd/common'
+
+const CDP_MATCHED_SEGMENTS = `
+  cdp_matched_segments AS (
+    SELECT DISTINCT
+      s.SOURCE_ID AS sourceId,
+      s.slug
+    FROM ANALYTICS.BRONZE_KAFKA_CROWD_DEV.SEGMENTS s
+    WHERE s.PARENT_SLUG IS NOT NULL
+      AND s.GRANDPARENTS_SLUG IS NOT NULL
+      AND s.SOURCE_ID IS NOT NULL
+  )`
+
+const ORG_ACCOUNTS = `
+  org_accounts AS (
+    SELECT account_id, account_name, website, domain_aliases, LOGO_URL, INDUSTRY, N_EMPLOYEES
+    FROM analytics.bronze_fivetran_salesforce.accounts
+    WHERE website IS NOT NULL
+    UNION ALL
+    SELECT account_id, account_name, website, domain_aliases, NULL AS LOGO_URL, NULL AS INDUSTRY, NULL AS N_EMPLOYEES
+    FROM analytics.bronze_fivetran_salesforce_b2b.accounts
+    WHERE website IS NOT NULL
+  )`
+
+const LF_SSO_LOOKUP = `
+  lf_sso_lookup AS (
+    SELECT INVITEE_LF_USER_ID AS LF_USER_ID, MAX(INVITEE_LF_SSO) AS LF_SSO
+    FROM ANALYTICS.SILVER_FACT.MEETING_ATTENDANCE
+    WHERE INVITEE_LF_SSO IS NOT NULL
+      AND INVITEE_LF_USER_ID IS NOT NULL
+    GROUP BY INVITEE_LF_USER_ID
+  )`
+
+export const buildSourceQuery = (sinceTimestamp?: string): string => {
+  let select = `
+  SELECT
+    MD5(COALESCE(CAST(t.PRIMARY_KEY AS VARCHAR), '') || '|' || COALESCE(CAST(t.COMMITTEE_ID AS VARCHAR), '')) AS GENERATED_SOURCE_ID,
+    CAST(t.MEETING_ID AS VARCHAR) AS MEETING_ID,
+    t.MEETING_NAME,
+    t.PROJECT_ID,
+    t.PROJECT_NAME,
+    t.PROJECT_SLUG,
+    t.ACCOUNT_ID,
+    t.ACCOUNT_NAME,
+    t.MEETING_DATE,
+    t.MEETING_TIME,
+    t.INVITEE_FULL_NAME,
+    COALESCE(t.INVITEE_LF_SSO, sso.LF_SSO) AS INVITEE_LF_SSO,
+    t.INVITEE_LF_USER_ID,
+    t.INVITEE_EMAIL,
+    t.INVITEE_ATTENDED,
+    t.WAS_INVITED,
+    t.RAW_COMMITTEE_TYPE,
+    t.UPDATED_TS,
+    org.website AS ORG_WEBSITE,
+    org.domain_aliases AS ORG_DOMAIN_ALIASES,
+    org.logo_url AS LOGO_URL,
+    org.industry AS ORGANIZATION_INDUSTRY,
+    CAST(org.n_employees AS VARCHAR) AS ORGANIZATION_SIZE
+  FROM ANALYTICS.SILVER_FACT.MEETING_ATTENDANCE t
+  INNER JOIN cdp_matched_segments cms
+    ON cms.slug = t.PROJECT_SLUG
+    AND cms.sourceId = t.PROJECT_ID
+  LEFT JOIN lf_sso_lookup sso
+    ON t.INVITEE_LF_USER_ID = sso.LF_USER_ID
+  LEFT JOIN org_accounts org
+    ON t.ACCOUNT_ID = org.account_id
+  WHERE (t.WAS_INVITED = TRUE OR t.INVITEE_ATTENDED = TRUE)
+    AND NULLIF(TRIM(t.INVITEE_EMAIL), '') IS NOT NULL`
+
+  if (!IS_PROD_ENV) {
+    select += ` AND t.PROJECT_SLUG = 'cncf'`
+  }
+
+  const dedup = `
+  QUALIFY ROW_NUMBER() OVER (PARTITION BY t.PRIMARY_KEY ORDER BY org.website DESC) = 1`
+
+  if (!sinceTimestamp) {
+    return `
+  WITH ${ORG_ACCOUNTS},
+  ${CDP_MATCHED_SEGMENTS},
+  ${LF_SSO_LOOKUP}
+  ${select}
+  ${dedup}`.trim()
+  }
+
+  return `
+  WITH ${ORG_ACCOUNTS},
+  ${CDP_MATCHED_SEGMENTS},
+  ${LF_SSO_LOOKUP},
+  new_cdp_segments AS (
+    SELECT DISTINCT
+      s.SOURCE_ID AS sourceId,
+      s.slug
+    FROM ANALYTICS.BRONZE_KAFKA_CROWD_DEV.SEGMENTS s
+    WHERE s.CREATED_TS >= '${sinceTimestamp}'
+      AND s.PARENT_SLUG IS NOT NULL
+      AND s.GRANDPARENTS_SLUG IS NOT NULL
+      AND s.SOURCE_ID IS NOT NULL
+  )
+
+  -- Updated records in existing segments
+  ${select}
+    AND t.UPDATED_TS > '${sinceTimestamp}'
+  ${dedup}
+
+  UNION
+
+  -- All records in newly created segments
+  ${select}
+    AND EXISTS (
+      SELECT 1 FROM new_cdp_segments ncs
+      WHERE ncs.slug = cms.slug AND ncs.sourceId = cms.sourceId
+    )
+  ${dedup}`.trim()
+}