Merge pull request #4117 from omnivore-app/fix/use-read-replica

fix/use read replica for RLS enabled transactions

packages/api/src/jobs/ai-summarize.ts

@@ -1,13 +1,13 @@
-import { logger } from '../utils/logger'
-import { loadSummarizationChain } from 'langchain/chains'
 import { ChatOpenAI } from '@langchain/openai'
+import { loadSummarizationChain } from 'langchain/chains'
 import { RecursiveCharacterTextSplitter } from 'langchain/text_splitter'
-import { authTrx } from '../repository'
-import { libraryItemRepository } from '../repository/library_item'
-import { htmlToMarkdown } from '../utils/parser'
 import { AISummary } from '../entity/AISummary'
 import { LibraryItemState } from '../entity/library_item'
+import { authTrx } from '../repository'
+import { libraryItemRepository } from '../repository/library_item'
 import { getAISummary } from '../services/ai-summaries'
+import { logger } from '../utils/logger'
+import { htmlToMarkdown } from '../utils/parser'
 
 export interface AISummarizeJobData {
   userId: string
@@ -24,8 +24,10 @@ export const aiSummarize = async (jobData: AISummarizeJobData) => {
         tx
           .withRepository(libraryItemRepository)
           .findById(jobData.libraryItemId),
-      undefined,
-      jobData.userId
+      {
+        uid: jobData.userId,
+        replicationMode: 'replica',
+      }
     )
     if (!libraryItem || libraryItem.state !== LibraryItemState.Succeeded) {
       logger.info(
@@ -84,8 +86,9 @@ export const aiSummarize = async (jobData: AISummarizeJobData) => {
           summary: summary,
         })
       },
-      undefined,
-      jobData.userId
+      {
+        uid: jobData.userId,
+      }
     )
   } catch (err) {
     console.log('error creating summary: ', err)

packages/api/src/jobs/email/inbound_emails.ts

@@ -248,8 +248,9 @@ export const saveAttachmentJob = async (data: EmailJobData) => {
         status: UploadFileStatus.Completed,
         user: { id: user.id },
       }),
