chore: migrate EmailVerification to pg

packages/server/__tests__/autoJoin.test.ts

@@ -1,6 +1,6 @@
 import faker from 'faker'
-import getRethink from '../database/rethinkDriver'
 import createEmailVerification from '../email/createEmailVerification'
+import getKysely from '../postgres/getKysely'
 import {getUserTeams, sendIntranet, sendPublic} from './common'
 
 const signUpVerified = async (email: string) => {
@@ -23,12 +23,14 @@ const signUpVerified = async (email: string) => {
   // manually generate verification token so also the founder can be verified
   await createEmailVerification({email, password})
 
-  const r = await getRethink()
-  const verificationToken = await r
-    .table('EmailVerification')
-    .getAll(email, {index: 'email'})
-    .nth(0)('token')
-    .run()
+  const pg = getKysely()
+  const verificationToken = (
+    await pg
+      .selectFrom('EmailVerification')
+      .select('token')
+      .where('email', '=', email)
+      .executeTakeFirstOrThrow(() => new Error(`No verification token found for ${email}`))
+  ).token
 
   const verifyEmail = await sendPublic({
     query: `
@@ -55,9 +57,9 @@ const signUpVerified = async (email: string) => {
   expect(verifyEmail).toMatchObject({
     data: {
       verifyEmail: {
-        authToken: expect.toBeString(),
+        authToken: expect.any(String),
         user: {
-          id: expect.toBeString()
+          id: expect.any(String)
         }
       }
     }
@@ -153,7 +155,7 @@ test.skip('autoJoin on multiple teams does not create duplicate `OrganizationUse
   const newEmail = `${faker.internet.userName()}@${domain}`.toLowerCase()
   const {user: newUser} = await signUpVerified(newEmail)
 
-  expect(newUser.tms).toIncludeSameMembers(teamIds)
+  expect(newUser.tms).toEqual(expect.arrayContaining(teamIds))
   expect(newUser.organizations).toMatchObject([
     {
       id: orgId

packages/server/database/types/EmailVerification.ts

@@ -1,8 +1,6 @@
 import {Threshold} from 'parabol-client/types/constEnums'
-import generateUID from '../../generateUID'
 
 interface Input {
-  id?: string
   token: string
   email: string
   expiration?: Date
@@ -12,16 +10,14 @@ interface Input {
 }
 
 export default class EmailVerification {
-  id: string
   invitationToken?: string
   token: string
   email: string
   expiration: Date
   hashedPassword?: string
   pseudoId?: string
   constructor(input: Input) {
-    const {id, invitationToken, token, email, expiration, hashedPassword, pseudoId} = input
-    this.id = id || generateUID()
+    const {invitationToken, token, email, expiration, hashedPassword, pseudoId} = input
     this.invitationToken = invitationToken || undefined
     this.token = token
     this.email = email

packages/server/email/createEmailVerficationForExistingUser.ts

@@ -1,6 +1,6 @@
 import base64url from 'base64url'
 import crypto from 'crypto'
-import getRethink from '../database/rethinkDriver'
+import getKysely from '../postgres/getKysely'
 import AuthIdentityLocal from '../database/types/AuthIdentityLocal'
 import EmailVerification from '../database/types/EmailVerification'
 import {DataLoaderWorker} from '../graphql/graphql'
@@ -34,15 +34,15 @@ const createEmailVerficationForExistingUser = async (
   if (!success) {
     return new Error('Unable to send verification email')
   }
-  const r = await getRethink()
   const emailVerification = new EmailVerification({
     email,
     token: verifiedEmailToken,
     hashedPassword,
     pseudoId,
     invitationToken
   })
-  await r.table('EmailVerification').insert(emailVerification).run()
+  const pg = getKysely()
+  await pg.insertInto('EmailVerification').values(emailVerification).execute()
   return undefined
 }
 

packages/server/email/createEmailVerification.ts

@@ -2,7 +2,7 @@ import base64url from 'base64url'
 import bcrypt from 'bcryptjs'
 import crypto from 'crypto'
 import {Security} from 'parabol-client/types/constEnums'
-import getRethink from '../database/rethinkDriver'
+import getKysely from '../postgres/getKysely'
 import EmailVerification from '../database/types/EmailVerification'
 import emailVerificationEmailCreator from './emailVerificationEmailCreator'
 import getMailManager from './getMailManager'
@@ -35,7 +35,6 @@ const createEmailVerification = async (props: SignUpWithPasswordMutationVariable
   if (!success) {
     return {error: {message: 'Unable to send verification email'}}
   }
-  const r = await getRethink()
   const hashedPassword = await bcrypt.hash(password, Security.SALT_ROUNDS)
   const emailVerification = new EmailVerification({
     email,
@@ -44,7 +43,8 @@ const createEmailVerification = async (props: SignUpWithPasswordMutationVariable
     pseudoId,
     invitationToken
   })
-  await r.table('EmailVerification').insert(emailVerification).run()
+  const pg = getKysely()
+  await pg.insertInto('EmailVerification').values(emailVerification).execute()
   return {error: {message: 'Verification required. Check your inbox.'}}
 }
 

packages/server/graphql/public/mutations/signUpWithPassword.ts

@@ -1,10 +1,8 @@
 import bcrypt from 'bcryptjs'
 import {AuthenticationError, Security} from 'parabol-client/types/constEnums'
-import {URLSearchParams} from 'url'
-import getRethink from '../../../database/rethinkDriver'
-import {RValue} from '../../../database/stricterR'
 import createEmailVerification from '../../../email/createEmailVerification'
 import {USER_PREFERRED_NAME_LIMIT} from '../../../postgres/constants'
+import getKysely from '../../../postgres/getKysely'
 import createNewLocalUser from '../../../utils/createNewLocalUser'
 import encodeAuthToken from '../../../utils/encodeAuthToken'
 import isEmailVerificationRequired from '../../../utils/isEmailVerificationRequired'
@@ -21,7 +19,7 @@ const signUpWithPassword: MutationResolvers['signUpWithPassword'] = async (
   if (email.length > USER_PREFERRED_NAME_LIMIT) {
     return {error: {message: 'Email is too long'}}
   }
-  const r = await getRethink()
+  const pg = getKysely()
   const isOrganic = !invitationToken
   const {ip, dataLoader} = context
   const loginAttempt = await attemptLogin(email, password, ip)
@@ -49,13 +47,13 @@ const signUpWithPassword: MutationResolvers['signUpWithPassword'] = async (
   }
   const verificationRequired = await isEmailVerificationRequired(email, dataLoader)
   if (verificationRequired) {
-    const existingVerification = await r
-      .table('EmailVerification')
-      .getAll(email, {index: 'email'})
-      .filter((row: RValue) => row('expiration').gt(new Date()))
-      .nth(0)
-      .default(null)
-      .run()
+    const existingVerification =
+      (await pg
+        .selectFrom('EmailVerification')
+        .selectAll()
+        .where('email', '=', email)
+        .where('expiration', '>', new Date())
+        .executeTakeFirst()) || null
     if (existingVerification) {
       return {error: {message: 'Verification email already sent'}}
     }

packages/server/graphql/public/mutations/verifyEmail.ts

@@ -1,8 +1,8 @@
 import {AuthIdentityTypeEnum} from '../../../../client/types/constEnums'
-import getRethink from '../../../database/rethinkDriver'
 import AuthIdentityLocal from '../../../database/types/AuthIdentityLocal'
 import AuthToken from '../../../database/types/AuthToken'
 import EmailVerification from '../../../database/types/EmailVerification'
+import getKysely from '../../../postgres/getKysely'
 import {getUserByEmail} from '../../../postgres/queries/getUsersByEmails'
 import updateUser from '../../../postgres/queries/updateUser'
 import createNewLocalUser from '../../../utils/createNewLocalUser'
@@ -16,14 +16,14 @@ const verifyEmail: MutationResolvers['verifyEmail'] = async (
   context
 ) => {
   const {dataLoader} = context
-  const r = await getRethink()
+  const pg = getKysely()
   const now = new Date()
-  const emailVerification = (await r
-    .table('EmailVerification')
-    .getAll(verificationToken, {index: 'token'})
-    .nth(0)
-    .default(null)
-    .run()) as EmailVerification
+  const emailVerification =
+    ((await pg
+      .selectFrom('EmailVerification')
+      .selectAll()
+      .where('token', '=', verificationToken)
+      .executeTakeFirst()) as EmailVerification) || null
 
   if (!emailVerification) {
     return {error: {message: 'Invalid verification token'}}

packages/server/postgres/migrations/1709351538000_addEmailVerification.ts

@@ -0,0 +1,31 @@
+import {Client} from 'pg'
+import getPgConfig from '../getPgConfig'
+
+export async function up() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`
+    CREATE TABLE "EmailVerification" (
+      "id" SERIAL PRIMARY KEY,
+      "email" "citext" NOT NULL,
+      "expiration" TIMESTAMP WITH TIME ZONE NOT NULL,
+      "token" VARCHAR(100) NOT NULL,
+      "hashedPassword" VARCHAR(100),
+      "invitationToken" VARCHAR(100),
+      "pseudoId" VARCHAR(100)
+    );
+
+    CREATE INDEX IF NOT EXISTS "idx_EmailVerification_email" ON "EmailVerification"("email");
+    CREATE INDEX IF NOT EXISTS "idx_EmailVerification_token" ON "EmailVerification"("token");
+  `)
+  await client.end()
+}
+
+export async function down() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`
+    DROP TABLE IF EXISTS "EmailVerification";
+  `)
+  await client.end()
+}

packages/server/postgres/migrations/1709351575000_moveEmailVerification.ts

@@ -0,0 +1,76 @@
+import {FirstParam} from 'parabol-client/types/generics'
+import {Client} from 'pg'
+import {r} from 'rethinkdb-ts'
+import getPgConfig from '../getPgConfig'
+import connectRethinkDB from '../../database/connectRethinkDB'
+import getPgp from '../getPgp'
+
+export async function up() {
+  await connectRethinkDB()
+  const {pgp, pg} = getPgp()
+  const batchSize = 1000
+
+  try {
+    await r.table('EmailVerification').indexCreate('expiration').run()
+    await r.table('EmailVerification').indexWait().run()
+  } catch {}
+
+  const columnSet = new pgp.helpers.ColumnSet(
+    [
+      'email',
+      'expiration',
+      'hashedPassword',
+      'token',
+      {name: 'invitationToken', def: null},
+      {name: 'pseudoId', def: null}
+    ],
+    {table: 'EmailVerification'}
+  )
+
+  const getNextData = async (leftBoundCursor: Date | undefined) => {
+    const expiration = leftBoundCursor || r.minval
+    const nextBatch = await r
+      .table('EmailVerification')
+      .between(expiration, r.maxval, {index: 'expiration', leftBound: 'open'})
+      .orderBy({index: 'expiration'})
+      .limit(batchSize)
+      .run()
+    if (nextBatch.length === 0) return null
+    if (nextBatch.length < batchSize) return nextBatch
+    const lastItem = nextBatch.pop()
+    const lastMatchingExpiration = nextBatch.findLastIndex(
+      (item) => item.expiration !== lastItem!.expiration
+    )
+    if (lastMatchingExpiration === -1) {
+      throw new Error(
+        'batchSize is smaller than the number of items that share the same cursor. Increase batchSize'
+      )
+    }
+    return nextBatch.slice(0, lastMatchingExpiration)
+  }
+
+  await pg.tx('EmailVerification', (task) => {
+    const fetchAndProcess: FirstParam<typeof task.sequence> = async (
+      _index,
+      leftBoundCursor: undefined | Date
+    ) => {
+      const nextData = await getNextData(leftBoundCursor)
+      if (!nextData) return undefined
+      const insert = pgp.helpers.insert(nextData, columnSet)
+      await task.none(insert)
+      return nextData.at(-1)!.runAt
+    }
+    return task.sequence(fetchAndProcess)
+  })
+  await r.getPoolMaster()?.drain()
+}
+
+export async function down() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`DELETE FROM "EmailVerification"`)
+  await client.end()
+  try {
+    await r.table('EmailVerification').indexDrop('expiration').run()
+  } catch {}
+}