Skip to content

Commit fd9b1c8

Browse files
mbani01mbani01
committed
feat: tnc connector implementation
1 parent 216f183 commit fd9b1c8

8 files changed

Lines changed: 664 additions & 0 deletions

File tree

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,12 @@ import { PlatformType } from '@crowd/types'
88

99
import { buildSourceQuery as cventBuildSourceQuery } from './cvent/event-registrations/buildSourceQuery'
1010
import { CventTransformer } from './cvent/event-registrations/transformer'
11+
import { buildSourceQuery as tncCertificatesBuildQuery } from './tnc/certificates/buildSourceQuery'
12+
import { TncCertificatesTransformer } from './tnc/certificates/transformer'
13+
import { buildSourceQuery as tncCourseActionsBuildQuery } from './tnc/course-actions/buildSourceQuery'
14+
import { TncCourseActionsTransformer } from './tnc/course-actions/transformer'
15+
import { buildSourceQuery as tncEnrollmentsBuildQuery } from './tnc/enrollments/buildSourceQuery'
16+
import { TncEnrollmentsTransformer } from './tnc/enrollments/transformer'
1117
import { DataSource, DataSourceName, PlatformDefinition } from './types'
1218

1319
export type { BuildSourceQuery, DataSource, PlatformDefinition } from './types'
@@ -23,6 +29,25 @@ const supported: Partial<Record<PlatformType, PlatformDefinition>> = {
2329
},
2430
],
2531
},
32+
[PlatformType.TNC]: {
33+
sources: [
34+
{
35+
name: DataSourceName.TNC_ENROLLMENTS,
36+
buildSourceQuery: tncEnrollmentsBuildQuery,
37+
transformer: new TncEnrollmentsTransformer(),
38+
},
39+
{
40+
name: DataSourceName.TNC_CERTIFICATES,
41+
buildSourceQuery: tncCertificatesBuildQuery,
42+
transformer: new TncCertificatesTransformer(),
43+
},
44+
{
45+
name: DataSourceName.TNC_COURSE_ACTIONS,
46+
buildSourceQuery: tncCourseActionsBuildQuery,
47+
transformer: new TncCourseActionsTransformer(),
48+
},
49+
],
50+
},
2651
}
2752

2853
const enabled = (process.env.CROWD_SNOWFLAKE_ENABLED_PLATFORMS || '')

‎services/apps/snowflake_connectors/src/integrations/tnc/certificates/buildSourceQuery.ts‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
import { IS_PROD_ENV } from '@crowd/common'
2+
3+
const CDP_MATCHED_SEGMENTS = `
4+
cdp_matched_segments AS (
5+
SELECT DISTINCT
6+
s.SOURCE_ID AS sourceId,
7+
s.slug
8+
FROM ANALYTICS.SILVER_DIM._CROWD_DEV_SEGMENTS_UNION s
9+
WHERE s.PARENT_SLUG IS NOT NULL
10+
AND s.GRANDPARENTS_SLUG IS NOT NULL
11+
AND s.SOURCE_ID IS NOT NULL
12+
)`
13+
14+
const ORG_ACCOUNTS = `
15+
org_accounts AS (
16+
SELECT account_id, account_name, website, domain_aliases, LOGO_URL, INDUSTRY, N_EMPLOYEES
17+
FROM analytics.bronze_fivetran_salesforce.accounts
18+
WHERE website IS NOT NULL
19+
UNION ALL
20+
SELECT account_id, account_name, website, domain_aliases, NULL AS LOGO_URL, NULL AS INDUSTRY, NULL AS N_EMPLOYEES
21+
FROM analytics.bronze_fivetran_salesforce_b2b.accounts
22+
WHERE website IS NOT NULL
23+
)`
24+
25+
const LFID_COALESCE = `COALESCE(mu.user_name, u.lf_username)`
26+
27+
export const buildSourceQuery = (sinceTimestamp?: string): string => {
28+
let select = `
29+
SELECT
30+
c.*,
31+
cms.slug AS PROJECT_SLUG,
32+
org.account_name AS ORGANIZATION_NAME,
33+
org.website AS ORG_WEBSITE,
34+
org.domain_aliases AS ORG_DOMAIN_ALIASES,
35+
org.logo_url AS LOGO_URL,
36+
org.industry AS ORGANIZATION_INDUSTRY,
37+
org.n_employees AS ORGANIZATION_SIZE,
38+
${LFID_COALESCE} AS LFID
39+
FROM analytics.silver_fact.certificates c
40+
INNER JOIN cdp_matched_segments cms
41+
ON cms.sourceId = c.project_id
42+
LEFT JOIN analytics.bronze_fivetran_salesforce.bronze_salesforce_merged_user mu
43+
ON c.user_id = mu.user_id
44+
AND mu.user_name IS NOT NULL
45+
LEFT JOIN analytics.silver_dim.users u
46+
ON LOWER(c.user_email) = LOWER(u.email)
47+
AND u.lf_username IS NOT NULL
48+
LEFT JOIN org_accounts org
49+
ON c.account_id = org.account_id
50+
WHERE c.user_email IS NOT NULL
51+
AND c.is_deleted = false`
52+
53+
if (!IS_PROD_ENV) {
54+
select += ` AND cms.slug = 'pytorch'`
55+
}
56+
57+
const dedup = `
58+
QUALIFY ROW_NUMBER() OVER (PARTITION BY c.certificate_id ORDER BY org.website DESC) = 1`
59+
60+
if (!sinceTimestamp) {
61+
return `
62+
WITH ${ORG_ACCOUNTS},
63+
${CDP_MATCHED_SEGMENTS}
64+
${select}
65+
${dedup}`.trim()
66+
}
67+
68+
return `
69+
WITH ${ORG_ACCOUNTS},
70+
${CDP_MATCHED_SEGMENTS},
71+
new_cdp_segments AS (
72+
SELECT DISTINCT
73+
s.SOURCE_ID AS sourceId,
74+
s.slug
75+
FROM ANALYTICS.SILVER_DIM._CROWD_DEV_SEGMENTS_UNION s
76+
WHERE s.CREATED_TS >= '${sinceTimestamp}'
77+
AND s.PARENT_SLUG IS NOT NULL
78+
AND s.GRANDPARENTS_SLUG IS NOT NULL
79+
AND s.SOURCE_ID IS NOT NULL
80+
)
81+
82+
-- New certificates since last export
83+
${select}
84+
AND c.issued_ts >= '${sinceTimestamp}'
85+
${dedup}
86+
87+
UNION
88+
89+
-- All certificates in newly created segments
90+
${select}
91+
AND EXISTS (
92+
SELECT 1 FROM new_cdp_segments ncs
93+
WHERE ncs.slug = cms.slug AND ncs.sourceId = cms.sourceId
94+
)
95+
${dedup}`.trim()
96+
}

