feat: committees implementation [CM-1066]

backend/src/database/migrations/V1775064222__addCommitteesActivityTypes.sql

@@ -0,0 +1,3 @@
+INSERT INTO "activityTypes" ("activityType", platform, "isCodeContribution", "isCollaboration", description, "label") VALUES
+('added-to-committee', 'committees', false, false, 'Member is added to a committee', 'Added to committee'),
+('removed-from-committee', 'committees', false, false, 'Member is removed from a committee', 'Removed from committee');

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

@@ -0,0 +1,116 @@
+import { IS_PROD_ENV } from '@crowd/common'
+
+// Main: FIVETRAN_INGEST.SFDC_CONNECTOR_PROD_PLATFORM.COMMUNITY__C
+// Joins:
+// - ANALYTICS.SILVER_DIM.COMMITTEE (committee metadata + project slug)
+// - ANALYTICS.BRONZE_KAFKA_CROWD_DEV.SEGMENTS (segment resolution)
+// - ANALYTICS.SILVER_DIM.USERS (member identity: email, lf_username, name)
+// - ANALYTICS.BRONZE_FIVETRAN_SALESFORCE_B2B.USERS (actor name + email)
+// - ANALYTICS.BRONZE_FIVETRAN_SALESFORCE_B2B.ACCOUNTS (org data)
+
+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
+    FROM ANALYTICS.BRONZE_FIVETRAN_SALESFORCE_B2B.ACCOUNTS
+    WHERE website IS NOT NULL
+  )`
+
+export const buildSourceQuery = (sinceTimestamp?: string): string => {
+  let select = `
+  SELECT
+    c.SFID,
+    c._FIVETRAN_DELETED AS FIVETRAN_DELETED,
+    c.CONTACTEMAIL__C,
+    c.CREATEDBYID,
+    c.COLLABORATION_NAME__C,
+    c.ACCOUNT__C,
+    c.ROLE__C,
+    c.CREATEDDATE::TIMESTAMP_NTZ AS CREATEDDATE,
+    c.LASTMODIFIEDDATE::TIMESTAMP_NTZ AS LASTMODIFIEDDATE,
+    cm.COMMITTEE_ID,
+    cm.COMMITTEE_NAME,
+    cm.PROJECT_ID,
+    cm.PROJECT_NAME,
+    cm.PROJECT_SLUG,
+    su.EMAIL AS SU_EMAIL,
+    su.LF_USERNAME,
+    su.PRIMARY_SOURCE_USER_ID,
+    su.FIRST_NAME AS SU_FIRST_NAME,
+    su.LAST_NAME AS SU_LAST_NAME,
+    su.FULL_NAME AS SU_FULL_NAME,
+    bu.FIRST_NAME AS BU_FIRST_NAME,
+    bu.LAST_NAME AS BU_LAST_NAME,
+    bu.EMAIL AS BU_EMAIL,
+    org.account_name AS ACCOUNT_NAME,
+    org.website AS ORG_WEBSITE,
+    org.domain_aliases AS ORG_DOMAIN_ALIASES
+  FROM FIVETRAN_INGEST.SFDC_CONNECTOR_PROD_PLATFORM.COMMUNITY__C c
+  JOIN ANALYTICS.SILVER_DIM.COMMITTEE cm
+    ON c.COLLABORATION_NAME__C = cm.COMMITTEE_ID
+  INNER JOIN cdp_matched_segments cms
+    ON cms.slug = cm.PROJECT_SLUG
+    AND cms.sourceId = cm.PROJECT_ID
+  LEFT JOIN ANALYTICS.SILVER_DIM.USERS su
+    ON LOWER(c.CONTACTEMAIL__C) = LOWER(su.EMAIL)
+  LEFT JOIN ANALYTICS.BRONZE_FIVETRAN_SALESFORCE_B2B.USERS bu
+    ON c.CREATEDBYID = bu.USER_ID
+  LEFT JOIN org_accounts org
+    ON c.ACCOUNT__C = org.account_id
+  WHERE c.LASTMODIFIEDDATE IS NOT NULL`
+
+  // Limit to a single project in non-prod to avoid exporting all project data
+  if (!IS_PROD_ENV) {
+    select += ` AND cm.PROJECT_SLUG = 'ccc'`
+  }
+
+  const dedup = `
+  QUALIFY ROW_NUMBER() OVER (PARTITION BY c.SFID ORDER BY org.website DESC) = 1`
+
+  if (!sinceTimestamp) {
+    return `
+  WITH ${ORG_ACCOUNTS},
+  ${CDP_MATCHED_SEGMENTS}
+  ${select}
+  ${dedup}`.trim()
+  }
+
+  return `
+  WITH ${ORG_ACCOUNTS},
+  ${CDP_MATCHED_SEGMENTS},
+  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 committee memberships since last export
+  ${select}
+    AND c.LASTMODIFIEDDATE > '${sinceTimestamp}'
+  ${dedup}
+
+  UNION
+
+  -- All committee memberships 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()
+}

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

