chore(rethinkdb): RetroReflection: Phase 2

packages/client/utils/constants.ts

@@ -5,10 +5,10 @@
  - Does the variable come from the GraphQL schema? If so, import it from a file in the __generated__ folder
  - Is the variable a string? Create a string union & pass in a plain string to get type safety
 */
-import {TaskStatusEnum} from '~/__generated__/UpdateTaskMutation.graphql'
 import {ReadableReasonToDowngradeEnum} from '../../server/graphql/types/ReasonToDowngrade'
 import {ReasonToDowngradeEnum} from '../__generated__/DowngradeToStarterMutation.graphql'
 import {TimelineEventEnum} from '../__generated__/MyDashboardTimelineQuery.graphql'
+import {TaskStatusEnum} from '../__generated__/UpdateTaskMutation.graphql'
 import {Threshold} from '../types/constEnums'
 
 /* Meeting Misc. */

packages/server/graphql/mutations/helpers/autoGroup/sanitizeAnalyzedEntititesResponse.ts

@@ -4,7 +4,8 @@ const sanitizeAnalyzedEntitiesResponse = (response: GoogleAnalyzedEntities | nul
   if (!response) return null
   const {entities} = response
   if (!Array.isArray(entities)) return null
-  const validEntities = {} as {[lowerCaseName: string]: number}
+  // very important to Object.create(null) since validEntities['constructor'] would return a function!
+  const validEntities = Object.create(null) as {[lowerCaseName: string]: number}
   entities.forEach((entity) => {
     const {name, salience} = entity
     if (!name || !salience) return

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

@@ -4,7 +4,12 @@ import getKysely from '../../../postgres/getKysely'
 import {checkRowCount, checkTableEq} from '../../../postgres/utils/checkEqBase'
 import {
   compareDateAlmostEqual,
+  compareOptionalPlaintextContent,
+  compareRValOptionalPluckedArray,
+  compareRValUndefinedAsEmptyArray,
+  compareRValUndefinedAsNull,
   compareRValUndefinedAsNullAndTruncateRVal,
+  compareRealNumber,
   defaultEqFn
 } from '../../../postgres/utils/rethinkEqualityFns'
 import {MutationResolvers} from '../resolverTypes'
@@ -27,43 +32,58 @@ const handleResult = async (
 
 const checkRethinkPgEquality: MutationResolvers['checkRethinkPgEquality'] = async (
   _source,
-  {tableName, writeToFile}
+  {tableName, writeToFile, maxErrors}
 ) => {
   const r = await getRethink()
 
-  if (tableName === 'RetroReflectionGroup') {
+  if (tableName === 'RetroReflection') {
     const rowCountResult = await checkRowCount(tableName)
     const rethinkQuery = (updatedAt: Date, id: string | number) => {
       return r
-        .table('RetroReflectionGroup' as any)
+        .table('RetroReflection')
         .between([updatedAt, id], [r.maxval, r.maxval], {
           index: 'updatedAtId',
           leftBound: 'open',
           rightBound: 'closed'
         })
         .orderBy({index: 'updatedAtId'}) as any
     }
-    const pgQuery = (ids: string[]) => {
+    const pgQuery = async (ids: string[]) => {
       return getKysely()
-        .selectFrom('RetroReflectionGroup')
+        .selectFrom('RetroReflection')
         .selectAll()
+        .select(({fn}) => [
+          fn('to_json', ['entities']).as('entities'),
+          fn('to_json', ['reactjis']).as('reactjis')
+        ])
         .where('id', 'in', ids)
         .execute()
     }
-    const errors = await checkTableEq(rethinkQuery, pgQuery, {
-      id: defaultEqFn,
-      createdAt: defaultEqFn,
-      updatedAt: compareDateAlmostEqual,
-      isActive: defaultEqFn,
-      meetingId: defaultEqFn,
-      promptId: defaultEqFn,
-      sortOrder: defaultEqFn,
-      voterIds: defaultEqFn,
-      smartTitle: compareRValUndefinedAsNullAndTruncateRVal(255),
-      title: compareRValUndefinedAsNullAndTruncateRVal(255),
-      summary: compareRValUndefinedAsNullAndTruncateRVal(2000),
-      discussionPromptQuestion: compareRValUndefinedAsNullAndTruncateRVal(2000)
-    })
+    const errors = await checkTableEq(
+      rethinkQuery,
+      pgQuery,
+      {
+        id: defaultEqFn,
+        createdAt: defaultEqFn,
+        updatedAt: compareDateAlmostEqual,
+        isActive: defaultEqFn,
+        meetingId: defaultEqFn,
+        promptId: defaultEqFn,
+        creatorId: compareRValUndefinedAsNull,
+        sortOrder: defaultEqFn,
+        reflectionGroupId: defaultEqFn,
+        content: compareRValUndefinedAsNullAndTruncateRVal(2000, 0.19),
+        plaintextContent: compareOptionalPlaintextContent,
+        entities: compareRValOptionalPluckedArray({
+          name: defaultEqFn,
+          salience: compareRealNumber,
+          lemma: compareRValUndefinedAsNull
+        }),
+        reactjis: compareRValUndefinedAsEmptyArray,
+        sentimentScore: compareRValUndefinedAsNull
+      },
+      maxErrors
+    )
     return handleResult(tableName, rowCountResult, errors, writeToFile)
   }
   return 'Table not found'

packages/server/postgres/migrations/1714598525167_addFavoriteTemplateIds.ts

@@ -10,7 +10,7 @@ export async function up() {
 
   await sql`
   ALTER TABLE "User"
-  ADD COLUMN "favoriteTemplateIds" TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[];
+  ADD COLUMN IF NOT EXISTS "favoriteTemplateIds" TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[];
 `.execute(pg)
 }
 

packages/server/postgres/migrations/1716914102795_removeKudos.ts

@@ -8,12 +8,16 @@ export async function up() {
     })
   })
 
-  await pg.schema.dropTable('Kudos').execute()
-  await pg.schema
-    .alterTable('Team')
-    .dropColumn('giveKudosWithEmoji')
-    .dropColumn('kudosEmoji')
-    .execute()
+  await pg.schema.dropTable('Kudos').ifExists().execute()
+  try {
+    await pg.schema
+      .alterTable('Team')
+      .dropColumn('giveKudosWithEmoji')
+      .dropColumn('kudosEmoji')
+      .execute()
+  } catch {
+    // noop
+  }
 }
 
 export async function down() {

packages/server/postgres/migrations/1716995191300_allowGlobalOAuth1Provider.ts

@@ -8,7 +8,7 @@ export async function up() {
   DO $$
   BEGIN
     ALTER TABLE "IntegrationProvider"
-      DROP CONSTRAINT global_provider_must_be_oauth2;
+      DROP CONSTRAINT IF EXISTS global_provider_must_be_oauth2;
   END $$;
   `)
   await client.end()

packages/server/postgres/migrations/1719162354561_RetroReflection-phase2.ts

@@ -0,0 +1,166 @@
+import {ContentState, convertToRaw} from 'draft-js'
+import {Kysely, PostgresDialect, sql} from 'kysely'
+import extractTextFromDraftString from 'parabol-client/utils/draftjs/extractTextFromDraftString'
+import {r} from 'rethinkdb-ts'
+import connectRethinkDB from '../../database/connectRethinkDB'
+import getPg from '../getPg'
+
+const convertTextToRaw = (text: string) => {
+  // plaintextContent can have a bunch of linebreaks like \n which get converted into new blocks.
+  // New blocks take up a BUNCH of space, so we'd rather preserve as much plaintextContent as possible.
+  const spaceFreeText = text
+    .split(/\s/)
+    .filter((s) => s.length)
+    .join(' ')
+  const contentState = ContentState.createFromText(spaceFreeText)
+  const raw = convertToRaw(contentState)
+  return JSON.stringify(raw)
+}
+
+export async function up() {
+  await connectRethinkDB()
+  const pg = new Kysely<any>({
+    dialect: new PostgresDialect({
+      pool: getPg()
+    })
+  })
+  try {
+    await r
+      .table('RetroReflection')
+      .indexCreate('updatedAtId', (row: any) => [row('updatedAt'), row('id')])
+      .run()
+    await r.table('RetroReflection').indexWait().run()
+  } catch {
+    // index already exists
+  }
+
+  const MAX_PG_PARAMS = 65545
+
+  const PG_COLS = [
+    'id',
+    'createdAt',
+    'updatedAt',
+    'isActive',
+    'meetingId',
+    'promptId',
+    'sortOrder',
+    'creatorId',
+    'content',
+    'plaintextContent',
+    'entities',
+    'sentimentScore',
+    'reactjis',
+    'reflectionGroupId'
+  ] as const
+  type RetroReflection = {
+    [K in (typeof PG_COLS)[number]]: any
+  }
+  const BATCH_SIZE = Math.trunc(MAX_PG_PARAMS / PG_COLS.length)
+
+  const capContent = (content: string, plaintextContent: string) => {
+    let nextPlaintextContent = plaintextContent || extractTextFromDraftString(content)
+    // if they got out of hand with formatting, extract the text & convert it back
+    let nextContent = content.length <= 2000 ? content : convertTextToRaw(nextPlaintextContent)
+    while (nextContent.length > 2000 || nextPlaintextContent.length > 2000) {
+      const maxLen = Math.max(nextContent.length, nextPlaintextContent.length)
+      const overage = maxLen - 2000
+      const stopIdx = nextPlaintextContent.length - overage - 1
+      nextPlaintextContent = nextPlaintextContent.slice(0, stopIdx)
+      nextContent = convertTextToRaw(nextPlaintextContent)
+    }
+    return {content: nextContent, plaintextContent: nextPlaintextContent}
+  }
+
+  let curUpdatedAt = r.minval
+  let curId = r.minval
+
+  for (let i = 0; i < 1e6; i++) {
+    console.log('inserting row', i * BATCH_SIZE, curUpdatedAt, curId)
+    const rawRowsToInsert = (await r
+      .table('RetroReflection')
+      .between([curUpdatedAt, curId], [r.maxval, r.maxval], {
+        index: 'updatedAtId',
+        leftBound: 'open',
+        rightBound: 'closed'
+      })
+      .orderBy({index: 'updatedAtId'})
+      .limit(BATCH_SIZE)
+      .pluck(...PG_COLS)
+      .run()) as RetroReflection[]
+
+    const rowsToInsert = rawRowsToInsert.map((row) => {
+      const nonzeroEntities = row.entities?.length > 0 ? row.entities : undefined
+      const normalizedEntities = nonzeroEntities?.map((e: any) => ({
+        ...e,
+        salience: typeof e.salience === 'number' ? e.salience : 0
+      }))
+      return {
+        ...row,
+        ...capContent(row.content, row.plaintextContent),
+        reactjis: row.reactjis?.map((r: any) => `(${r.id},${r.userId})`),
+        entities: normalizedEntities
+          ? sql`(select array_agg((name, salience, lemma)::"GoogleAnalyzedEntity") from json_populate_recordset(null::"GoogleAnalyzedEntity", ${JSON.stringify(normalizedEntities)}))`
+          : undefined
+      }
+    })
+    if (rowsToInsert.length === 0) break
+    const lastRow = rowsToInsert[rowsToInsert.length - 1]
+    curUpdatedAt = lastRow.updatedAt
+    curId = lastRow.id
+    let isFailure = false
+    // NOTE: This migration inserts row-by-row because there are so many referential integrity errors
+    // Do not copy this migration logic for future migrations, it is slow!
+    const insertSingleRow = async (row: RetroReflection) => {
+      if (isFailure) return
+      try {
+        await pg
+          .insertInto('RetroReflection')
+          .values(row)
+          .onConflict((oc) => oc.doNothing())
+          .execute()
+      } catch (e) {
+        if (e.constraint === 'fk_reflectionGroupId') {
+          await pg
+            .insertInto('RetroReflectionGroup')
+            .values({
+              id: row.reflectionGroupId,
+              createdAt: row.createdAt,
+              updatedAt: row.updatedAt,
+              isActive: row.isActive,
+              meetingId: row.meetingId,
+              promptId: row.promptId
+            })
+            // multiple reflections may be trying to create the same group
+            .onConflict((oc) => oc.doNothing())
+            .execute()
+          await insertSingleRow(row)
+        } else if (e.constraint === 'fk_creatorId') {
+          await r.table('RetroReflection').get(row.id).update({creatorId: null}).run()
+          await insertSingleRow({...row, creatorId: null})
+        } else {
+          isFailure = true
+          console.log(e, row)
+        }
+      }
+    }
+    await Promise.all(rowsToInsert.map(insertSingleRow))
+    if (isFailure) {
+      throw 'Failed batch'
+    }
+  }
+}
+
+export async function down() {
+  await connectRethinkDB()
+  try {
+    await r.table('RetroReflection').indexDrop('updatedAtId').run()
+  } catch {
+    // index already dropped
+  }
+  const pg = new Kysely<any>({
+    dialect: new PostgresDialect({
+      pool: getPg()
+    })
+  })
+  await pg.deleteFrom('RetroReflection').execute()
+}

packages/server/postgres/migrations/1719262354561_reactji-rename.ts

@@ -0,0 +1,20 @@
+import {Client} from 'pg'
+import getPgConfig from '../getPgConfig'
+
+export async function up() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+
+  await client.query(`
+    ALTER TYPE "Reactji" RENAME ATTRIBUTE "userid" to "userId";
+    ALTER TYPE "Reactji" RENAME ATTRIBUTE "shortname" to "id";
+    `)
+  await client.end()
+}
+
+export async function down() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`` /* Do undo magic */)
+  await client.end()
+}

packages/server/postgres/queries/getTeamPromptResponsesByIds.ts

@@ -21,11 +21,7 @@ export const mapToTeamPromptResponse = (
   return results.map((teamPromptResponse: any) => {
     return {
       ...teamPromptResponse,
-      id: TeamPromptResponseId.join(teamPromptResponse.id),
-      reactjis: teamPromptResponse.reactjis.map(
-        (reactji: {shortname: string; userid: string}) =>
-          new Reactji({id: reactji.shortname, userId: reactji.userid})
-      )
+      id: TeamPromptResponseId.join(teamPromptResponse.id)
     } as TeamPromptResponse
   })
 }

packages/server/postgres/utils/checkEqBase.ts

@@ -36,8 +36,9 @@ export async function checkTableEq(
   rethinkQuery: (updatedAt: Date, id: string | number) => RSelection,
   pgQuery: (ids: string[]) => Promise<PGDoc[] | null>,
   equalityMap: Record<string, (a: unknown, b: unknown) => boolean>,
-  maxErrors = 10
+  maxErrors: number | null | undefined
 ) {
+  maxErrors = maxErrors || 10
   const batchSize = 3000
   const errors = [] as Diff[]
   const propsToCheck = Object.keys(equalityMap)