chore(rethinkdb): OrganizationUser: Phase 2

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

@@ -2,17 +2,7 @@ import getRethink from '../../../database/rethinkDriver'
 import getFileStoreManager from '../../../fileStorage/getFileStoreManager'
 import getKysely from '../../../postgres/getKysely'
 import {checkRowCount, checkTableEq} from '../../../postgres/utils/checkEqBase'
-import {
-  compareDateAlmostEqual,
-  compareRValOptionalPluckedObject,
-  compareRValStringAsNumber,
-  compareRValUndefinedAsEmptyArray,
-  compareRValUndefinedAsFalse,
-  compareRValUndefinedAsNull,
-  compareRValUndefinedAsNullAndTruncateRVal,
-  compareRValUndefinedAsZero,
-  defaultEqFn
-} from '../../../postgres/utils/rethinkEqualityFns'
+import {compareRValUndefinedAsNull, defaultEqFn} from '../../../postgres/utils/rethinkEqualityFns'
 import {MutationResolvers} from '../resolverTypes'
 
 const handleResult = async (
@@ -37,55 +27,35 @@ const checkRethinkPgEquality: MutationResolvers['checkRethinkPgEquality'] = asyn
 ) => {
   const r = await getRethink()
 
-  if (tableName === 'Organization') {
+  if (tableName === 'OrganizationUser') {
     const rowCountResult = await checkRowCount(tableName)
-    const rethinkQuery = (updatedAt: Date, id: string | number) => {
+    const rethinkQuery = (joinedAt: Date, id: string | number) => {
       return r
-        .table('Organization' as any)
-        .between([updatedAt, id], [r.maxval, r.maxval], {
-          index: 'updatedAtId',
+        .table('OrganizationUser' as any)
+        .between([joinedAt, id], [r.maxval, r.maxval], {
+          index: 'joinedAtId',
           leftBound: 'open',
           rightBound: 'closed'
         })
-        .orderBy({index: 'updatedAtId'}) as any
+        .orderBy({index: 'joinedAtId'}) as any
     }
     const pgQuery = async (ids: string[]) => {
-      return getKysely()
-        .selectFrom('Organization')
-        .selectAll()
-        .select(({fn}) => [fn('to_json', ['creditCard']).as('creditCard')])
-        .where('id', 'in', ids)
-        .execute()
+      return getKysely().selectFrom('OrganizationUser').selectAll().where('id', 'in', ids).execute()
     }
     const errors = await checkTableEq(
       rethinkQuery,
       pgQuery,
       {
         id: defaultEqFn,
-        activeDomain: compareRValUndefinedAsNullAndTruncateRVal(100),
-        isActiveDomainTouched: compareRValUndefinedAsFalse,
-        creditCard: compareRValOptionalPluckedObject({
-          brand: compareRValUndefinedAsNull,
-          expiry: compareRValUndefinedAsNull,
-          last4: compareRValStringAsNumber
-        }),
-        createdAt: defaultEqFn,
-        name: compareRValUndefinedAsNullAndTruncateRVal(100),
-        payLaterClickCount: compareRValUndefinedAsZero,
-        periodEnd: compareRValUndefinedAsNull,
-        periodStart: compareRValUndefinedAsNull,
-        picture: compareRValUndefinedAsNull,
-        showConversionModal: compareRValUndefinedAsFalse,
-        stripeId: compareRValUndefinedAsNull,
-        stripeSubscriptionId: compareRValUndefinedAsNull,
-        upcomingInvoiceEmailSentAt: compareRValUndefinedAsNull,
+        suggestedTier: compareRValUndefinedAsNull,
+        inactive: defaultEqFn,
+        joinedAt: defaultEqFn,
+        orgId: defaultEqFn,
+        removedAt: defaultEqFn,
+        role: compareRValUndefinedAsNull,
+        userId: defaultEqFn,
         tier: defaultEqFn,
-        tierLimitExceededAt: compareRValUndefinedAsNull,
-        trialStartDate: compareRValUndefinedAsNull,
-        scheduledLockAt: compareRValUndefinedAsNull,
-        lockedAt: compareRValUndefinedAsNull,
-        updatedAt: compareDateAlmostEqual,
-        featureFlags: compareRValUndefinedAsEmptyArray
+        trialStartDate: compareRValUndefinedAsNull
       },
       maxErrors
     )

packages/server/postgres/migrations/1720640784862_OrganizationUser-phase2.ts

@@ -0,0 +1,110 @@
+import {Kysely, PostgresDialect} 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('OrganizationUser')
+      .indexCreate('joinedAtId', (row: any) => [row('joinedAt'), row('id')])
+      .run()
+    await r.table('OrganizationUser').indexWait().run()
+  } catch {
+    // index already exists
+  }
+  await r.table('OrganizationUser').get('aGhostOrganizationUser').update({tier: 'enterprise'}).run()
+  await console.log('Adding index complete')
+  const MAX_PG_PARAMS = 65545
+  const PG_COLS = [
+    'id',
+    'suggestedTier',
+    'inactive',
+    'joinedAt',
+    'orgId',
+    'removedAt',
+    'role',
+    'userId',
+    'tier',
+    'trialStartDate'
+  ] as const
+  type OrganizationUser = {
+    [K in (typeof PG_COLS)[number]]: any
+  }
+  const BATCH_SIZE = Math.trunc(MAX_PG_PARAMS / PG_COLS.length)
+
+  let curjoinedAt = r.minval
+  let curId = r.minval
+  for (let i = 0; i < 1e6; i++) {
+    console.log('inserting row', i * BATCH_SIZE, curjoinedAt, curId)
+    const rawRowsToInsert = (await r
+      .table('OrganizationUser')
+      .between([curjoinedAt, curId], [r.maxval, r.maxval], {
+        index: 'joinedAtId',
+        leftBound: 'open',
+        rightBound: 'closed'
+      })
+      .orderBy({index: 'joinedAtId'})
+      .limit(BATCH_SIZE)
+      .pluck(...PG_COLS)
+      .run()) as OrganizationUser[]
+
+    const rowsToInsert = rawRowsToInsert.map((row) => {
+      const {newUserUntil, ...rest} = row as any
+      return {
+        ...rest
+      }
+    })
+    if (rowsToInsert.length === 0) break
+    const lastRow = rowsToInsert[rowsToInsert.length - 1]
+    curjoinedAt = lastRow.joinedAt
+    curId = lastRow.id
+    try {
+      await pg
+        .insertInto('OrganizationUser')
+        .values(rowsToInsert)
+        .onConflict((oc) => oc.doNothing())
+        .execute()
+    } catch (e) {
+      await Promise.all(
+        rowsToInsert.map(async (row) => {
+          try {
+            await pg
+              .insertInto('OrganizationUser')
+              .values(row)
+              .onConflict((oc) => oc.doNothing())
+              .execute()
+          } catch (e) {
+            if (e.constraint === 'fk_userId' || e.constraint === 'fk_orgId') {
+              console.log(`Skipping ${row.id} because it has no user/org`)
+              return
+            }
+            console.log(e, row)
+          }
+        })
+      )
+    }
+  }
+}
+
+export async function down() {
+  await connectRethinkDB()
+  try {
+    await r.table('OrganizationUser').indexDrop('joinedAtId').run()
+  } catch {
+    // index already dropped
+  }
+  const pg = new Kysely<any>({
+    dialect: new PostgresDialect({
+      pool: getPg()
+    })
+  })
+  await pg.deleteFrom('OrganizationUser').execute()
+}

packages/server/postgres/utils/checkEqBase.ts

@@ -33,7 +33,7 @@ export const checkRowCount = async (tableName: string) => {
 }
 
 export async function checkTableEq(
-  rethinkQuery: (updatedAt: Date, id: string | number) => RSelection,
+  rethinkQuery: (joinedAt: Date, id: string | number) => RSelection,
   pgQuery: (ids: string[]) => Promise<PGDoc[] | null>,
   equalityMap: Record<string, (a: unknown, b: unknown) => boolean>,
   maxErrors: number | null | undefined
@@ -51,7 +51,7 @@ export async function checkTableEq(
       .run()) as RethinkDoc[]
     if (rethinkRows.length === 0) break
     const lastRow = rethinkRows[rethinkRows.length - 1]!
-    curUpdatedDate = lastRow.updatedAt
+    curUpdatedDate = lastRow.joinedAt
     curId = lastRow.id
     const ids = rethinkRows.map((t) => t.id)
     const pgRows = (await pgQuery(ids)) ?? []