@@ -0,0 +1,196 @@
+import { COMMITTEES_GRID, CommitteesActivityType } from '@crowd/integrations'
+import { getServiceChildLogger } from '@crowd/logging'
+import {
+  IActivityData,
+  IMemberData,
+  IOrganizationIdentity,
+  MemberIdentityType,
+  OrganizationIdentityType,
+  OrganizationSource,
+  PlatformType,
+} from '@crowd/types'
+
+import { TransformedActivity, TransformerBase } from '../../../core/transformerBase'
+
+const log = getServiceChildLogger('committeesCommitteesTransformer')
+
+export class CommitteesCommitteesTransformer extends TransformerBase {
+  readonly platform = PlatformType.COMMITTEES
+
+  transformRow(row: Record<string, unknown>): TransformedActivity | null {
+    const email = (row.CONTACTEMAIL__C as string | null)?.trim() || null
+    if (!email) {
+      log.warn(
+        { sfid: row.SFID, committeeId: row.COMMITTEE_ID, rawEmail: row.CONTACTEMAIL__C },
+        'Skipping row: missing email',
+      )
+      return null
+    }
+
+    const committeeId = (row.COMMITTEE_ID as string).trim()
+    const fivetranDeleted = row.FIVETRAN_DELETED as boolean
+    const lfUsername = (row.LF_USERNAME as string | null)?.trim() || null
+    const suFullName = (row.SU_FULL_NAME as string | null)?.trim() || null
+    const suFirstName = (row.SU_FIRST_NAME as string | null)?.trim() || null
+    const suLastName = (row.SU_LAST_NAME as string | null)?.trim() || null
+
+    const displayName =
+      suFullName ||
+      (suFirstName && suLastName ? `${suFirstName} ${suLastName}` : suFirstName || suLastName) ||
+      email.split('@')[0]
+
+    const type = fivetranDeleted
+      ? CommitteesActivityType.REMOVED_FROM_COMMITTEE
+      : CommitteesActivityType.ADDED_TO_COMMITTEE
+
+    const identities: IMemberData['identities'] = []
+
+    if (lfUsername) {
+      identities.push(
+        {
+          platform: PlatformType.COMMITTEES,
+          value: email,
+          type: MemberIdentityType.EMAIL,
+          verified: true,
+          verifiedBy: PlatformType.COMMITTEES,
+        },
+        {
+          platform: PlatformType.COMMITTEES,
+          value: lfUsername,
+          type: MemberIdentityType.USERNAME,
+          verified: true,
+          verifiedBy: PlatformType.COMMITTEES,
+        },
+        {
+          platform: PlatformType.LFID,
+          value: lfUsername,
+          type: MemberIdentityType.USERNAME,
+          verified: true,
+          verifiedBy: PlatformType.COMMITTEES,
+        },
+      )
+    }
+
+    if (!lfUsername) {
+      identities.push({
+        platform: PlatformType.COMMITTEES,
+        value: email,
+        type: MemberIdentityType.USERNAME,
+        verified: true,
+        verifiedBy: PlatformType.COMMITTEES,
+      })
+    }
+
+    const activity: IActivityData = {
+      type,
+      platform: PlatformType.COMMITTEES,
+      timestamp: (row.LASTMODIFIEDDATE as string | null) || null,
+      score: COMMITTEES_GRID[type].score,
+      sourceId: committeeId,
+      sourceParentId: null,
+      member: {
+        displayName,
+        identities,
+        organizations: this.buildOrganizations(row),
+      },
+      attributes: {
+        committeeId: (row.COLLABORATION_NAME__C as string | null) || null,
+        committeeName: (row.COMMITTEE_NAME as string | null) || null,
+        role: (row.ROLE__C as string | null) || null,
+        projectId: (row.PROJECT_ID as string | null) || null,
+        projectName: (row.PROJECT_NAME as string | null) || null,
+        organizationId: (row.ACCOUNT__C as string | null) || null,
+        organizationName: (row.ACCOUNT_NAME as string | null) || null,
+        member: {
+          userId: (row.PRIMARY_SOURCE_USER_ID as string | null) || null,
+          firstName: (row.SU_FIRST_NAME as string | null) || null,
+          lastName: (row.SU_LAST_NAME as string | null) || null,
+          email,
+        },
+        actor: {
+          userId: (row.CREATEDBYID as string | null) || null,
+          firstName: (row.BU_FIRST_NAME as string | null) || null,
+          lastName: (row.BU_LAST_NAME as string | null) || null,
+          email: (row.BU_EMAIL as string | null) || null,
+        },
+        activityDate: (row.CREATEDDATE as string | null) || null,
+      },
+    }
+
+    const segmentSlug = (row.PROJECT_SLUG as string | null)?.trim() || null
+    const segmentSourceId = (row.PROJECT_ID as string | null)?.trim() || null
+
+    if (!segmentSlug || !segmentSourceId) {
+      log.warn(
+        { sfid: row.SFID, committeeId, segmentSlug, segmentSourceId },
+        'Skipping row: missing segment slug or sourceId',
+      )
+      return null
+    }
+
+    return { activity, segment: { slug: segmentSlug, sourceId: segmentSourceId } }
+  }
+
+  private buildOrganizations(row: Record<string, unknown>): IActivityData['member']['organizations'] {
+    const website = (row.ORG_WEBSITE as string | null)?.trim() || null
+    const domainAliases = (row.ORG_DOMAIN_ALIASES as string | null)?.trim() || null
+
+    if (!website && !domainAliases) {
+      return undefined
+    }
+
+    const displayName = (row.ACCOUNT_NAME as string | null)?.trim() || website
+
+    if (this.isIndividualNoAccount(displayName)) {
+      return [
+        {
+          displayName,
+          source: OrganizationSource.COMMITTEES,
+          identities: website
+            ? [
+                {
+                  platform: PlatformType.COMMITTEES,
+                  value: website,
+                  type: OrganizationIdentityType.PRIMARY_DOMAIN,
+                  verified: true,
+                },
+              ]
+            : [],
+        },
+      ]
+    }
+
+    const identities: IOrganizationIdentity[] = []
+
+    if (website) {
+      identities.push({
+        platform: PlatformType.COMMITTEES,
+        value: website,
+        type: OrganizationIdentityType.PRIMARY_DOMAIN,
+        verified: true,
+      })
+    }
+
+    if (domainAliases) {
+      for (const alias of domainAliases.split(',')) {
+        const trimmed = alias.trim()
+        if (trimmed) {
+          identities.push({
+            platform: PlatformType.COMMITTEES,
+            value: trimmed,
+            type: OrganizationIdentityType.ALTERNATIVE_DOMAIN,
+            verified: true,
+          })
+        }
+      }
+    }
+
+    return [
+      {
+        displayName,
+        source: OrganizationSource.COMMITTEES,
+        identities,
+      },
+    ]
+  }
+}

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