‎services/apps/snowflake_connectors/src/integrations/tnc/certificates/transformer.ts‎

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
import { TNC_GRID, TncActivityType } from '@crowd/integrations'
2+
import { getServiceChildLogger } from '@crowd/logging'
3+
import {
4+
IActivityData,
5+
IMemberData,
6+
IOrganizationIdentity,
7+
MemberAttributeName,
8+
MemberIdentityType,
9+
OrganizationIdentityType,
10+
OrganizationSource,
11+
PlatformType,
12+
} from '@crowd/types'
13+
14+
import { TransformedActivity, TransformerBase } from '../../../core/transformerBase'
15+
16+
const log = getServiceChildLogger('tncCertificatesTransformer')
17+
18+
export class TncCertificatesTransformer extends TransformerBase {
19+
readonly platform = PlatformType.TNC
20+
21+
transformRow(row: Record<string, unknown>): TransformedActivity | null {
22+
const email = (row.USER_EMAIL as string | null)?.trim() || null
23+
if (!email) {
24+
log.debug({ certificateId: row.CERTIFICATE_ID }, 'Skipping row: missing email')
25+
return null
26+
}
27+
28+
const certificateId = (row.CERTIFICATE_ID as string)?.trim()
29+
const learnerName = (row.LEARNER_NAME as string | null)?.trim() || null
30+
const lfUsername = (row.LFID as string | null)?.trim() || null
31+
32+
const identities: IMemberData['identities'] = []
33+
const sourceId = (row.USER_ID as string | null) || undefined
34+
35+
if (lfUsername) {
36+
identities.push(
37+
{
38+
platform: PlatformType.TNC,
39+
value: email,
40+
type: MemberIdentityType.EMAIL,
41+
verified: true,
42+
sourceId,
43+
},
44+
{
45+
platform: PlatformType.TNC,
46+
value: lfUsername,
47+
type: MemberIdentityType.USERNAME,
48+
verified: true,
49+
sourceId,
50+
},
51+
{
52+
platform: PlatformType.LFID,
53+
value: lfUsername,
54+
type: MemberIdentityType.USERNAME,
55+
verified: true,
56+
sourceId,
57+
},
58+
)
59+
} else {
60+
identities.push({
61+
platform: PlatformType.TNC,
62+
value: email,
63+
type: MemberIdentityType.USERNAME,
64+
verified: true,
65+
sourceId,
66+
})
67+
}
68+
69+
const activity: IActivityData = {
70+
type: TncActivityType.ISSUED_CERTIFICATION,
71+
platform: PlatformType.TNC,
72+
timestamp: (row.ISSUED_TS as string | null) || null,
73+
score: TNC_GRID[TncActivityType.ISSUED_CERTIFICATION].score,
74+
sourceId: certificateId,
75+
sourceParentId: (row.COURSE_ID as string | null) || undefined,
76+
member: {
77+
displayName: learnerName || email,
78+
identities,
79+
organizations: this.buildOrganizations(row),
80+
attributes: {
81+
...((row.JOB_TITLE as string | null) && {
82+
[MemberAttributeName.JOB_TITLE]: { [PlatformType.TNC]: row.JOB_TITLE as string },
83+
}),
84+
...((row.USER_COUNTRY as string | null) && {
85+
[MemberAttributeName.COUNTRY]: { [PlatformType.TNC]: row.USER_COUNTRY as string },
86+
}),
87+
},
88+
},
89+
attributes: {
90+
productName: (row.COURSE_NAME as string | null) || null,
91+
productType: 'Certification',
92+
technology: (row.TECHNOLOGIES_LIST as string | null) || null,
93+
didExpire: row.DID_EXPIRE as boolean | null,
94+
expirationDate: (row.EXPIRATION_DATE as string | null) || null,
95+
},
96+
}
97+
98+
const segmentSlug = (row.PROJECT_SLUG as string | null)?.trim() || null
99+
const segmentSourceId = (row.PROJECT_ID as string | null)?.trim() || null
100+
if (!segmentSlug || !segmentSourceId) {
101+
return null
102+
}
103+
104+
return { activity, segment: { slug: segmentSlug, sourceId: segmentSourceId } }
105+
}
106+
107+
private buildOrganizations(
108+
row: Record<string, unknown>,
109+
): IActivityData['member']['organizations'] {
110+
const website = (row.ORG_WEBSITE as string | null)?.trim() || null
111+
const domainAliases = (row.ORG_DOMAIN_ALIASES as string | null)?.trim() || null
112+
113+
if (!website && !domainAliases) {
114+
return undefined
115+
}
116+
117+
const identities: IOrganizationIdentity[] = []
118+
119+
if (website) {
120+
identities.push({
121+
platform: PlatformType.TNC,
122+
value: website,
123+
type: OrganizationIdentityType.PRIMARY_DOMAIN,
124+
verified: true,
125+
})
126+
}
127+
128+
if (domainAliases) {
129+
for (const alias of domainAliases.split(',')) {
130+
const trimmed = alias.trim()
131+
if (trimmed) {
132+
identities.push({
133+
platform: PlatformType.TNC,
134+
value: trimmed,
135+
type: OrganizationIdentityType.ALTERNATIVE_DOMAIN,
136+
verified: true,
137+
})
138+
}
139+
}
140+
}
141+
142+
return [
143+
{
144+
displayName: (row.ORGANIZATION_NAME as string | null)?.trim() || website,
145+
source: OrganizationSource.TNC,
146+
identities,
147+
logo: (row.LOGO_URL as string | null)?.trim() || undefined,
148+
size: (row.ORGANIZATION_SIZE as string | null)?.trim() || undefined,
149+
industry: (row.ORGANIZATION_INDUSTRY as string | null)?.trim() || undefined,
150+
},
151+
]
152+
}
153+
}

