chore(rethinkdb): NewMeeting: Phase 2

packages/server/graphql/private/mutations/checkRethinkPgEquality.ts

@@ -1,12 +1,14 @@
 import getRethink from '../../../database/rethinkDriver'
 import getFileStoreManager from '../../../fileStorage/getFileStoreManager'
-import getKysely from '../../../postgres/getKysely'
+import {selectNewMeetings} from '../../../postgres/select'
 import {checkRowCount, checkTableEq} from '../../../postgres/utils/checkEqBase'
 import {
  compareDateAlmostEqual,
+ compareRValStringAsNumber,
  compareRValUndefinedAsFalse,
  compareRValUndefinedAsNull,
  compareRValUndefinedAsNullAndTruncateRVal,
+ compareRValUndefinedAsZero,
  defaultEqFn
 } from '../../../postgres/utils/rethinkEqualityFns'
 import {MutationResolvers} from '../resolverTypes'
@@ -33,37 +35,65 @@ const checkRethinkPgEquality: MutationResolvers['checkRethinkPgEquality'] = asyn
 ) => {
  const r = await getRethink()
 
- if (tableName === 'TeamMember') {
+ if (tableName === 'NewMeeting') {
  const rowCountResult = await checkRowCount(tableName)
- const rethinkQuery = (joinedAt: Date, id: string | number) => {
+ const rethinkQuery = (updatedAt: Date, id: string | number) => {
  return r
- .table('TeamMember' as any)
- .between([joinedAt, id], [r.maxval, r.maxval], {
+ .table('NewMeeting' as any)
+ .between([updatedAt, id], [r.maxval, r.maxval], {
  index: 'updatedAtId',
  leftBound: 'open',
  rightBound: 'closed'
  })
  .orderBy({index: 'updatedAtId'}) as any
  }
  const pgQuery = async (ids: string[]) => {
- return getKysely().selectFrom('TeamMember').selectAll().where('id', 'in', ids).execute()
+ return selectNewMeetings().where('id', 'in', ids).execute()
  }
  const errors = await checkTableEq(
  rethinkQuery,
  pgQuery,
  {
  id: defaultEqFn,
- isNotRemoved: compareRValUndefinedAsFalse,
- isLead: compareRValUndefinedAsFalse,
- isSpectatingPoker: compareRValUndefinedAsFalse,
- email: defaultEqFn,
- openDrawer: compareRValUndefinedAsNull,
- picture: defaultEqFn,
- preferredName: compareRValUndefinedAsNullAndTruncateRVal(100),
- teamId: defaultEqFn,
- userId: defaultEqFn,
+ isLegacy: compareRValUndefinedAsFalse,
  createdAt: compareDateAlmostEqual,
- updatedAt: compareDateAlmostEqual
+ updatedAt: compareDateAlmostEqual,
+ createdBy: defaultEqFn,
+ endedAt: compareRValUndefinedAsNull,
+ facilitatorStageId: defaultEqFn,
+ facilitatorUserId: defaultEqFn,
+ meetingCount: compareRValUndefinedAsZero,
+ meetingNumber: compareRValUndefinedAsZero,
+ name: compareRValUndefinedAsNullAndTruncateRVal(100),
+ summarySentAt: compareRValUndefinedAsNull,
+ teamId: defaultEqFn,
+ meetingType: defaultEqFn,
+ phases: defaultEqFn,
+ showConversionModal: compareRValUndefinedAsFalse,
+ meetingSeriesId: compareRValUndefinedAsNull,
+ scheduledEndTime: compareRValUndefinedAsNull,
+ summary: compareRValUndefinedAsNullAndTruncateRVal(10000),
+ sentimentScore: compareRValUndefinedAsNull,
+ usedReactjis: compareRValUndefinedAsNull,
+ slackTs: compareRValStringAsNumber,
+ engagement: compareRValUndefinedAsNull,
+ totalVotes: compareRValUndefinedAsNull,
+ maxVotesPerGroup: compareRValUndefinedAsNull,
+ disableAnonymity: compareRValUndefinedAsNull,
+ commentCount: compareRValUndefinedAsNull,
+ taskCount: compareRValUndefinedAsNull,
+ agendaItemCount: compareRValUndefinedAsNull,
+ storyCount: compareRValUndefinedAsNull,
+ templateId: compareRValUndefinedAsNull,
+ topicCount: compareRValUndefinedAsNull,
+ reflectionCount: compareRValUndefinedAsNull,
+ transcription: compareRValUndefinedAsNull,
+ recallBotId: compareRValUndefinedAsNull,
+ videoMeetingURL: compareRValUndefinedAsNull,
+ autogroupReflectionGroups: compareRValUndefinedAsNull,
+ resetReflectionGroups: compareRValUndefinedAsNull,
+ templateRefId: compareRValUndefinedAsNull,
+ meetingPrompt: compareRValUndefinedAsNull
  },
  maxErrors
  )

packages/server/postgres/migrations/1726602922665_NewMeeting-phase2.ts

@@ -0,0 +1,170 @@
+import {Kysely, PostgresDialect, sql} from 'kysely'
+import {r} from 'rethinkdb-ts'
+import connectRethinkDB from '../../database/connectRethinkDB'
+import getPg from '../getPg'
+
+export async function up() {
+ await connectRethinkDB()
+ const pg = new Kysely<any>({
+ dialect: new PostgresDialect({
+ pool: getPg()
+ })
+ })
+
+ try {
+ console.log('Adding index')
+ await r
+ .table('NewMeeting')
+ .indexCreate('updatedAtId', (row: any) => [row('updatedAt'), row('id')])
+ .run()
+ await r.table('NewMeeting').indexWait().run()
+ } catch {
+ // index already exists
+ }
+
+ console.log('Adding index complete')
+
+ await sql`ALTER TABLE "NewMeeting" DISABLE TRIGGER "check_meeting_overlap"`.execute(pg)
+ const MAX_PG_PARAMS = 65545
+ const PG_COLS = [
+ 'id',
+ 'isLegacy',
+ 'createdAt',
+ 'updatedAt',
+ 'createdBy',
+ 'endedAt',
+ 'facilitatorStageId',
+ 'facilitatorUserId',
+ 'meetingCount',
+ 'meetingNumber',
+ 'name',
+ 'summarySentAt',
+ 'teamId',
+ 'meetingType',
+ 'phases',
+ 'showConversionModal',
+ 'meetingSeriesId',
+ 'scheduledEndTime',
+ 'summary',
+ 'sentimentScore',
+ 'usedReactjis',
+ 'slackTs',
+ 'engagement',
+ 'totalVotes',
+ 'maxVotesPerGroup',
+ 'disableAnonymity',
+ 'commentCount',
+ 'taskCount',
+ 'agendaItemCount',
+ 'storyCount',
+ 'templateId',
+ 'topicCount',
+ 'reflectionCount',
+ 'transcription',
+ 'recallBotId',
+ 'videoMeetingURL',
+ 'autogroupReflectionGroups',
+ 'resetReflectionGroups',
+ 'templateRefId',
+ 'meetingPrompt'
+ ] as const
+ type NewMeeting = {
+ [K in (typeof PG_COLS)[number]]: any
+ }
+ const BATCH_SIZE = Math.trunc(MAX_PG_PARAMS / PG_COLS.length)
+
+ let curUpdatedAt = r.minval
+ let curId = r.minval
+
+ const insertRow = async (row) => {
+ if (!row.facilitatorStageId) {
+ console.log('Meeting has no facilitatorId, skipping insert', row.id, row.teamId)
+ return
+ }
+ try {
+ await pg
+ .insertInto('NewMeeting')
+ .values(row)
+ .onConflict((oc) => oc.doNothing())
+ .execute()
+ } catch (e) {
+ if (e.constraint === 'fk_createdBy') {
+ return insertRow({...row, createdBy: null})
+ }
+ if (e.constraint === 'fk_facilitatorUserId') {
+ return insertRow({...row, facilitatorUserId: null})
+ }
+ if (e.constraint === 'fk_teamId') {
+ console.log('Meeting has no team, skipping insert', row.id)
+ return
+ }
+ if (e.constraint === 'fk_meetingSeriesId') {
+ return insertRow({...row, meetingSeriesId: null})
+ }
+ if (e.constraint === 'fk_templateId') {
+ console.log('Meeting has no template, skipping insert', row.id)
+ return
+ }
+ throw e
+ }
+ }
+ for (let i = 0; i < 1e6; i++) {
+ console.log('inserting row', i * BATCH_SIZE, String(curUpdatedAt), String(curId))
+ const rawRowsToInsert = (await r
+ .table('NewMeeting')
+ .between([curUpdatedAt, curId], [r.maxval, r.maxval], {
+ index: 'updatedAtId',
+ leftBound: 'open',
+ rightBound: 'closed'
+ })
+ .orderBy({index: 'updatedAtId'})
+ .limit(BATCH_SIZE)
+ .pluck(...PG_COLS)
+ .run()) as NewMeeting[]
+
+ const rowsToInsert = rawRowsToInsert.map((row) => {
+ const {
+ phases,
+ name,
+ summary,
+ usedReactjis,
+ slackTs,
+ transcription,
+ autogroupReflectionGroups,
+ resetReflectionGroups,
+ meetingPrompt,
+ meetingCount,
+ ...rest
+ } = row as any
+ return {
+ ...rest,
+ phases: JSON.stringify(phases),
+ name: name.slice(0, 100),
+ summary: summary ? summary.slice(0, 10000) : null,
+ usedReactjis: JSON.stringify(usedReactjis),
+ slackTs: isNaN(Number(slackTs)) ? null : Number(slackTs),
+ transcription: JSON.stringify(transcription),
+ autogroupReflectionGroups: JSON.stringify(autogroupReflectionGroups),
+ resetReflectionGroups: JSON.stringify(resetReflectionGroups),
+ meetingPrompt: meetingPrompt ? meetingPrompt.slice(0, 255) : null,
+ meetingCount: meetingCount || 0
+ }
+ })
+
+ if (rowsToInsert.length === 0) break
+ const lastRow = rowsToInsert[rowsToInsert.length - 1]
+ curUpdatedAt = lastRow.updatedAt
+ curId = lastRow.id
+ await Promise.all(rowsToInsert.map(async (row) => insertRow(row)))
+ }
+ await sql`ALTER TABLE "NewMeeting" ENABLE TRIGGER "check_meeting_overlap"`.execute(pg)
+}
+
+export async function down() {
+ const pg = new Kysely<any>({
+ dialect: new PostgresDialect({
+ pool: getPg()
+ })
+ })
+ await sql`TRUNCATE TABLE "NewMeeting" CASCADE`.execute(pg)
+}

packages/server/postgres/utils/checkEqBase.ts

@@ -65,7 +65,7 @@ export async function checkTableEq(
  const pgRow = pgRowsById[id]
 
  if (!pgRow) {
- errors.push({id, prop: id, rVal: id, pgVal: null})
+ errors.push({id, prop: '', rVal: null, pgVal: null})
  if (errors.length >= maxErrors) return errors
  continue
  }

packages/server/postgres/utils/rethinkEqualityFns.ts

@@ -1,10 +1,32 @@
 import isValidDate from 'parabol-client/utils/isValidDate'
 import stringSimilarity from 'string-similarity'
 
+function sortObjectKeys(obj: any): any {
+ if (Array.isArray(obj)) {
+ // If it's an array, recurse into each element
+ return obj.map(sortObjectKeys)
+ } else if (obj !== null && typeof obj === 'object') {
+ if (obj instanceof Date) return obj
+ // If it's an object, sort the keys and recurse on each value
+ const sortedObj: {[key: string]: any} = {}
+ Object.keys(obj)
+ .sort()
+ .forEach((key) => {
+ sortedObj[key] = sortObjectKeys(obj[key])
+ })
+ return sortedObj
+ } else {
+ // If it's a primitive value, just return it
+ return obj
+ }
+}
+
 export const defaultEqFn = (a: unknown, b: unknown) => {
  if (a instanceof Date && b instanceof Date) return a.getTime() === b.getTime()
- if (Array.isArray(a) && Array.isArray(b)) return JSON.stringify(a) === JSON.stringify(b)
- if (typeof a === 'object' && typeof b === 'object') return JSON.stringify(a) === JSON.stringify(b)
+ if (Array.isArray(a) && Array.isArray(b))
+ return JSON.stringify(sortObjectKeys(a)) === JSON.stringify(sortObjectKeys(b))
+ if (typeof a === 'object' && typeof b === 'object')
+ return JSON.stringify(sortObjectKeys(a)) === JSON.stringify(sortObjectKeys(b))
  return a === b
 }
 export const compareDateAlmostEqual = (rVal: unknown, pgVal: unknown) => {
@@ -47,7 +69,7 @@ export const compareRValUndefinedAsEmptyArray = (rVal: unknown, pgVal: unknown)
 }
 
 export const compareRValStringAsNumber = (rVal: unknown, pgVal: unknown) => {
- const normalizedRVal = Number(rVal)
+ const normalizedRVal = rVal ? Number(rVal) : null
  return defaultEqFn(normalizedRVal, pgVal)
 }