@@ -6,6 +6,8 @@
  */
 import { PlatformType } from '@crowd/types'
 
+import { buildSourceQuery as committeesCommitteesBuildQuery } from './committees/committees/buildSourceQuery'
+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 tncCertificatesBuildQuery } from './tnc/certificates/buildSourceQuery'
@@ -20,6 +22,15 @@ export type { BuildSourceQuery, DataSource, PlatformDefinition } from './types'
 export { DataSourceName } from './types'
 
 const supported: Partial<Record<PlatformType, PlatformDefinition>> = {
+  [PlatformType.COMMITTEES]: {
+    sources: [
+      {
+        name: DataSourceName.COMMITTEES_COMMITTEES,
+        buildSourceQuery: committeesCommitteesBuildQuery,
+        transformer: new CommitteesCommitteesTransformer(),
+      },
+    ],
+  },
   [PlatformType.CVENT]: {
     sources: [
       {

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

@@ -8,6 +8,7 @@ export enum DataSourceName {
   TNC_ENROLLMENTS = 'enrollments',
   TNC_CERTIFICATES = 'certificates',
   TNC_COURSES = 'courses',
+  COMMITTEES_COMMITTEES = 'committees',
 }
 
 export interface DataSource {

services/libs/data-access-layer/src/organizations/attributesConfig.ts

@@ -233,6 +233,7 @@ export const ORG_DB_ATTRIBUTE_SOURCE_PRIORITY = [
   OrganizationAttributeSource.ENRICHMENT_PEOPLEDATALABS,
   OrganizationAttributeSource.CVENT,
   OrganizationAttributeSource.TNC,
+  OrganizationAttributeSource.COMMITTEES,
   // legacy - keeping this for backward compatibility
   OrganizationAttributeSource.ENRICHMENT,
   OrganizationAttributeSource.GITHUB,

services/libs/integrations/src/integrations/committees/types.ts

@@ -0,0 +1,11 @@
+import { IActivityScoringGrid } from '@crowd/types'
+
+export enum CommitteesActivityType {
+  ADDED_TO_COMMITTEE = 'added-to-committee',
+  REMOVED_FROM_COMMITTEE = 'removed-from-committee',
+}
+
+export const COMMITTEES_GRID: Record<CommitteesActivityType, IActivityScoringGrid> = {
+  [CommitteesActivityType.ADDED_TO_COMMITTEE]: { score: 1 },
+  [CommitteesActivityType.REMOVED_FROM_COMMITTEE]: { score: 1 },
+}

services/libs/integrations/src/integrations/index.ts

@@ -52,4 +52,6 @@ export * from './cvent/types'
 
 export * from './tnc/types'
 
+export * from './committees/types'
+
 export * from './activityDisplayService'

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

@@ -14,6 +14,7 @@ export enum OrganizationSource {
   UI = 'ui',
   CVENT = 'cvent',
   TNC = 'tnc',
+  COMMITTEES = 'committees',
 }
 
 export enum OrganizationMergeSuggestionType {
@@ -42,6 +43,7 @@ export enum OrganizationAttributeSource {
   ENRICHMENT_PEOPLEDATALABS = 'enrichment-peopledatalabs',
   CVENT = 'cvent',
   TNC = 'tnc',
+  COMMITTEES = 'committees',
   // legacy - keeping this for backward compatibility
   ENRICHMENT = 'enrichment',
   GITHUB = 'github',