feat(snowflake-connectors): meetings implementation [CM-1034]#3998
feat(snowflake-connectors): meetings implementation [CM-1034]#3998
Conversation
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
|
|
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Pull request overview
Adds “meetings” platform support to the Snowflake connectors pipeline, including new activity types for meeting invitations and attendance, plus the necessary enum/config updates so meetings is recognized across the codebase.
Changes:
- Introduces
MEETINGSas a supported platform/org source and wires it into source priority. - Adds meetings activity types (
invited-meeting,attended-meeting) with scoring grid + DB registration. - Extends the transformer pipeline to allow a single row to produce multiple activities and updates the consumer to emit them all.
Reviewed changes
Copilot reviewed 13 out of 14 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| services/libs/types/src/enums/platforms.ts | Adds MEETINGS to PlatformType. |
| services/libs/types/src/enums/organizations.ts | Adds MEETINGS to organization source enums. |
| services/libs/integrations/src/integrations/meetings/types.ts | Defines meetings activity types + scoring grid. |
| services/libs/integrations/src/integrations/index.ts | Exports meetings integration types. |
| services/libs/data-access-layer/src/organizations/attributesConfig.ts | Adds meetings to org attribute source priority list. |
| services/apps/snowflake_connectors/src/integrations/types.ts | Registers meetings meeting-attendance as a data source name. |
| services/apps/snowflake_connectors/src/integrations/meetings/meeting-attendance/transformer.ts | Implements meetings attendance transformer producing invited/attended activities. |
| services/apps/snowflake_connectors/src/integrations/meetings/meeting-attendance/buildSourceQuery.ts | Adds Snowflake query for exporting meeting attendance rows + incremental logic. |
| services/apps/snowflake_connectors/src/integrations/index.ts | Registers meetings platform and its source in the connectors registry. |
| services/apps/snowflake_connectors/src/core/transformerBase.ts | Updates transformer API to allow multiple activities per row; normalizes safeTransformRow to arrays. |
| services/apps/snowflake_connectors/src/consumer/transformerConsumer.ts | Updates consumer to handle multiple transformed activities per input row. |
| backend/src/database/migrations/V1775219382__addMeetingsActivityTypes.sql | Inserts new meetings activity types into activityTypes. |
| backend/src/database/migrations/U1775219382__addMeetingsActivityTypes.sql | (Present but empty) rollback migration file for the above insert. |
| .claude/skills/scaffold-snowflake-connector/SKILL.md | Updates internal scaffolding guidance related to org mapping discovery. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
|
|
||
| -- Updated records in existing segments | ||
| ${select} | ||
| AND t.MEETING_DATE > '${sinceTimestamp}'::DATE |
There was a problem hiding this comment.
The incremental export filter casts sinceTimestamp (an ISO datetime) to DATE and compares against t.MEETING_DATE. This drops the time component and can miss same-day records when exports run more than once per day. Consider comparing against a true timestamp (e.g., combine MEETING_DATE + MEETING_TIME into a TIMESTAMP) or use an updated/created timestamp column if available, so the sinceTimestamp semantics stay consistent with the other connectors.
| AND t.MEETING_DATE > '${sinceTimestamp}'::DATE | |
| AND TO_TIMESTAMP_NTZ( | |
| TO_VARCHAR(t.MEETING_DATE) || ' ' || COALESCE(TO_VARCHAR(t.MEETING_TIME), '00:00:00') | |
| ) > TO_TIMESTAMP_NTZ('${sinceTimestamp}') |
| private buildOrganizations( | ||
| row: Record<string, unknown>, | ||
| ): IActivityData['member']['organizations'] { | ||
| const accountName = (row.ACCOUNT_NAME as string | null)?.trim() || null | ||
| if (!accountName) { | ||
| return undefined | ||
| } | ||
|
|
There was a problem hiding this comment.
buildOrganizations drops org info if ACCOUNT_NAME is missing, even when ORG_WEBSITE (and/or aliases) are present from the joined org_accounts CTE. Other Snowflake connectors (e.g. CventTransformer.buildOrganizations) fall back to the website for displayName and still emit org identities. Consider matching that pattern here to avoid silently losing org attribution.
| const attributes = { | ||
| meetingID: row.MEETING_ID, | ||
| scheduledTime: timestamp, | ||
| topic: (row.MEETING_NAME as string | null) || null, | ||
| projectID: (row.PROJECT_ID as string | null) || null, | ||
| projectName: (row.PROJECT_NAME as string | null) || null, | ||
| organizationId: (row.ACCOUNT_ID as string | null) || null, | ||
| organizationName: (row.ACCOUNT_NAME as string | null) || null, | ||
| meetingType: (row.RAW_COMMITTEE_TYPE as string | null) || null, |
There was a problem hiding this comment.
In attributes, key casing is inconsistent (meetingID/projectID vs organizationId). To keep attribute keys predictable for downstream consumers, consider using a consistent style (e.g., meetingId, projectId).
| 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++ | ||
| } |
There was a problem hiding this comment.
Now that transformers may return multiple activities per row, this loop resolves the same segment repeatedly when multiple activities share it (common when a row yields multiple activity types). Consider de-duplicating integrationResolver.resolve(...) calls per row (e.g., cache by slug+sourceId) to avoid extra Redis/DB lookups and reduce latency.
| 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'); |
There was a problem hiding this comment.
This migration introduces new activityTypes rows but there is no rollback SQL (the corresponding U1775219382__addMeetingsActivityTypes.sql is empty). If rollbacks are supported in this repo, add a down migration that deletes these inserted activity types (scoped to platform/type) to keep migrations reversible.
| const email = (row.INVITEE_EMAIL as string | null)?.trim() || null | ||
| if (!email) { | ||
| log.debug({ primaryKey: row.PRIMARY_KEY }, 'Skipping row: missing email') | ||
| return null | ||
| } |
There was a problem hiding this comment.
FYI @joanagmaia, many records lack the email. I attempted to retrieve it via a join, but it’s not possible as we only have full names, lf_id/sso are also null. So we're skipping those
| platform: PlatformType.MEETINGS, | ||
| timestamp, | ||
| score: MEETINGS_GRID[MeetingsActivityType.INVITED_MEETING].score, | ||
| sourceId: `${primaryKey}_invited`, |
There was a problem hiding this comment.
@joanagmaia, this one is different from specs, as meeting_id isn't unique (similar to committees). Used primaryKey as it seems to be unique
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
...s/apps/snowflake_connectors/src/integrations/meetings/meeting-attendance/buildSourceQuery.ts
Outdated
Show resolved
Hide resolved
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
| : null | ||
| if (!date || isNaN(date.getTime())) return null | ||
| if (typeof rawTime === 'number') { | ||
| date.setTime(date.getTime() + rawTime) |
There was a problem hiding this comment.
Parquet TIME value unit assumption may produce wrong timestamps
High Severity
toISOTimestamp adds rawTime directly to the date via date.setTime(date.getTime() + rawTime), assuming the Parquet TIME column value is in milliseconds. Snowflake typically exports TIME columns to Parquet using the TIME_MICROS logical type (microseconds since midnight). If @dsnp/parquetjs returns the raw microsecond value as a number, every meeting timestamp would be off by a factor of 1000 — e.g., a 10 AM meeting would appear ~416 days in the future. This is the first transformer in the codebase to handle a separate TIME column, so there's no prior precedent to validate the assumption.
services/apps/snowflake_connectors/src/integrations/meetings/meeting-attendance/transformer.ts
Show resolved
Hide resolved
Signed-off-by: Mouad BANI <mouad-mb@outlook.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
There are 2 total unresolved issues (including 1 from previous review).
Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
|
|
||
| -- Updated records in existing segments | ||
| ${select} | ||
| AND t.MEETING_DATE >= '${sinceTimestamp}'::DATE |
There was a problem hiding this comment.
Incremental query uses business date, may miss records
Medium Severity
The incremental filter uses MEETING_DATE (the scheduled meeting date) instead of a record-level creation/update timestamp. Other connectors (CVENT uses updated_ts, TNC uses enrollment_ts) filter on when the record was created or modified. Using the business date means any meeting records added or modified after the last export but whose MEETING_DATE falls before sinceTimestamp will be permanently missed in incremental runs. Only a full re-export would pick them up.
This pull request adds support for processing and tracking meeting attendance activities from the "meetings" platform. It introduces new activity types for meeting invitations and attendance, integrates the meetings data source into the Snowflake connectors pipeline, and updates relevant enums and configurations to recognize meetings as a valid platform and organization source.
Meetings platform support:
MEETINGSas a newPlatformType,OrganizationSource, andOrganizationAttributeSource, and updated the source priority list to include meetings. [1] [2] [3] [4]invited-meeting,attended-meeting) and their scoring grid, and registered them in the database. [1] [2]Snowflake connectors integration:
MeetingAttendanceTransformerand its source query, and registered the source in the Snowflake connectors integration index. [1] [2] [3] [4] [5]Transformer pipeline improvements:
transformRowandsafeTransformRownow handle arrays of activities, and the consumer processes all returned activities. [1] [2] [3] [4]Integrations exports:
Note
Medium Risk
Medium risk because it extends the snowflake connector transformation contract to allow multiple activities per row (affecting all transformers/consumers) and introduces a new platform/source with new activity types and migrations.
Overview
Adds a new
MEETINGSsnowflake connector source (meeting-attendance) that queriesANALYTICS.SILVER_FACT.MEETING_ATTENDANCE(with segment matching and optional org enrichment) and transforms each row intoinvited-meetingand/orattended-meetingactivities.Updates the transformation pipeline so
transformRowcan return one or many activities:TransformerBase.safeTransformRow()now normalizes to an array andTransformerConsumeremits all resulting activities.Registers meetings across the system (new
PlatformType.MEETINGS, org source/attribute source + priority, exportsMeetingsActivityType/MEETINGS_GRID) and adds a DB migration inserting the two new meetings activity types.Written by Cursor Bugbot for commit d7f4e8c. This will update automatically on new commits. Configure here.