chore(rethinkdb): QueryMap: One-shot (#10005)

Signed-off-by: Matt Krick <[email protected]>

packages/server/__tests__/common.ts

@@ -49,14 +49,18 @@ const persistFunction = (text: string) => {
 }
 
 const persistQuery = async (query: string) => {
-  const r = await getRethink()
+  const pg = getKysely()
   const id = persistFunction(query.trim())
   const record = {
     id,
     query,
     createdAt: new Date()
   }
-  await r.table('QueryMap').insert(record, {conflict: 'replace'}).run()
+  await pg
+    .insertInto('QueryMap')
+    .values(record)
+    .onConflict((oc) => oc.doNothing())
+    .execute()
   return id
 }
 

packages/server/database/rethinkDriver.ts

@@ -119,10 +119,6 @@ export type RethinkSchema = {
     type: PushInvitation
     index: 'userId'
   }
-  QueryMap: {
-    type: any
-    index: ''
-  }
   MeetingTemplate: {
     type: MeetingTemplate
     index: 'teamId' | 'orgId'

packages/server/graphql/CompiledQueryCache.ts

@@ -1,7 +1,7 @@
 import tracer from 'dd-trace'
 import {GraphQLSchema, parse} from 'graphql'
 import {CompiledQuery} from 'graphql-jit'
-import getRethink from '../database/rethinkDriver'
+import getKysely from '../postgres/getKysely'
 import {MutationResolvers, QueryResolvers, Resolver} from './public/resolverTypes'
 import {tracedCompileQuery} from './traceGraphQL'
 
@@ -42,8 +42,13 @@ export default class CompiledQueryCache {
   async fromID(docId: string, schema: GraphQLSchema) {
     const compiledQuery = this.store[docId]
     if (compiledQuery) return compiledQuery
-    const r = await getRethink()
-    let queryString = await r.table('QueryMap').get(docId)('query').default(null).run()
+    const pg = getKysely()
+    const queryStringRes = await pg
+      .selectFrom('QueryMap')
+      .select('query')
+      .where('id', '=', docId)
+      .executeTakeFirst()
+    let queryString = queryStringRes?.query
     if (!queryString && !__PRODUCTION__) {
       // try/catch block required for building the toolbox
       try {

packages/server/graphql/DocumentCache.ts

@@ -5,7 +5,7 @@
  */
 
 import {DocumentNode, parse} from 'graphql'
-import getRethink from '../database/rethinkDriver'
+import getKysely from '../postgres/getKysely'
 
 export default class DocumentCache {
   store = {} as {[docId: string]: DocumentNode}
@@ -16,8 +16,13 @@ export default class DocumentCache {
     // looks up query string for a persisted query, parses into an AST, caches and returns it
     let document = this.store[docId]
     if (!document) {
-      const r = await getRethink()
-      let queryString = await r.table('QueryMap').get(docId)('query').default(null).run()
+      const pg = getKysely()
+      const queryStringRes = await pg
+        .selectFrom('QueryMap')
+        .select('query')
+        .where('id', '=', docId)
+        .executeTakeFirst()
+      let queryString = queryStringRes?.query
       if (!queryString && !__PRODUCTION__) {
         // In development, use the frequently changing queryMap to look up persisted queries by hash
         const queryMap = require('../../../queryMap.json')

packages/server/postgres/migrations/1721348544381_QueryMapOneShot.ts

@@ -0,0 +1,36 @@
+import {Kysely, PostgresDialect, sql} from 'kysely'
+import {Client} from 'pg'
+import {r} from 'rethinkdb-ts'
+import connectRethinkDB from '../../database/connectRethinkDB'
+import getPg from '../getPg'
+import getPgConfig from '../getPgConfig'
+
+export async function up() {
+  await connectRethinkDB()
+  const pg = new Kysely<any>({
+    dialect: new PostgresDialect({
+      pool: getPg()
+    })
+  })
+  await sql`
+    CREATE TABLE IF NOT EXISTS "QueryMap" (
+      "id" VARCHAR(24) PRIMARY KEY,
+      "query" TEXT NOT NULL,
+      "createdAt" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
+    )
+`.execute(pg)
+  const queries = await r
+    .table('QueryMap')
+    .filter((row) => row('createdAt').default(null).ne(null))
+    .run()
+  await pg.insertInto('QueryMap').values(queries).execute()
+}
+
+export async function down() {
+  const client = new Client(getPgConfig())
+  await client.connect()
+  await client.query(`
+    DROP TABLE "QueryMap";
+    ` /* Do undo magic */)
+  await client.end()
+}

scripts/toolboxSrc/preDeploy.ts

@@ -1,7 +1,7 @@
 import dotenv from 'dotenv'
 import dotenvExpand from 'dotenv-expand'
+import getKysely from 'parabol-server/postgres/getKysely'
 import path from 'path'
-import getRethink from '../../packages/server/database/rethinkDriver'
 import queryMap from '../../queryMap.json'
 import getProjectRoot from '../webpack/utils/getProjectRoot'
 import {applyEnvVarsToClientAssets} from './applyEnvVarsToClientAssets'
@@ -21,13 +21,9 @@ const storePersistedQueries = async () => {
     createdAt: now
   }))
 
-  const r = await getRethink()
-  const res = await r.table('QueryMap').insert(records, {conflict: 'replace'}).run()
-  // without this sleep RethinkDB closes the connection before the query completes. It doesn't make sense!
-  await new Promise((resolve) => setTimeout(resolve, 50))
-  await r.getPoolMaster()?.drain()
-
-  console.log(`🔗 QueryMap Persistence Complete: ${res.inserted} records added`)
+  const pg = getKysely()
+  const res = await pg.insertInto('QueryMap').values(records).onConflict((oc) => oc.doNothing()).returning('id').execute()
+  console.log(`🔗 QueryMap Persistence Complete: ${res.length} records added`)
 }
 
 const preDeploy = async () => {
@@ -41,6 +37,7 @@ const preDeploy = async () => {
 
   // The we can prime the DB & CDN
   await Promise.all([storePersistedQueries(), primeIntegrations(), pushToCDN()])
+  await getKysely().destroy()
   console.log(`🚀 Predeploy Complete`)
   process.exit()
 }

scripts/toolboxSrc/primeIntegrations.ts

@@ -44,9 +44,7 @@ const upsertGlobalIntegrationProvidersFromEnv = async () => {
 
 const primeIntegrations = async () => {
   console.log('⛓️ Prime Integrationgs Started')
-  const pg = getPg()
   await upsertGlobalIntegrationProvidersFromEnv()
-  await pg.end()
   console.log('⛓️ Prime Integrations Complete')
 }