‎services/apps/snowflake_connectors/src/integrations/tnc/course-actions/buildSourceQuery.ts‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
import { IS_PROD_ENV } from '@crowd/common'
2+
3+
// TODO: user resolution — course_actions only has INTERNAL_TI_USER_ID.
4+
// Need to identify the join table that maps INTERNAL_TI_USER_ID to user_email, learner_name, account_id, etc.
5+
6+
const CDP_MATCHED_SEGMENTS = `
7+
cdp_matched_segments AS (
8+
SELECT DISTINCT
9+
s.SOURCE_ID AS sourceId,
10+
s.slug
11+
FROM ANALYTICS.SILVER_DIM._CROWD_DEV_SEGMENTS_UNION s
12+
WHERE s.PARENT_SLUG IS NOT NULL
13+
AND s.GRANDPARENTS_SLUG IS NOT NULL
14+
AND s.SOURCE_ID IS NOT NULL
15+
)`
16+
17+
export const buildSourceQuery = (sinceTimestamp?: string): string => {
18+
let select = `
19+
SELECT
20+
ca.course_action_id,
21+
ca.course_id,
22+
ca.timestamp,
23+
ca.type,
24+
ca.source,
25+
ca.internal_ti_user_id,
26+
co.title AS COURSE_NAME,
27+
co.course_group_id,
28+
co.slug AS COURSE_SLUG,
29+
co.instruction_type,
30+
co.product_type,
31+
co.is_training,
32+
co.is_certification
33+
FROM analytics.bronze_census_ti.course_actions ca
34+
INNER JOIN analytics.bronze_census_ti.courses co
35+
ON ca.course_id = co.course_id
36+
WHERE ca.type = 'status_change'
37+
AND ca.source = 'course_started'
38+
AND co.is_test_or_archived = false`
39+
40+
if (!IS_PROD_ENV) {
41+
select += ` AND 1=1` // TODO: add non-prod project filter once segment join is available
42+
}
43+
44+
if (!sinceTimestamp) {
45+
return select.trim()
46+
}
47+
48+
return `${select}
49+
AND ca.timestamp >= '${sinceTimestamp}'`.trim()
50+
}

0 commit comments

Comments
 (0)