allow defining replica mode in the auth transaction

packages/api/src/repository/index.ts

@@ -60,16 +60,15 @@ export const setClaims = async (
 }
 
 interface AuthTrxOptions {
-  entityManager?: EntityManager
   uid?: string
   userRole?: string
+  replicationMode?: 'master' | 'slave'
 }
 
 export const authTrx = async <T>(
   fn: (manager: EntityManager) => Promise<T>,
   options: AuthTrxOptions = {}
 ): Promise<T> => {
-  const entityManage = options.entityManager || appDataSource.manager
   let { uid, userRole } = options
 
   // if uid and dbRole are not passed in, then get them from the claims
@@ -79,10 +78,25 @@ export const authTrx = async <T>(
     userRole = claims?.userRole
   }
 
-  return entityManage.transaction(async (tx) => {
-    await setClaims(tx, uid, userRole)
-    return fn(tx)
-  })
+  const queryRunner = appDataSource.createQueryRunner(options.replicationMode)
+
+  // lets now open a new transaction:
+  await queryRunner.startTransaction()
+
+  try {
+    await setClaims(queryRunner.manager, uid, userRole)
+    const result = await fn(queryRunner.manager)
+
+    await queryRunner.commitTransaction()
+
+    return result
+  } catch (err) {
+    await queryRunner.rollbackTransaction()
+
+    throw err
+  } finally {
+    await queryRunner.release()
+  }
 }
 
 export const getRepository = <T extends ObjectLiteral>(

packages/api/src/services/create_user.ts

@@ -8,7 +8,7 @@ import { StatusType, User } from '../entity/user'
 import { env } from '../env'
 import { SignupErrorCode } from '../generated/graphql'
 import { createPubSubClient } from '../pubsub'
-import { authTrx, getRepository } from '../repository'
+import { getRepository } from '../repository'
 import { userRepository } from '../repository/user'
 import { AuthProvider } from '../routers/auth/auth_types'
 import { analytics } from '../utils/analytics'
@@ -104,13 +104,14 @@ export const createUser = async (input: {
         })
       }
 
-      await addPopularReadsForNewUser(user.id, t)
       await createDefaultFiltersForUser(t)(user.id)
 
       return [user, profile]
     }
   )
 
+  await addPopularReadsForNewUser(user.id)
+
   const customAttributes: { source_user_id: string } = {
     source_user_id: user.sourceUserId,
   }
@@ -185,13 +186,10 @@ const validateInvite = async (
     logger.info('rejecting invite, expired', invite)
     return false
   }
-  const numMembers = await authTrx(
-    (t) =>
-      t.getRepository(GroupMembership).countBy({ invite: { id: invite.id } }),
-    {
-      entityManager,
-    }
-  )
+  const numMembers = await entityManager
+    .getRepository(GroupMembership)
+    .countBy({ invite: { id: invite.id } })
+
   if (numMembers >= invite.maxMembers) {
     logger.info('rejecting invite, too many users', { invite, numMembers })
     return false

packages/api/src/services/popular_reads.ts

@@ -1,36 +1,25 @@
 import * as httpContext from 'express-http-context2'
-import { EntityManager } from 'typeorm'
-import { appDataSource } from '../data_source'
 import { authTrx } from '../repository'
 import { libraryItemRepository } from '../repository/library_item'
 import { logger } from '../utils/logger'
 
-export const addPopularRead = async (
-  userId: string,
-  name: string,
-  entityManager?: EntityManager
-) => {
+export const addPopularRead = async (userId: string, name: string) => {
   return authTrx(
     async (tx) =>
       tx
         .withRepository(libraryItemRepository)
         .createByPopularRead(name, userId),
     {
-      entityManager,
       uid: userId,
     }
   )
 }
 
-const addPopularReads = async (
-  names: string[],
-  userId: string,
-  entityManager: EntityManager
-) => {
+const addPopularReads = async (names: string[], userId: string) => {
   // insert one by one to ensure that the order is preserved
   for (const name of names) {
     try {
-      await addPopularRead(userId, name, entityManager)
+      await addPopularRead(userId, name)
     } catch (error) {
       logger.error('failed to add popular read', error)
       continue
@@ -39,8 +28,7 @@ const addPopularReads = async (
 }
 
 export const addPopularReadsForNewUser = async (
-  userId: string,
-  em = appDataSource.manager
+  userId: string
 ): Promise<void> => {
   const defaultReads = ['omnivore_organize', 'power_read_it_later']
 
@@ -62,5 +50,5 @@ export const addPopularReadsForNewUser = async (
   // We always want this to be the top-most article in the user's
   // list. So we save it last to have the greatest saved_at
   defaultReads.push('omnivore_get_started')
-  await addPopularReads(defaultReads, userId, em)
+  await addPopularReads(defaultReads, userId)
 }

packages/api/src/services/webhook.ts

@@ -1,25 +1,21 @@
-import { ArrayContains, DeepPartial, EntityManager } from 'typeorm'
+import { ArrayContains, DeepPartial } from 'typeorm'
 import { Webhook } from '../entity/webhook'
 import { authTrx } from '../repository'
 
 export const createWebhooks = async (
   webhooks: DeepPartial<Webhook>[],
-  userId?: string,
-  entityManager?: EntityManager
+  userId?: string
 ) => {
   return authTrx((tx) => tx.getRepository(Webhook).save(webhooks), {
-    entityManager: entityManager,
     uid: userId,
   })
 }
 
 export const createWebhook = async (
   webhook: DeepPartial<Webhook>,
-  userId?: string,
-  entityManager?: EntityManager
+  userId?: string
 ) => {
   return authTrx((tx) => tx.getRepository(Webhook).save(webhook), {
-    entityManager: entityManager,
     uid: userId,
   })
 }

packages/api/test/resolvers/webhooks.test.ts

@@ -24,7 +24,7 @@ describe('Webhooks API', () => {
       .post('/local/debug/fake-user-login')
       .send({ fakeEmail: user.email })
 
-    authToken = res.body.authToken
+    authToken = res.body.authToken as string
 
     // create test webhooks
     await createWebhooks(
@@ -129,15 +129,15 @@ describe('Webhooks API', () => {
     let webhookId: string
     let enabled: boolean
 
-    beforeEach(async () => {
+    beforeEach(() => {
       query = `
         mutation {
           setWebhook(
             input: {
               id: "${webhookId}",
               url: "${webhookUrl}",
-              eventTypes: [${eventTypes}],
-              enabled: ${enabled}
+              eventTypes: [${eventTypes.toString()}],
+              enabled: ${enabled.toString()}
             }
           ) {
             ... on SetWebhookSuccess {
@@ -209,7 +209,7 @@ describe('Webhooks API', () => {
     let query: string
     let webhookId: string
 
-    beforeEach(async () => {
+    beforeEach(() => {
       query = `
         mutation {
           deleteWebhook(id: "${webhookId}") {