-    undefined,
-    user.id
+    {
+      uid: user.id,
+    }
   )
 
   const uploadFileDetails = await getStorageFileDetails(

packages/api/src/jobs/rss/refreshAllFeeds.ts

@@ -22,8 +22,12 @@ export const refreshAllFeeds = async (db: DataSource): Promise<boolean> => {
     refreshID: uuid(),
     startedAt: new Date().toISOString(),
   } as RSSRefreshContext
-  const subscriptionGroups = (await db.createEntityManager().query(
-    `
+  let subscriptionGroups = []
+
+  const slaveQueryRunner = db.createQueryRunner('slave')
+  try {
+    subscriptionGroups = (await slaveQueryRunner.query(
+      `
       SELECT
         url,
         ARRAY_AGG(s.id) AS "subscriptionIds",
@@ -45,8 +49,11 @@ export const refreshAllFeeds = async (db: DataSource): Promise<boolean> => {
       GROUP BY
         url
       `,
-    ['RSS', 'ACTIVE', 'following', 'ACTIVE']
-  )) as RssSubscriptionGroup[]
+      ['RSS', 'ACTIVE', 'following', 'ACTIVE']
+    )) as RssSubscriptionGroup[]
+  } finally {
+    await slaveQueryRunner.release()
+  }
 
   logger.info(`rss: checking ${subscriptionGroups.length}`, {
     refreshContext,

packages/api/src/jobs/update_db.ts

@@ -28,8 +28,9 @@ export const updateLabels = async (data: UpdateLabelsData) => {
           WHERE id = $1`,
         [data.libraryItemId]
       ),
-    undefined,
-    data.userId
+    {
+      uid: data.userId,
+    }
   )
 }
 
@@ -46,7 +47,8 @@ export const updateHighlight = async (data: UpdateHighlightData) => {
           WHERE id = $1`,
         [data.libraryItemId]
       ),
-    undefined,
-    data.userId
+    {
+      uid: data.userId,
+    }
   )
 }

packages/api/src/repository/index.ts

@@ -1,13 +1,14 @@
 import * as httpContext from 'express-http-context2'
+import { DatabaseError } from 'pg'
 import {
   EntityManager,
   EntityTarget,
   ObjectLiteral,
   QueryBuilder,
   QueryFailedError,
+  ReplicationMode,
   Repository,
 } from 'typeorm'
-import { DatabaseError } from 'pg'
 import { appDataSource } from '../data_source'
 import { Claims } from '../resolvers/types'
 import { SetClaimsRole } from '../utils/dictionary'
@@ -59,12 +60,18 @@ export const setClaims = async (
   ])
 }
 
+interface AuthTrxOptions {
+  uid?: string
+  userRole?: string
+  replicationMode?: 'primary' | 'replica'
+}
+
 export const authTrx = async <T>(
   fn: (manager: EntityManager) => Promise<T>,
-  em = appDataSource.manager,
-  uid?: string,
-  userRole?: string
+  options: AuthTrxOptions = {}
 ): Promise<T> => {
+  let { uid, userRole } = options
+
   // if uid and dbRole are not passed in, then get them from the claims
   if (!uid && !userRole) {
     const claims: Claims | undefined = httpContext.get('claims')
@@ -72,10 +79,34 @@ export const authTrx = async <T>(
     userRole = claims?.userRole
   }
 
-  return em.transaction(async (tx) => {
-    await setClaims(tx, uid, userRole)
-    return fn(tx)
-  })
+  const replicationModes: Record<'primary' | 'replica', ReplicationMode> = {
+    primary: 'master',
+    replica: 'slave',
+  }
+
+  const replicationMode = options.replicationMode
+    ? replicationModes[options.replicationMode]
+    : undefined
+
+  const queryRunner = appDataSource.createQueryRunner(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/resolvers/article/index.ts

@@ -63,7 +63,7 @@ import {
   UpdatesSinceError,
   UpdatesSinceSuccess,
 } from '../../generated/graphql'
-import { getColumns } from '../../repository'
+import { authTrx, getColumns } from '../../repository'
 import { getInternalLabelWithColor } from '../../repository/label'
 import { libraryItemRepository } from '../../repository/library_item'
 import { userRepository } from '../../repository/user'
@@ -376,7 +376,7 @@ export const getArticleResolver = authorized<
   Merge<ArticleSuccess, { article: LibraryItem }>,
   ArticleError,
   QueryArticleArgs
->(async (_obj, { slug, format }, { authTrx, uid, log }, info) => {
+>(async (_obj, { slug, format }, { uid, log }, info) => {
   try {
     const selectColumns = getColumns(libraryItemRepository)
     const includeOriginalHtml =
@@ -386,36 +386,44 @@ export const getArticleResolver = authorized<
       selectColumns.splice(selectColumns.indexOf('originalContent'), 1)
     }
 
-    const libraryItem = await authTrx((tx) => {
-      const qb = tx
-        .createQueryBuilder(LibraryItem, 'libraryItem')
-        .select(selectColumns.map((column) => `libraryItem.${column}`))
-        .leftJoinAndSelect('libraryItem.labels', 'labels')
-        .leftJoinAndSelect('libraryItem.highlights', 'highlights')
-        .leftJoinAndSelect('highlights.labels', 'highlights_labels')
-        .leftJoinAndSelect('highlights.user', 'highlights_user')
-        .leftJoinAndSelect('highlights_user.profile', 'highlights_user_profile')
-        .leftJoinAndSelect('libraryItem.uploadFile', 'uploadFile')
-        .leftJoinAndSelect('libraryItem.recommendations', 'recommendations')
-        .leftJoinAndSelect('recommendations.group', 'recommendations_group')
-        .leftJoinAndSelect(
-          'recommendations.recommender',
-          'recommendations_recommender'
-        )
-        .leftJoinAndSelect(
-          'recommendations_recommender.profile',
-          'recommendations_recommender_profile'
-        )
-        .where('libraryItem.user_id = :uid', { uid })
+    const libraryItem = await authTrx(
+      (tx) => {
+        const qb = tx
+          .createQueryBuilder(LibraryItem, 'libraryItem')
+          .select(selectColumns.map((column) => `libraryItem.${column}`))
+          .leftJoinAndSelect('libraryItem.labels', 'labels')
+          .leftJoinAndSelect('libraryItem.highlights', 'highlights')
+          .leftJoinAndSelect('highlights.labels', 'highlights_labels')
+          .leftJoinAndSelect('highlights.user', 'highlights_user')
+          .leftJoinAndSelect(
+            'highlights_user.profile',
+            'highlights_user_profile'
+          )
+          .leftJoinAndSelect('libraryItem.uploadFile', 'uploadFile')
+          .leftJoinAndSelect('libraryItem.recommendations', 'recommendations')
+          .leftJoinAndSelect('recommendations.group', 'recommendations_group')
+          .leftJoinAndSelect(
+            'recommendations.recommender',
+            'recommendations_recommender'
+          )
+          .leftJoinAndSelect(
+            'recommendations_recommender.profile',
+            'recommendations_recommender_profile'
+          )
+          .where('libraryItem.user_id = :uid', { uid })
 
-      // We allow the backend to use the ID instead of a slug to fetch the article
-      // query against id if slug is a uuid
-      slug.match(/^[0-9a-f]{8}-([0-9a-f]{4}-){3}[0-9a-f]{12}$/i)
-        ? qb.andWhere('libraryItem.id = :id', { id: slug })
-        : qb.andWhere('libraryItem.slug = :slug', { slug })
+        // We allow the backend to use the ID instead of a slug to fetch the article
+        // query against id if slug is a uuid
+        slug.match(/^[0-9a-f]{8}-([0-9a-f]{4}-){3}[0-9a-f]{12}$/i)
+          ? qb.andWhere('libraryItem.id = :id', { id: slug })
+          : qb.andWhere('libraryItem.slug = :slug', { slug })
 
-      return qb.andWhere('libraryItem.deleted_at IS NULL').getOne()
-    })
+        return qb.andWhere('libraryItem.deleted_at IS NULL').getOne()
+      },
+      {
+        replicationMode: 'replica',
+      }
+    )
 
     if (!libraryItem) {
       return { errorCodes: [ArticleErrorCode.NotFound] }
@@ -499,7 +507,7 @@ export const saveArticleReadingProgressResolver = authorized<
         force,
       },
     },
-    { authTrx, pubsub, uid, dataSources }
+    { pubsub, uid, dataSources }
   ) => {
     if (
       readingProgressPercent < 0 ||
@@ -515,13 +523,17 @@ export const saveArticleReadingProgressResolver = authorized<
     // We don't need to update the values of reading progress here
     // because the function resolver will handle that for us when
     // it resolves the properties of the Article object
-    let updatedItem = await authTrx((tx) =>
-      tx.getRepository(LibraryItem).findOne({
-        where: {
-          id,
-        },
-        relations: ['user'],
-      })
+    let updatedItem = await authTrx(
+      (tx) =>
+        tx.getRepository(LibraryItem).findOne({
+          where: {
+            id,
+          },
+          relations: ['user'],
+        }),
+      {
+        replicationMode: 'replica',
+      }
     )
     if (!updatedItem) {
       return {
@@ -838,7 +850,7 @@ export const moveToFolderResolver = authorized<
   MoveToFolderSuccess,
   MoveToFolderError,
   MutationMoveToFolderArgs
->(async (_, { id, folder }, { authTrx, log, pubsub, uid }) => {
+>(async (_, { id, folder }, { log, pubsub, uid }) => {
   analytics.capture({
     distinctId: uid,
     event: 'move_to_folder',
@@ -848,13 +860,17 @@ export const moveToFolderResolver = authorized<
     },
   })
 
-  const item = await authTrx((tx) =>
-    tx.getRepository(LibraryItem).findOne({
-      where: {
-        id,
-      },
-      relations: ['user'],
-    })
+  const item = await authTrx(
+    (tx) =>
+      tx.getRepository(LibraryItem).findOne({
+        where: {
+          id,
+        },
+        relations: ['user'],
+      }),
+    {
+      replicationMode: 'replica',
+    }
   )
 
   if (!item) {
@@ -913,7 +929,7 @@ export const fetchContentResolver = authorized<
   FetchContentSuccess,
   FetchContentError,
   MutationFetchContentArgs
->(async (_, { id }, { authTrx, uid, log, pubsub }) => {
+>(async (_, { id }, { uid, log, pubsub }) => {
   analytics.capture({
     distinctId: uid,
     event: 'fetch_content',
@@ -922,13 +938,17 @@ export const fetchContentResolver = authorized<
     },
   })
 
-  const item = await authTrx((tx) =>
-    tx.getRepository(LibraryItem).findOne({
-      where: {
-        id,
-      },
-      relations: ['user'],
-    })
+  const item = await authTrx(
+    (tx) =>
+      tx.getRepository(LibraryItem).findOne({
+        where: {
+          id,
+        },
+        relations: ['user'],
+      }),
+    {
+      replicationMode: 'replica',
+    }
   )
   if (!item) {
     return {

packages/api/src/resolvers/highlight/index.ts

@@ -32,6 +32,7 @@ import {
   UpdateHighlightErrorCode,
   UpdateHighlightSuccess,
 } from '../../generated/graphql'
+import { authTrx } from '../../repository'
 import { highlightRepository } from '../../repository/highlight'
 import {
   createHighlight,
@@ -87,7 +88,7 @@ export const mergeHighlightResolver = authorized<
   Merge<MergeHighlightSuccess, { highlight: HighlightEntity }>,
   MergeHighlightError,
   MutationMergeHighlightArgs
->(async (_, { input }, { authTrx, log, pubsub, uid }) => {
+>(async (_, { input }, { log, pubsub, uid }) => {
   const { overlapHighlightIdList, ...newHighlightInput } = input
 
   /* Compute merged annotation form the order of highlights appearing on page */
@@ -96,10 +97,14 @@ export const mergeHighlightResolver = authorized<
   const mergedColors: string[] = []
 
   try {
-    const existingHighlights = await authTrx((tx) =>
-      tx
-        .withRepository(highlightRepository)
-        .findByLibraryItemId(input.articleId, uid)
+    const existingHighlights = await authTrx(
+      (tx) =>
+        tx
+          .withRepository(highlightRepository)
+          .findByLibraryItemId(input.articleId, uid),
+      {
+        replicationMode: 'replica',
+      }
     )
 
     existingHighlights.forEach((highlight) => {

packages/api/src/resolvers/labels/index.ts

@@ -28,6 +28,7 @@ import {
   UpdateLabelErrorCode,
   UpdateLabelSuccess,
 } from '../../generated/graphql'
+import { authTrx } from '../../repository'
 import { labelRepository } from '../../repository/label'
 import { userRepository } from '../../repository/user'
 import { findHighlightById } from '../../services/highlights'
@@ -43,7 +44,7 @@ import { analytics } from '../../utils/analytics'
 import { authorized } from '../../utils/gql-utils'
 
 export const labelsResolver = authorized<LabelsSuccess, LabelsError>(
-  async (_obj, _params, { authTrx, log, uid }) => {
+  async (_obj, _params, { log, uid }) => {
     try {
       const user = await userRepository.findById(uid)
       if (!user) {
@@ -52,16 +53,21 @@ export const labelsResolver = authorized<LabelsSuccess, LabelsError>(
         }
       }
 
-      const labels = await authTrx(async (tx) => {
-        return tx.withRepository(labelRepository).find({
-          where: {
-            user: { id: uid },
-          },
-          order: {
-            name: 'ASC',
-          },
-        })
-      })
+      const labels = await authTrx(
+        async (tx) => {
+          return tx.withRepository(labelRepository).find({
+            where: {
+              user: { id: uid },
+            },
+            order: {
+              name: 'ASC',
+            },
+          })
+        },
+        {
+          replicationMode: 'replica',
+        }
+      )
 
       analytics.capture({
         distinctId: uid,

packages/api/src/routers/page_router.ts

@@ -82,8 +82,9 @@ export function pageRouter() {
           status: UploadFileStatus.Initialized,
           contentType: 'application/pdf',
         }),
-      undefined,
-      claims.uid
+      {
+        uid: claims.uid,
+      }
     )
 
     const uploadFilePathName = generateUploadFilePathName(

packages/api/src/routers/svc/email_attachment.ts

@@ -67,8 +67,9 @@ export function emailAttachmentRouter() {
             contentType,
             user: { id: user.id },
           }),
-        undefined,
-        user.id
+        {
+          uid: user.id,
+        }
       )
 
       if (uploadFileData.id) {

packages/api/src/routers/svc/webhooks.ts

@@ -47,8 +47,9 @@ export function webhooksServiceRouter() {
             .andWhere(':eventType = ANY(event_types)', { eventType })
             .andWhere('enabled = true')
             .getMany(),
-        undefined,
-        userId
+        {
+          uid: userId,
+        }
       )
 
       if (webhooks.length <= 0) {

packages/api/src/routers/text_to_speech.ts

@@ -53,8 +53,9 @@ export function textToSpeechRouter() {
           state,
         })
       },
-      undefined,
-      userId
+      {
+        uid: userId,
+      }
     )
 
     res.send('OK')

packages/api/src/services/ai-summaries.ts

@@ -27,8 +27,9 @@ export const getAISummary = async (data: {
         })
       }
     },
-    undefined,
-    data.userId
+    {
+      uid: data.userId,
+    }
   )
   return aiSummary ?? undefined
 }

packages/api/src/services/api_key.ts

@@ -27,8 +27,9 @@ export const findApiKeys = async (
           createdAt: 'DESC',
         },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -36,9 +37,7 @@ export const deleteApiKey = async (
   criteria: string[] | FindOptionsWhere<ApiKey>,
   userId: string
 ) => {
-  return authTrx(
-    async (t) => t.getRepository(ApiKey).delete(criteria),
-    undefined,
-    userId
-  )
+  return authTrx(async (t) => t.getRepository(ApiKey).delete(criteria), {
+    uid: userId,
+  })
 }

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,11 +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/explain.ts

@@ -1,9 +1,9 @@
-import { OpenAI } from '@langchain/openai'
 import { PromptTemplate } from '@langchain/core/prompts'
+import { OpenAI } from '@langchain/openai'
 import { authTrx } from '../repository'
 import { libraryItemRepository } from '../repository/library_item'
-import { htmlToMarkdown } from '../utils/parser'
 import { OPENAI_MODEL } from '../utils/ai'
+import { htmlToMarkdown } from '../utils/parser'
 
 export const explainText = async (
   userId: string,
@@ -20,8 +20,10 @@ export const explainText = async (
   const libraryItem = await authTrx(
     async (tx) =>
       tx.withRepository(libraryItemRepository).findById(libraryItemId),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 
   if (!libraryItem) {

packages/api/src/services/features.ts

@@ -2,13 +2,12 @@ import * as jwt from 'jsonwebtoken'
 import { DeepPartial, FindOptionsWhere, IsNull, Not } from 'typeorm'
 import { appDataSource } from '../data_source'
 import { Feature } from '../entity/feature'
+import { LibraryItem } from '../entity/library_item'
+import { Subscription, SubscriptionStatus } from '../entity/subscription'
 import { env } from '../env'
+import { OptInFeatureErrorCode } from '../generated/graphql'
 import { authTrx, getRepository } from '../repository'
 import { logger } from '../utils/logger'
-import { OptInFeatureErrorCode } from '../generated/graphql'
-import { Subscription, SubscriptionStatus } from '../entity/subscription'
-import { libraryItemRepository } from '../repository/library_item'
-import { LibraryItem } from '../entity/library_item'
 
 const MAX_ULTRA_REALISTIC_USERS = 1500
 const MAX_YOUTUBE_TRANSCRIPT_USERS = 500
@@ -182,8 +181,10 @@ export const userDigestEligible = async (uid: string): Promise<boolean> => {
         where: { user: { id: uid }, status: SubscriptionStatus.Active },
       })
     },
-    undefined,
-    uid
+    {
+      uid,
+      replicationMode: 'replica',
+    }
   )
 
   const libraryItemsCount = await authTrx(
@@ -192,8 +193,10 @@ export const userDigestEligible = async (uid: string): Promise<boolean> => {
         where: { user: { id: uid } },
       })
     },
-    undefined,
-    uid
+    {
+      uid,
+      replicationMode: 'replica',
+    }
   )
 
   return subscriptionsCount >= 2 && libraryItemsCount >= 10

packages/api/src/services/highlights.ts

@@ -23,10 +23,14 @@ export type HighlightEvent = Merge<
 export const batchGetHighlightsFromLibraryItemIds = async (
   libraryItemIds: readonly string[]
 ): Promise<Highlight[][]> => {
-  const highlights = await authTrx(async (tx) =>
-    tx.getRepository(Highlight).find({
-      where: { libraryItem: { id: In(libraryItemIds as string[]) } },
-    })
+  const highlights = await authTrx(
+    async (tx) =>
+      tx.getRepository(Highlight).find({
+        where: { libraryItem: { id: In(libraryItemIds as string[]) } },
+      }),
+    {
+      replicationMode: 'replica',
+    }
   )
 
   return libraryItemIds.map((libraryItemId) =>
@@ -50,8 +54,9 @@ export const createHighlights = async (
   return authTrx(
     async (tx) =>
       tx.withRepository(highlightRepository).createAndSaves(highlights),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -73,8 +78,9 @@ export const createHighlight = async (
         },
       })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   const data = deepDelete(newHighlight, columnsToDelete)
@@ -221,8 +227,9 @@ export const deleteHighlightById = async (
       await highlightRepo.delete(highlightId)
       return highlight
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   await enqueueUpdateHighlight({
@@ -239,8 +246,9 @@ export const deleteHighlightsByIds = async (
 ) => {
   await authTrx(
     async (tx) => tx.getRepository(Highlight).delete(highlightIds),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -256,8 +264,10 @@ export const findHighlightById = async (
         user: { id: userId },
       })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -274,8 +284,10 @@ export const findHighlightsByLibraryItemId = async (
           labels: true,
         },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -322,7 +334,9 @@ export const searchHighlights = async (
 
       return queryBuilder.getMany()
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }

packages/api/src/services/home.ts

@@ -4,12 +4,16 @@ import { authTrx } from '../repository'
 export const batchGetPublicItems = async (
   ids: readonly string[]
 ): Promise<Array<PublicItem | undefined>> => {
-  const publicItems = await authTrx(async (tx) =>
-    tx
-      .getRepository(PublicItem)
-      .createQueryBuilder('public_item')
-      .where('public_item.id IN (:...ids)', { ids })
-      .getMany()
+  const publicItems = await authTrx(
+    async (tx) =>
+      tx
+        .getRepository(PublicItem)
+        .createQueryBuilder('public_item')
+        .where('public_item.id IN (:...ids)', { ids })
+        .getMany(),
+    {
+      replicationMode: 'replica',
+    }
   )
 
   return ids.map((id) => publicItems.find((pi) => pi.id === id))
@@ -48,7 +52,9 @@ export const findUnseenPublicItems = async (
         .take(options.limit)
         .skip(options.offset)
         .getMany(),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }

packages/api/src/services/integrations/index.ts

@@ -27,11 +27,9 @@ export const deleteIntegrations = async (
   userId: string,
   criteria: string[] | FindOptionsWhere<Integration>
 ) => {
-  return authTrx(
-    async (t) => t.getRepository(Integration).delete(criteria),
-    undefined,
-    userId
-  )
+  return authTrx(async (t) => t.getRepository(Integration).delete(criteria), {
+    uid: userId,
+  })
 }
 
 export const removeIntegration = async (
@@ -40,8 +38,9 @@ export const removeIntegration = async (
 ) => {
   return authTrx(
     async (t) => t.getRepository(Integration).remove(integration),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -55,8 +54,9 @@ export const findIntegration = async (
         ...where,
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -71,8 +71,9 @@ export const findIntegrationByName = async (name: string, userId: string) => {
         })
         .andWhere('LOWER(name) = LOWER(:name)', { name }) // case insensitive
         .getOne(),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -86,8 +87,9 @@ export const findIntegrations = async (
         ...where,
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -101,8 +103,9 @@ export const saveIntegration = async (
       const newIntegration = await repo.save(integration)
       return repo.findOneByOrFail({ id: newIntegration.id })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -113,7 +116,8 @@ export const updateIntegration = async (
 ) => {
   return authTrx(
     async (t) => t.getRepository(Integration).update(id, integration),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }

packages/api/src/services/labels.ts

@@ -25,11 +25,15 @@ export type LabelEvent = Merge<
 export const batchGetLabelsFromLibraryItemIds = async (
   libraryItemIds: readonly string[]
 ): Promise<Label[][]> => {
-  const labels = await authTrx(async (tx) =>
-    tx.getRepository(EntityLabel).find({
-      where: { libraryItemId: In(libraryItemIds as string[]) },
-      relations: ['label'],
-    })
+  const labels = await authTrx(
+    async (tx) =>
+      tx.getRepository(EntityLabel).find({
+        where: { libraryItemId: In(libraryItemIds as string[]) },
+        relations: ['label'],
+      }),
+    {
+      replicationMode: 'replica',
+    }
   )
 
   return libraryItemIds.map((libraryItemId) =>
@@ -42,11 +46,15 @@ export const batchGetLabelsFromLibraryItemIds = async (
 export const batchGetLabelsFromHighlightIds = async (
   highlightIds: readonly string[]
 ): Promise<Label[][]> => {
-  const labels = await authTrx(async (tx) =>
-    tx.getRepository(EntityLabel).find({
-      where: { highlightId: In(highlightIds as string[]) },
-      relations: ['label'],
-    })
+  const labels = await authTrx(
+    async (tx) =>
+      tx.getRepository(EntityLabel).find({
+        where: { highlightId: In(highlightIds as string[]) },
+        relations: ['label'],
+      }),
+    {
+      replicationMode: 'replica',
+    }
   )
 
   return highlightIds.map((highlightId) =>
@@ -72,8 +80,9 @@ export const findOrCreateLabels = async (
         user: { id: userId },
       })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -156,8 +165,9 @@ export const saveLabelsInLibraryItem = async (
         }))
       )
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   if (source === 'user') {
@@ -209,8 +219,9 @@ export const addLabelsToLibraryItem = async (
         [libraryItemId, source, labelIds]
       )
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   // update labels in library item
@@ -250,8 +261,10 @@ export const findLabelsByIds = async (
         user: { id: userId },
       })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -263,8 +276,9 @@ export const createLabel = async (
   return authTrx(
     (t) =>
       t.withRepository(labelRepository).createLabel({ name, color }, userId),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -274,8 +288,9 @@ export const deleteLabels = async (
 ) => {
   return authTrx(
     async (t) => t.withRepository(labelRepository).delete(criteria),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -311,8 +326,9 @@ export const updateLabel = async (
 
       return repo.findOneByOrFail({ id })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   const libraryItemIds = await findLibraryItemIdsByLabelId(id, userId)
@@ -333,8 +349,10 @@ export const findLabelsByUserId = async (userId: string): Promise<Label[]> => {
         where: { user: { id: userId } },
         order: { position: 'ASC' },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -344,8 +362,10 @@ export const findLabelById = async (id: string, userId: string) => {
       tx
         .withRepository(labelRepository)
         .findOneBy({ id, user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -365,7 +385,8 @@ export const findLabelsByLibraryItemId = async (
         source: el.source,
       }))
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }

packages/api/src/services/library_item.ts

@@ -138,13 +138,17 @@ export const batchGetLibraryItems = async (ids: readonly string[]) => {
   const select = getColumns(libraryItemRepository).filter(
     (select) => ['originalContent', 'readableContent'].indexOf(select) === -1
   )
-  const items = await authTrx(async (tx) =>
-    tx.getRepository(LibraryItem).find({
-      select,
-      where: {
-        id: In(ids as string[]),
-      },
-    })
+  const items = await authTrx(
+    async (tx) =>
+      tx.getRepository(LibraryItem).find({
+        select,
+        where: {
+          id: In(ids as string[]),
+        },
+      }),
+    {
+      replicationMode: 'replica',
+    }
   )
 
   return ids.map((id) => items.find((item) => item.id === id) || undefined)
@@ -707,8 +711,10 @@ export const createSearchQueryBuilder = (
 export const countLibraryItems = async (args: SearchArgs, userId: string) => {
   return authTrx(
     async (tx) => createSearchQueryBuilder(args, userId, tx).getCount(),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -729,8 +735,10 @@ export const searchLibraryItems = async (
         .skip(from)
         .take(size)
         .getMany(),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -774,8 +782,10 @@ export const findRecentLibraryItems = async (
         .take(limit)
         .skip(offset)
         .getMany(),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -798,8 +808,10 @@ export const findLibraryItemsByIds = async (
         .select(selectColumns)
         .where('library_item.id IN (:...ids)', { ids })
         .getMany(),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -826,8 +838,10 @@ export const findLibraryItemById = async (
         where: { id },
         relations: options?.relations,
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -848,8 +862,10 @@ export const findLibraryItemByUrl = async (
         .where('library_item.user_id = :userId', { userId })
         .andWhere('md5(library_item.original_url) = md5(:url)', { url })
         .getOne(),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -889,8 +905,9 @@ export const softDeleteLibraryItem = async (
 
       return itemRepo.findOneByOrFail({ id })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   await pubsub.entityDeleted(EntityType.ITEM, id, userId)
@@ -924,8 +941,9 @@ export const updateLibraryItem = async (
 
       return itemRepo.findOneByOrFail({ id })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   if (skipPubSub || libraryItem.state === LibraryItemState.Processing) {
@@ -998,8 +1016,9 @@ export const updateLibraryItemReadingProgress = async (
       `,
         [id, topPercent, bottomPercent, anchorIndex]
       ),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )) as [LibraryItem[], number]
   if (result[1] === 0) {
     return null
@@ -1017,8 +1036,9 @@ export const createLibraryItems = async (
 ): Promise<LibraryItem[]> => {
   return authTrx(
     async (tx) => tx.withRepository(libraryItemRepository).save(libraryItems),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -1094,8 +1114,9 @@ export const createOrUpdateLibraryItem = async (
       // create or update library item
       return repo.upsertLibraryItemById(libraryItem)
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   // set recently saved item in redis if redis is enabled
@@ -1142,17 +1163,21 @@ export const findLibraryItemsByPrefix = async (
 ): Promise<LibraryItem[]> => {
   const prefixWildcard = `${prefix}%`
 
-  return authTrx(async (tx) =>
-    tx
-      .createQueryBuilder(LibraryItem, 'library_item')
-      .where('library_item.user_id = :userId', { userId })
-      .andWhere(
-        '(library_item.title ILIKE :prefix OR library_item.site_name ILIKE :prefix)',
-        { prefix: prefixWildcard }
-      )
-      .orderBy('library_item.savedAt', 'DESC')
-      .limit(limit)
-      .getMany()
+  return authTrx(
+    async (tx) =>
+      tx
+        .createQueryBuilder(LibraryItem, 'library_item')
+        .where('library_item.user_id = :userId', { userId })
+        .andWhere(
+          '(library_item.title ILIKE :prefix OR library_item.site_name ILIKE :prefix)',
+          { prefix: prefixWildcard }
+        )
+        .orderBy('library_item.savedAt', 'DESC')
+        .limit(limit)
+        .getMany(),
+    {
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -1171,8 +1196,10 @@ export const countBySavedAt = async (
           endDate,
         })
         .getCount(),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 
@@ -1253,8 +1280,9 @@ export const batchUpdateLibraryItems = async (
 
       const libraryItemIds = await authTrx(
         async (tx) => getLibraryItemIds(userId, tx),
-        undefined,
-        userId
+        {
+          uid: userId,
+        }
       )
       // add labels to library items
       for (const libraryItemId of libraryItemIds) {
@@ -1266,8 +1294,9 @@ export const batchUpdateLibraryItems = async (
     case BulkActionType.MarkAsRead: {
       const libraryItemIds = await authTrx(
         async (tx) => getLibraryItemIds(userId, tx),
-        undefined,
-        userId
+        {
+          uid: userId,
+        }
       )
       // update reading progress for library items
       for (const libraryItemId of libraryItemIds) {
@@ -1301,16 +1330,18 @@ export const batchUpdateLibraryItems = async (
       const libraryItemIds = await getLibraryItemIds(userId, tx, true)
       await tx.getRepository(LibraryItem).update(libraryItemIds, values)
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
 export const deleteLibraryItemById = async (id: string, userId?: string) => {
   return authTrx(
     async (tx) => tx.withRepository(libraryItemRepository).delete(id),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -1321,8 +1352,9 @@ export const deleteLibraryItems = async (
   return authTrx(
     async (tx) =>
       tx.withRepository(libraryItemRepository).delete(items.map((i) => i.id)),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -1332,8 +1364,9 @@ export const deleteLibraryItemByUrl = async (url: string, userId: string) => {
       tx
         .withRepository(libraryItemRepository)
         .delete({ originalUrl: url, user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -1343,8 +1376,9 @@ export const deleteLibraryItemsByUserId = async (userId: string) => {
       tx.withRepository(libraryItemRepository).delete({
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -1403,8 +1437,10 @@ export const findLibraryItemIdsByLabelId = async (
 
       return result.map((r) => r.library_item_id)
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }
 

packages/api/src/services/popular_reads.ts

@@ -1,34 +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,
-    userId
+    {
+      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
@@ -37,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']
 
@@ -60,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/received_emails.ts

@@ -23,8 +23,9 @@ export const saveReceivedEmail = async (
         user: { id: userId },
         replyTo,
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -38,16 +39,18 @@ export const updateReceivedEmail = async (
       t
         .getRepository(ReceivedEmail)
         .update({ id, user: { id: userId } }, { type }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
 export const deleteReceivedEmail = async (id: string, userId: string) => {
   return authTrx(
     (t) => t.getRepository(ReceivedEmail).delete({ id, user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -55,7 +58,9 @@ export const findReceivedEmailById = async (id: string, userId: string) => {
   return authTrx(
     (t) =>
       t.getRepository(ReceivedEmail).findOneBy({ id, user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+      replicationMode: 'replica',
+    }
   )
 }

packages/api/src/services/recommendation.ts

@@ -116,8 +116,9 @@ export const createRecommendation = async (
 ) => {
   return authTrx(
     async (tx) => tx.getRepository(Recommendation).save(recommendation),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -134,7 +135,8 @@ export const findRecommendationsByLibraryItemId = async (
           recommender: true,
         },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }

packages/api/src/services/rules.ts

@@ -28,8 +28,9 @@ export const createRule = async (
         ...rule,
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -41,16 +42,18 @@ export const deleteRule = async (id: string, userId: string) => {
       await repo.delete(id)
       return rule
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
 export const deleteRules = async (userId: string) => {
   return authTrx(
     (t) => t.getRepository(Rule).delete({ user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -72,7 +75,8 @@ export const markRuleAsFailed = async (id: string, userId: string) => {
       t.getRepository(Rule).update(id, {
         failedAt: new Date(),
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }

packages/api/src/services/update_pdf_content.ts

@@ -43,8 +43,9 @@ export const updateContentForFileItem = async (msg: UpdateContentMessage) => {
         .innerJoinAndSelect('item.uploadFile', 'file')
         .where('file.id = :fileId', { fileId })
         .getOne(),
-    undefined,
-    uploadFile.user.id
+    {
+      uid: uploadFile.user.id,
+    }
   )
   if (!libraryItem) {
     logger.info(`No upload file found for id: ${fileId}`)

packages/api/src/services/upload_file.ts

@@ -59,8 +59,9 @@ export const setFileUploadComplete = async (id: string, userId?: string) => {
 
       return repo.findOneByOrFail({ id })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 

packages/api/src/services/user.ts

@@ -17,17 +17,16 @@ export const deleteUser = async (userId: string) => {
     async (t) => {
       await t.withRepository(userRepository).delete(userId)
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
 export const updateUser = async (userId: string, update: Partial<User>) => {
-  return authTrx(
-    async (t) => t.getRepository(User).update(userId, update),
-    undefined,
-    userId
-  )
+  return authTrx(async (t) => t.getRepository(User).update(userId, update), {
+    uid: userId,
+  })
 }
 
 export const softDeleteUser = async (userId: string) => {
@@ -51,8 +50,9 @@ export const softDeleteUser = async (userId: string) => {
         sourceUserId: `deleted_user_${userId}`,
       })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -67,21 +67,15 @@ export const findUsersByIds = async (ids: string[]): Promise<User[]> => {
 export const deleteUsers = async (
   criteria: FindOptionsWhere<User> | string[]
 ) => {
-  return authTrx(
-    async (t) => t.getRepository(User).delete(criteria),
-    undefined,
-    undefined,
-    SetClaimsRole.ADMIN
-  )
+  return authTrx(async (t) => t.getRepository(User).delete(criteria), {
+    userRole: SetClaimsRole.ADMIN,
+  })
 }
 
 export const createUsers = async (users: DeepPartial<User>[]) => {
-  return authTrx(
-    async (t) => t.getRepository(User).save(users),
-    undefined,
-    undefined,
-    SetClaimsRole.ADMIN
-  )
+  return authTrx(async (t) => t.getRepository(User).save(users), {
+    userRole: SetClaimsRole.ADMIN,
+  })
 }
 
 export const batchDelete = async (criteria: FindOptionsWhere<User>) => {
@@ -95,7 +89,7 @@ export const batchDelete = async (criteria: FindOptionsWhere<User>) => {
   const sql = `
   -- Set batch size
   DO $$
-  DECLARE 
+  DECLARE
       batch_size INT := ${batchSize};
       user_ids UUID[];
   BEGIN
@@ -103,7 +97,7 @@ export const batchDelete = async (criteria: FindOptionsWhere<User>) => {
       FOR i IN 0..CEIL((${userCountSql}) * 1.0 / batch_size) - 1 LOOP
           -- GET batch of user ids
           ${userSubQuery} LIMIT batch_size;
-          
+
           -- Loop through batches of items
           FOR j IN 0..CEIL((SELECT COUNT(1) FROM omnivore.library_item WHERE user_id = ANY(user_ids)) * 1.0 / batch_size) - 1 LOOP
               -- Delete batch of items
@@ -122,12 +116,9 @@ export const batchDelete = async (criteria: FindOptionsWhere<User>) => {
   END $$
   `
 
-  return authTrx(
-    async (t) => t.query(sql),
-    undefined,
-    undefined,
-    SetClaimsRole.ADMIN
-  )
+  return authTrx(async (t) => t.query(sql), {
+    userRole: SetClaimsRole.ADMIN,
+  })
 }
 
 export const sendPushNotifications = async (
@@ -160,7 +151,8 @@ export const findUserAndPersonalization = async (id: string) => {
           userPersonalization: true,
         },
       }),
-    undefined,
-    id
+    {
+      uid: id,
+    }
   )
 }

packages/api/src/services/user_device_tokens.ts

@@ -11,8 +11,9 @@ export const findDeviceTokenById = async (
   return authTrx(
     (t) =>
       t.getRepository(UserDeviceToken).findOneBy({ id, user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -25,8 +26,9 @@ export const findDeviceTokenByToken = async (
       t
         .getRepository(UserDeviceToken)
         .findOneBy({ token, user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -38,8 +40,9 @@ export const findDeviceTokensByUserId = async (
       t.getRepository(UserDeviceToken).findBy({
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -61,8 +64,9 @@ export const createDeviceToken = async (
         token,
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -95,7 +99,8 @@ export const deleteDeviceTokens = async (
     async (t) => {
       await t.getRepository(UserDeviceToken).delete(criteria)
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }

packages/api/src/services/user_personalization.ts

@@ -12,8 +12,9 @@ export const findUserPersonalization = async (userId: string) => {
       t.getRepository(UserPersonalization).findOneBy({
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -23,8 +24,9 @@ export const deleteUserPersonalization = async (userId: string) => {
       t.getRepository(UserPersonalization).delete({
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -34,8 +36,9 @@ export const saveUserPersonalization = async (
 ) => {
   return authTrx(
     (t) => t.getRepository(UserPersonalization).save(userPersonalization),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -45,8 +48,9 @@ export const getShortcuts = async (userId: string): Promise<Shortcut[]> => {
       t.getRepository(UserPersonalization).findOneBy({
         user: { id: userId },
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
   if (personalization?.shortcuts) {
     return personalization?.shortcuts as Shortcut[]
@@ -67,8 +71,9 @@ export const resetShortcuts = async (userId: string): Promise<boolean> => {
         })
         .execute()
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
   if (!result) {
     throw Error('Could not update shortcuts')
@@ -90,8 +95,9 @@ export const setShortcuts = async (
           shortcuts: shortcuts,
         }
       ),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
   if (!result.affected || result.affected < 1) {
     throw Error('Could not update shortcuts')

packages/api/src/services/webhook.ts

@@ -1,36 +1,31 @@
-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,
-    userId
-  )
+  return authTrx((tx) => tx.getRepository(Webhook).save(webhooks), {
+    uid: userId,
+  })
 }
 
 export const createWebhook = async (
   webhook: DeepPartial<Webhook>,
-  userId?: string,
-  entityManager?: EntityManager
+  userId?: string
 ) => {
-  return authTrx(
-    (tx) => tx.getRepository(Webhook).save(webhook),
-    entityManager,
-    userId
-  )
+  return authTrx((tx) => tx.getRepository(Webhook).save(webhook), {
+    uid: userId,
+  })
 }
 
 export const findWebhooks = async (userId: string) => {
   return authTrx(
     (tx) => tx.getRepository(Webhook).findBy({ user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -45,16 +40,18 @@ export const findWebhooksByEventType = async (
         enabled: true,
         eventTypes: ArrayContains([eventType]),
       }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
 export const findWebhookById = async (id: string, userId: string) => {
   return authTrx(
     (tx) => tx.getRepository(Webhook).findOneBy({ id, user: { id: userId } }),
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }
 
@@ -66,7 +63,8 @@ export const deleteWebhook = async (id: string, userId: string) => {
       await repo.delete(id)
       return webhook
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 }

packages/api/src/utils/logger.ts

@@ -36,7 +36,7 @@ export class CustomTypeOrmLogger
 
   logQuery(query: string, parameters?: any[], queryRunner?: QueryRunner) {
     this.logger.info(query, {
-      isReplicated: queryRunner?.connection?.driver?.isReplicated,
+      replicationMode: queryRunner?.getReplicationMode(),
       parameters,
     })
   }

packages/api/test/db.ts

@@ -158,8 +158,9 @@ export const saveLabelsInLibraryItem = async (
         }))
       )
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   // update labels in library item
@@ -184,8 +185,9 @@ export const createHighlight = async (
         },
       })
     },
-    undefined,
-    userId
+    {
+      uid: userId,
+    }
   )
 
   const job = await enqueueUpdateHighlight({

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}") {

packages/api/test/services/create_user.test.ts

@@ -24,8 +24,9 @@ describe('create user', () => {
       const user = await createTestUser('filter_user')
       const filters = await authTrx(
         (t) => t.getRepository(Filter).findBy({ user: { id: user.id } }),
-        undefined,
-        user.id
+        {
+          uid: user.id,
+        }
       )
 
       expect(filters).not.to.be.empty