From 1168d12018baae9e667828f2b53c7d34e292bddf Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 27 Jun 2024 17:10:58 +0800 Subject: [PATCH 1/8] use read replica for raw query --- packages/api/src/jobs/rss/refreshAllFeeds.ts | 15 +++++++++++---- packages/api/src/utils/logger.ts | 2 +- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/packages/api/src/jobs/rss/refreshAllFeeds.ts b/packages/api/src/jobs/rss/refreshAllFeeds.ts index 8d49b6096..a75385187 100644 --- a/packages/api/src/jobs/rss/refreshAllFeeds.ts +++ b/packages/api/src/jobs/rss/refreshAllFeeds.ts @@ -22,8 +22,12 @@ export const refreshAllFeeds = async (db: DataSource): Promise => { 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 => { 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, diff --git a/packages/api/src/utils/logger.ts b/packages/api/src/utils/logger.ts index 7e94ea108..6da5ccaa9 100644 --- a/packages/api/src/utils/logger.ts +++ b/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, }) } From 45986184c44dec0248ce658d2312bad6951b8462 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Tue, 18 Jun 2024 11:01:00 +0800 Subject: [PATCH 2/8] tidy --- packages/api/src/jobs/ai-summarize.ts | 20 ++-- packages/api/src/jobs/email/inbound_emails.ts | 5 +- packages/api/src/jobs/update_db.ts | 10 +- packages/api/src/repository/index.ts | 17 ++- .../api/src/resolvers/function_resolvers.ts | 1 + packages/api/src/routers/page_router.ts | 5 +- .../api/src/routers/svc/email_attachment.ts | 5 +- packages/api/src/routers/svc/webhooks.ts | 5 +- packages/api/src/routers/text_to_speech.ts | 5 +- packages/api/src/services/ai-summaries.ts | 5 +- packages/api/src/services/api_key.ts | 13 ++- packages/api/src/services/create_user.ts | 4 +- packages/api/src/services/explain.ts | 9 +- packages/api/src/services/features.ts | 17 +-- packages/api/src/services/highlights.ts | 35 +++--- packages/api/src/services/home.ts | 5 +- .../api/src/services/integrations/index.ts | 38 ++++--- packages/api/src/services/labels.ts | 50 +++++---- packages/api/src/services/library_item.ts | 100 +++++++++++------- packages/api/src/services/popular_reads.ts | 6 +- packages/api/src/services/profile.ts | 2 +- packages/api/src/services/received_emails.ts | 20 ++-- packages/api/src/services/recommendation.ts | 10 +- packages/api/src/services/rules.ts | 20 ++-- .../api/src/services/update_pdf_content.ts | 5 +- packages/api/src/services/upload_file.ts | 5 +- packages/api/src/services/user.ts | 54 ++++------ .../api/src/services/user_device_tokens.ts | 25 +++-- .../api/src/services/user_personalization.ts | 30 +++--- packages/api/src/services/webhook.ts | 38 +++---- packages/api/test/db.ts | 10 +- .../api/test/services/create_user.test.ts | 5 +- 32 files changed, 332 insertions(+), 247 deletions(-) diff --git a/packages/api/src/jobs/ai-summarize.ts b/packages/api/src/jobs/ai-summarize.ts index 974a6b347..94ca44f4e 100644 --- a/packages/api/src/jobs/ai-summarize.ts +++ b/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,9 @@ export const aiSummarize = async (jobData: AISummarizeJobData) => { tx .withRepository(libraryItemRepository) .findById(jobData.libraryItemId), - undefined, - jobData.userId + { + uid: jobData.userId, + } ) if (!libraryItem || libraryItem.state !== LibraryItemState.Succeeded) { logger.info( @@ -84,8 +85,9 @@ export const aiSummarize = async (jobData: AISummarizeJobData) => { summary: summary, }) }, - undefined, - jobData.userId + { + uid: jobData.userId, + } ) } catch (err) { console.log('error creating summary: ', err) diff --git a/packages/api/src/jobs/email/inbound_emails.ts b/packages/api/src/jobs/email/inbound_emails.ts index ab953e6af..6f1721660 100644 --- a/packages/api/src/jobs/email/inbound_emails.ts +++ b/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( diff --git a/packages/api/src/jobs/update_db.ts b/packages/api/src/jobs/update_db.ts index d178ad384..e2ba7ba15 100644 --- a/packages/api/src/jobs/update_db.ts +++ b/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, + } ) } diff --git a/packages/api/src/repository/index.ts b/packages/api/src/repository/index.ts index 21c59becd..615989867 100644 --- a/packages/api/src/repository/index.ts +++ b/packages/api/src/repository/index.ts @@ -1,4 +1,5 @@ import * as httpContext from 'express-http-context2' +import { DatabaseError } from 'pg' import { EntityManager, EntityTarget, @@ -7,7 +8,6 @@ import { QueryFailedError, Repository, } from 'typeorm' -import { DatabaseError } from 'pg' import { appDataSource } from '../data_source' import { Claims } from '../resolvers/types' import { SetClaimsRole } from '../utils/dictionary' @@ -59,12 +59,19 @@ export const setClaims = async ( ]) } +interface AuthTrxOptions { + entityManager?: EntityManager + uid?: string + userRole?: string +} + export const authTrx = async ( fn: (manager: EntityManager) => Promise, - em = appDataSource.manager, - uid?: string, - userRole?: string + options: AuthTrxOptions = {} ): Promise => { + const entityManage = options.entityManager || appDataSource.manager + 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,7 +79,7 @@ export const authTrx = async ( userRole = claims?.userRole } - return em.transaction(async (tx) => { + return entityManage.transaction(async (tx) => { await setClaims(tx, uid, userRole) return fn(tx) }) diff --git a/packages/api/src/resolvers/function_resolvers.ts b/packages/api/src/resolvers/function_resolvers.ts index aa226adc3..afa09e8bd 100644 --- a/packages/api/src/resolvers/function_resolvers.ts +++ b/packages/api/src/resolvers/function_resolvers.ts @@ -11,6 +11,7 @@ import { EXISTING_NEWSLETTER_FOLDER, NewsletterEmail, } from '../entity/newsletter_email' +import { Post } from '../entity/post' import { PublicItem } from '../entity/public_item' import { Recommendation } from '../entity/recommendation' import { diff --git a/packages/api/src/routers/page_router.ts b/packages/api/src/routers/page_router.ts index bcf2e5b46..504983482 100644 --- a/packages/api/src/routers/page_router.ts +++ b/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( diff --git a/packages/api/src/routers/svc/email_attachment.ts b/packages/api/src/routers/svc/email_attachment.ts index 2cf88f726..df7164f1b 100644 --- a/packages/api/src/routers/svc/email_attachment.ts +++ b/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) { diff --git a/packages/api/src/routers/svc/webhooks.ts b/packages/api/src/routers/svc/webhooks.ts index d11b45b95..f4bb1aaae 100644 --- a/packages/api/src/routers/svc/webhooks.ts +++ b/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) { diff --git a/packages/api/src/routers/text_to_speech.ts b/packages/api/src/routers/text_to_speech.ts index ddf208f36..8b59a2f7f 100644 --- a/packages/api/src/routers/text_to_speech.ts +++ b/packages/api/src/routers/text_to_speech.ts @@ -53,8 +53,9 @@ export function textToSpeechRouter() { state, }) }, - undefined, - userId + { + uid: userId, + } ) res.send('OK') diff --git a/packages/api/src/services/ai-summaries.ts b/packages/api/src/services/ai-summaries.ts index 3c94ca4cd..7bcc8a265 100644 --- a/packages/api/src/services/ai-summaries.ts +++ b/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 } diff --git a/packages/api/src/services/api_key.ts b/packages/api/src/services/api_key.ts index 95b613a2b..2f5cc8fea 100644 --- a/packages/api/src/services/api_key.ts +++ b/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, userId: string ) => { - return authTrx( - async (t) => t.getRepository(ApiKey).delete(criteria), - undefined, - userId - ) + return authTrx(async (t) => t.getRepository(ApiKey).delete(criteria), { + uid: userId, + }) } diff --git a/packages/api/src/services/create_user.ts b/packages/api/src/services/create_user.ts index a52b76c22..cb0795d45 100644 --- a/packages/api/src/services/create_user.ts +++ b/packages/api/src/services/create_user.ts @@ -188,7 +188,9 @@ const validateInvite = async ( const numMembers = await authTrx( (t) => t.getRepository(GroupMembership).countBy({ invite: { id: invite.id } }), - entityManager + { + entityManager, + } ) if (numMembers >= invite.maxMembers) { logger.info('rejecting invite, too many users', { invite, numMembers }) diff --git a/packages/api/src/services/explain.ts b/packages/api/src/services/explain.ts index 50a6389ab..2b4d6b812 100644 --- a/packages/api/src/services/explain.ts +++ b/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,9 @@ export const explainText = async ( const libraryItem = await authTrx( async (tx) => tx.withRepository(libraryItemRepository).findById(libraryItemId), - undefined, - userId + { + uid: userId, + } ) if (!libraryItem) { diff --git a/packages/api/src/services/features.ts b/packages/api/src/services/features.ts index f8fa2ded0..d1c11b4fe 100644 --- a/packages/api/src/services/features.ts +++ b/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,9 @@ export const userDigestEligible = async (uid: string): Promise => { where: { user: { id: uid }, status: SubscriptionStatus.Active }, }) }, - undefined, - uid + { + uid, + } ) const libraryItemsCount = await authTrx( @@ -192,8 +192,9 @@ export const userDigestEligible = async (uid: string): Promise => { where: { user: { id: uid } }, }) }, - undefined, - uid + { + uid, + } ) return subscriptionsCount >= 2 && libraryItemsCount >= 10 diff --git a/packages/api/src/services/highlights.ts b/packages/api/src/services/highlights.ts index 533d29eb2..b491cc06d 100644 --- a/packages/api/src/services/highlights.ts +++ b/packages/api/src/services/highlights.ts @@ -50,8 +50,9 @@ export const createHighlights = async ( return authTrx( async (tx) => tx.withRepository(highlightRepository).createAndSaves(highlights), - undefined, - userId + { + uid: userId, + } ) } @@ -73,8 +74,9 @@ export const createHighlight = async ( }, }) }, - undefined, - userId + { + uid: userId, + } ) const data = deepDelete(newHighlight, columnsToDelete) @@ -221,8 +223,9 @@ export const deleteHighlightById = async ( await highlightRepo.delete(highlightId) return highlight }, - undefined, - userId + { + uid: userId, + } ) await enqueueUpdateHighlight({ @@ -239,8 +242,9 @@ export const deleteHighlightsByIds = async ( ) => { await authTrx( async (tx) => tx.getRepository(Highlight).delete(highlightIds), - undefined, - userId + { + uid: userId, + } ) } @@ -255,8 +259,9 @@ export const findHighlightById = async ( id: highlightId, }) }, - undefined, - userId + { + uid: userId, + } ) } @@ -273,8 +278,9 @@ export const findHighlightsByLibraryItemId = async ( labels: true, }, }), - undefined, - userId + { + uid: userId, + } ) } @@ -321,7 +327,8 @@ export const searchHighlights = async ( return queryBuilder.getMany() }, - undefined, - userId + { + uid: userId, + } ) } diff --git a/packages/api/src/services/home.ts b/packages/api/src/services/home.ts index b78772377..6390d6dc6 100644 --- a/packages/api/src/services/home.ts +++ b/packages/api/src/services/home.ts @@ -48,7 +48,8 @@ export const findUnseenPublicItems = async ( .take(options.limit) .skip(options.offset) .getMany(), - undefined, - userId + { + uid: userId, + } ) } diff --git a/packages/api/src/services/integrations/index.ts b/packages/api/src/services/integrations/index.ts index 40e83fd0b..a8b11f916 100644 --- a/packages/api/src/services/integrations/index.ts +++ b/packages/api/src/services/integrations/index.ts @@ -27,11 +27,9 @@ export const deleteIntegrations = async ( userId: string, criteria: string[] | FindOptionsWhere ) => { - 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, + } ) } diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts index 9e47ea65f..97bdf5fc9 100644 --- a/packages/api/src/services/labels.ts +++ b/packages/api/src/services/labels.ts @@ -72,8 +72,9 @@ export const findOrCreateLabels = async ( user: { id: userId }, }) }, - undefined, - userId + { + uid: userId, + } ) } @@ -156,8 +157,9 @@ export const saveLabelsInLibraryItem = async ( })) ) }, - undefined, - userId + { + uid: userId, + } ) if (source === 'user') { @@ -197,8 +199,9 @@ export const addLabelsToLibraryItem = async ( [libraryItemId, source, labelIds] ) }, - undefined, - userId + { + uid: userId, + } ) // update labels in library item @@ -265,8 +268,9 @@ export const findLabelsByIds = async ( user: { id: userId }, }) }, - undefined, - userId + { + uid: userId, + } ) } @@ -278,8 +282,9 @@ export const createLabel = async ( return authTrx( (t) => t.withRepository(labelRepository).createLabel({ name, color }, userId), - undefined, - userId + { + uid: userId, + } ) } @@ -289,8 +294,9 @@ export const deleteLabels = async ( ) => { return authTrx( async (t) => t.withRepository(labelRepository).delete(criteria), - undefined, - userId + { + uid: userId, + } ) } @@ -326,8 +332,9 @@ export const updateLabel = async ( return repo.findOneByOrFail({ id }) }, - undefined, - userId + { + uid: userId, + } ) const libraryItemIds = await findLibraryItemIdsByLabelId(id, userId) @@ -348,8 +355,9 @@ export const findLabelsByUserId = async (userId: string): Promise => { where: { user: { id: userId } }, order: { position: 'ASC' }, }), - undefined, - userId + { + uid: userId, + } ) } @@ -359,8 +367,9 @@ export const findLabelById = async (id: string, userId: string) => { tx .withRepository(labelRepository) .findOneBy({ id, user: { id: userId } }), - undefined, - userId + { + uid: userId, + } ) } @@ -380,7 +389,8 @@ export const findLabelsByLibraryItemId = async ( source: el.source, })) }, - undefined, - userId + { + uid: userId, + } ) } diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index 4eee6af0c..c7756912e 100644 --- a/packages/api/src/services/library_item.ts +++ b/packages/api/src/services/library_item.ts @@ -707,8 +707,9 @@ export const createSearchQueryBuilder = ( export const countLibraryItems = async (args: SearchArgs, userId: string) => { return authTrx( async (tx) => createSearchQueryBuilder(args, userId, tx).getCount(), - undefined, - userId + { + uid: userId, + } ) } @@ -729,8 +730,9 @@ export const searchLibraryItems = async ( .skip(from) .take(size) .getMany(), - undefined, - userId + { + uid: userId, + } ) } @@ -774,8 +776,9 @@ export const findRecentLibraryItems = async ( .take(limit) .skip(offset) .getMany(), - undefined, - userId + { + uid: userId, + } ) } @@ -798,8 +801,9 @@ export const findLibraryItemsByIds = async ( .select(selectColumns) .where('library_item.id IN (:...ids)', { ids }) .getMany(), - undefined, - userId + { + uid: userId, + } ) } @@ -826,8 +830,9 @@ export const findLibraryItemById = async ( where: { id }, relations: options?.relations, }), - undefined, - userId + { + uid: userId, + } ) } @@ -848,8 +853,9 @@ 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, + } ) } @@ -889,8 +895,9 @@ export const softDeleteLibraryItem = async ( return itemRepo.findOneByOrFail({ id }) }, - undefined, - userId + { + uid: userId, + } ) await pubsub.entityDeleted(EntityType.ITEM, id, userId) @@ -924,8 +931,9 @@ export const updateLibraryItem = async ( return itemRepo.findOneByOrFail({ id }) }, - undefined, - userId + { + uid: userId, + } ) if (skipPubSub || libraryItem.state === LibraryItemState.Processing) { @@ -998,8 +1006,9 @@ export const updateLibraryItemReadingProgress = async ( `, [id, topPercent, bottomPercent, anchorIndex] ), - undefined, - userId + { + uid: userId, + } )) as [LibraryItem[], number] if (result[1] === 0) { return null @@ -1017,8 +1026,9 @@ export const createLibraryItems = async ( ): Promise => { return authTrx( async (tx) => tx.withRepository(libraryItemRepository).save(libraryItems), - undefined, - userId + { + uid: userId, + } ) } @@ -1094,8 +1104,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 @@ -1171,8 +1182,9 @@ export const countBySavedAt = async ( endDate, }) .getCount(), - undefined, - userId + { + uid: userId, + } ) } @@ -1253,8 +1265,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 +1279,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 +1315,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 +1337,9 @@ export const deleteLibraryItems = async ( return authTrx( async (tx) => tx.withRepository(libraryItemRepository).delete(items.map((i) => i.id)), - undefined, - userId + { + uid: userId, + } ) } @@ -1332,8 +1349,9 @@ export const deleteLibraryItemByUrl = async (url: string, userId: string) => { tx .withRepository(libraryItemRepository) .delete({ originalUrl: url, user: { id: userId } }), - undefined, - userId + { + uid: userId, + } ) } @@ -1343,8 +1361,9 @@ export const deleteLibraryItemsByUserId = async (userId: string) => { tx.withRepository(libraryItemRepository).delete({ user: { id: userId }, }), - undefined, - userId + { + uid: userId, + } ) } @@ -1403,8 +1422,9 @@ export const findLibraryItemIdsByLabelId = async ( return result.map((r) => r.library_item_id) }, - undefined, - userId + { + uid: userId, + } ) } diff --git a/packages/api/src/services/popular_reads.ts b/packages/api/src/services/popular_reads.ts index de6daab4d..8ff25d907 100644 --- a/packages/api/src/services/popular_reads.ts +++ b/packages/api/src/services/popular_reads.ts @@ -15,8 +15,10 @@ export const addPopularRead = async ( tx .withRepository(libraryItemRepository) .createByPopularRead(name, userId), - entityManager, - userId + { + entityManager, + uid: userId, + } ) } diff --git a/packages/api/src/services/profile.ts b/packages/api/src/services/profile.ts index 6547ab09f..de62bca6c 100644 --- a/packages/api/src/services/profile.ts +++ b/packages/api/src/services/profile.ts @@ -1,6 +1,6 @@ import { Profile } from '../entity/profile' import { User } from '../entity/user' -import { getRepository } from '../repository' +import { authTrx, getRepository } from '../repository' export const findProfile = async (user: User): Promise => { return getRepository(Profile).findOneBy({ user: { id: user.id } }) diff --git a/packages/api/src/services/received_emails.ts b/packages/api/src/services/received_emails.ts index 003c82d9d..973e3f20d 100644 --- a/packages/api/src/services/received_emails.ts +++ b/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,8 @@ export const findReceivedEmailById = async (id: string, userId: string) => { return authTrx( (t) => t.getRepository(ReceivedEmail).findOneBy({ id, user: { id: userId } }), - undefined, - userId + { + uid: userId, + } ) } diff --git a/packages/api/src/services/recommendation.ts b/packages/api/src/services/recommendation.ts index 9e58224f4..bbf8ffdf1 100644 --- a/packages/api/src/services/recommendation.ts +++ b/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, + } ) } diff --git a/packages/api/src/services/rules.ts b/packages/api/src/services/rules.ts index 6393a8c40..c0c394fb4 100644 --- a/packages/api/src/services/rules.ts +++ b/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, + } ) } diff --git a/packages/api/src/services/update_pdf_content.ts b/packages/api/src/services/update_pdf_content.ts index c96b67dce..03bec21e3 100644 --- a/packages/api/src/services/update_pdf_content.ts +++ b/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}`) diff --git a/packages/api/src/services/upload_file.ts b/packages/api/src/services/upload_file.ts index e2949600f..0b32d01ac 100644 --- a/packages/api/src/services/upload_file.ts +++ b/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, + } ) } diff --git a/packages/api/src/services/user.ts b/packages/api/src/services/user.ts index 171b92f1d..6b81a7352 100644 --- a/packages/api/src/services/user.ts +++ b/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) => { - 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 => { export const deleteUsers = async ( criteria: FindOptionsWhere | 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[]) => { - 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) => { @@ -95,7 +89,7 @@ export const batchDelete = async (criteria: FindOptionsWhere) => { 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) => { 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) => { 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, + } ) } diff --git a/packages/api/src/services/user_device_tokens.ts b/packages/api/src/services/user_device_tokens.ts index a8d8821e1..e622a2da3 100644 --- a/packages/api/src/services/user_device_tokens.ts +++ b/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, + } ) } diff --git a/packages/api/src/services/user_personalization.ts b/packages/api/src/services/user_personalization.ts index 5539ea462..66e6ea6e3 100644 --- a/packages/api/src/services/user_personalization.ts +++ b/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 => { 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 => { }) .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') diff --git a/packages/api/src/services/webhook.ts b/packages/api/src/services/webhook.ts index 736b6fc54..8c299ba95 100644 --- a/packages/api/src/services/webhook.ts +++ b/packages/api/src/services/webhook.ts @@ -7,11 +7,10 @@ export const createWebhooks = async ( userId?: string, entityManager?: EntityManager ) => { - return authTrx( - (tx) => tx.getRepository(Webhook).save(webhooks), - entityManager, - userId - ) + return authTrx((tx) => tx.getRepository(Webhook).save(webhooks), { + entityManager: entityManager, + uid: userId, + }) } export const createWebhook = async ( @@ -19,18 +18,18 @@ export const createWebhook = async ( userId?: string, entityManager?: EntityManager ) => { - return authTrx( - (tx) => tx.getRepository(Webhook).save(webhook), - entityManager, - userId - ) + return authTrx((tx) => tx.getRepository(Webhook).save(webhook), { + entityManager: entityManager, + uid: userId, + }) } export const findWebhooks = async (userId: string) => { return authTrx( (tx) => tx.getRepository(Webhook).findBy({ user: { id: userId } }), - undefined, - userId + { + uid: userId, + } ) } @@ -45,16 +44,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 +67,8 @@ export const deleteWebhook = async (id: string, userId: string) => { await repo.delete(id) return webhook }, - undefined, - userId + { + uid: userId, + } ) } diff --git a/packages/api/test/db.ts b/packages/api/test/db.ts index 5de68f713..fdb5c5b3f 100644 --- a/packages/api/test/db.ts +++ b/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({ diff --git a/packages/api/test/services/create_user.test.ts b/packages/api/test/services/create_user.test.ts index 3449b5f9d..0dcdb0f9a 100644 --- a/packages/api/test/services/create_user.test.ts +++ b/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 From 17eb5efa3701b34689be1dcc8e885fac015de03b Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 27 Jun 2024 17:16:17 +0800 Subject: [PATCH 3/8] resolve conflicts --- packages/api/src/resolvers/function_resolvers.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/api/src/resolvers/function_resolvers.ts b/packages/api/src/resolvers/function_resolvers.ts index afa09e8bd..aa226adc3 100644 --- a/packages/api/src/resolvers/function_resolvers.ts +++ b/packages/api/src/resolvers/function_resolvers.ts @@ -11,7 +11,6 @@ import { EXISTING_NEWSLETTER_FOLDER, NewsletterEmail, } from '../entity/newsletter_email' -import { Post } from '../entity/post' import { PublicItem } from '../entity/public_item' import { Recommendation } from '../entity/recommendation' import { From 0830f0b31280091f219f9e43dc598ce85f0efd0e Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 27 Jun 2024 18:38:43 +0800 Subject: [PATCH 4/8] allow defining replica mode in the auth transaction --- packages/api/src/repository/index.ts | 26 +++++++++++++++----- packages/api/src/services/create_user.ts | 16 ++++++------ packages/api/src/services/popular_reads.ts | 22 ++++------------- packages/api/src/services/webhook.ts | 10 +++----- packages/api/test/resolvers/webhooks.test.ts | 10 ++++---- 5 files changed, 40 insertions(+), 44 deletions(-) diff --git a/packages/api/src/repository/index.ts b/packages/api/src/repository/index.ts index 615989867..5667c45b2 100644 --- a/packages/api/src/repository/index.ts +++ b/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 ( fn: (manager: EntityManager) => Promise, options: AuthTrxOptions = {} ): Promise => { - 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 ( 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 = ( diff --git a/packages/api/src/services/create_user.ts b/packages/api/src/services/create_user.ts index cb0795d45..00a06658b 100644 --- a/packages/api/src/services/create_user.ts +++ b/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 diff --git a/packages/api/src/services/popular_reads.ts b/packages/api/src/services/popular_reads.ts index 8ff25d907..fb36fed4a 100644 --- a/packages/api/src/services/popular_reads.ts +++ b/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 => { 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) } diff --git a/packages/api/src/services/webhook.ts b/packages/api/src/services/webhook.ts index 8c299ba95..a0e4604ab 100644 --- a/packages/api/src/services/webhook.ts +++ b/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[], - userId?: string, - entityManager?: EntityManager + userId?: string ) => { return authTrx((tx) => tx.getRepository(Webhook).save(webhooks), { - entityManager: entityManager, uid: userId, }) } export const createWebhook = async ( webhook: DeepPartial, - userId?: string, - entityManager?: EntityManager + userId?: string ) => { return authTrx((tx) => tx.getRepository(Webhook).save(webhook), { - entityManager: entityManager, uid: userId, }) } diff --git a/packages/api/test/resolvers/webhooks.test.ts b/packages/api/test/resolvers/webhooks.test.ts index 989ab214b..0897ef68e 100644 --- a/packages/api/test/resolvers/webhooks.test.ts +++ b/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}") { From fb3d2eada657267631f482761ef8ce7627bf312b Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 27 Jun 2024 18:42:42 +0800 Subject: [PATCH 5/8] use primary and replica instead of master and slave --- packages/api/src/repository/index.ts | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/packages/api/src/repository/index.ts b/packages/api/src/repository/index.ts index 5667c45b2..d8c574725 100644 --- a/packages/api/src/repository/index.ts +++ b/packages/api/src/repository/index.ts @@ -6,6 +6,7 @@ import { ObjectLiteral, QueryBuilder, QueryFailedError, + ReplicationMode, Repository, } from 'typeorm' import { appDataSource } from '../data_source' @@ -62,7 +63,7 @@ export const setClaims = async ( interface AuthTrxOptions { uid?: string userRole?: string - replicationMode?: 'master' | 'slave' + replicationMode?: 'primary' | 'replica' } export const authTrx = async ( @@ -78,7 +79,16 @@ export const authTrx = async ( userRole = claims?.userRole } - const queryRunner = appDataSource.createQueryRunner(options.replicationMode) + 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() From 2ed480bb4984c2ed06aa59e4a66571fe802f1c10 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 27 Jun 2024 19:02:15 +0800 Subject: [PATCH 6/8] use read replica for selecting library items, labels and highlights --- packages/api/src/jobs/ai-summarize.ts | 1 + packages/api/src/services/explain.ts | 1 + packages/api/src/services/features.ts | 2 + packages/api/src/services/highlights.ts | 15 ++++-- packages/api/src/services/home.ts | 17 ++++--- packages/api/src/services/labels.ts | 31 ++++++++---- packages/api/src/services/library_item.ts | 52 +++++++++++++------- packages/api/src/services/profile.ts | 2 +- packages/api/src/services/received_emails.ts | 1 + 9 files changed, 83 insertions(+), 39 deletions(-) diff --git a/packages/api/src/jobs/ai-summarize.ts b/packages/api/src/jobs/ai-summarize.ts index 94ca44f4e..59740fdca 100644 --- a/packages/api/src/jobs/ai-summarize.ts +++ b/packages/api/src/jobs/ai-summarize.ts @@ -26,6 +26,7 @@ export const aiSummarize = async (jobData: AISummarizeJobData) => { .findById(jobData.libraryItemId), { uid: jobData.userId, + replicationMode: 'replica', } ) if (!libraryItem || libraryItem.state !== LibraryItemState.Succeeded) { diff --git a/packages/api/src/services/explain.ts b/packages/api/src/services/explain.ts index 2b4d6b812..77efe64bb 100644 --- a/packages/api/src/services/explain.ts +++ b/packages/api/src/services/explain.ts @@ -22,6 +22,7 @@ export const explainText = async ( tx.withRepository(libraryItemRepository).findById(libraryItemId), { uid: userId, + replicationMode: 'replica', } ) diff --git a/packages/api/src/services/features.ts b/packages/api/src/services/features.ts index d1c11b4fe..5e410842f 100644 --- a/packages/api/src/services/features.ts +++ b/packages/api/src/services/features.ts @@ -183,6 +183,7 @@ export const userDigestEligible = async (uid: string): Promise => { }, { uid, + replicationMode: 'replica', } ) @@ -194,6 +195,7 @@ export const userDigestEligible = async (uid: string): Promise => { }, { uid, + replicationMode: 'replica', } ) diff --git a/packages/api/src/services/highlights.ts b/packages/api/src/services/highlights.ts index b491cc06d..45c3bca57 100644 --- a/packages/api/src/services/highlights.ts +++ b/packages/api/src/services/highlights.ts @@ -23,10 +23,14 @@ export type HighlightEvent = Merge< export const batchGetHighlightsFromLibraryItemIds = async ( libraryItemIds: readonly string[] ): Promise => { - 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) => @@ -261,6 +265,7 @@ export const findHighlightById = async ( }, { uid: userId, + replicationMode: 'replica', } ) } @@ -280,6 +285,7 @@ export const findHighlightsByLibraryItemId = async ( }), { uid: userId, + replicationMode: 'replica', } ) } @@ -329,6 +335,7 @@ export const searchHighlights = async ( }, { uid: userId, + replicationMode: 'replica', } ) } diff --git a/packages/api/src/services/home.ts b/packages/api/src/services/home.ts index 6390d6dc6..5c037c3c5 100644 --- a/packages/api/src/services/home.ts +++ b/packages/api/src/services/home.ts @@ -4,12 +4,16 @@ import { authTrx } from '../repository' export const batchGetPublicItems = async ( ids: readonly string[] ): Promise> => { - 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)) @@ -50,6 +54,7 @@ export const findUnseenPublicItems = async ( .getMany(), { uid: userId, + replicationMode: 'replica', } ) } diff --git a/packages/api/src/services/labels.ts b/packages/api/src/services/labels.ts index 97bdf5fc9..22edb56a1 100644 --- a/packages/api/src/services/labels.ts +++ b/packages/api/src/services/labels.ts @@ -25,11 +25,15 @@ export type LabelEvent = Merge< export const batchGetLabelsFromLibraryItemIds = async ( libraryItemIds: readonly string[] ): Promise => { - 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 => { - 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) => @@ -270,6 +278,7 @@ export const findLabelsByIds = async ( }, { uid: userId, + replicationMode: 'replica', } ) } @@ -357,6 +366,7 @@ export const findLabelsByUserId = async (userId: string): Promise => { }), { uid: userId, + replicationMode: 'replica', } ) } @@ -369,6 +379,7 @@ export const findLabelById = async (id: string, userId: string) => { .findOneBy({ id, user: { id: userId } }), { uid: userId, + replicationMode: 'replica', } ) } diff --git a/packages/api/src/services/library_item.ts b/packages/api/src/services/library_item.ts index c7756912e..facb0b8c8 100644 --- a/packages/api/src/services/library_item.ts +++ b/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) @@ -709,6 +713,7 @@ export const countLibraryItems = async (args: SearchArgs, userId: string) => { async (tx) => createSearchQueryBuilder(args, userId, tx).getCount(), { uid: userId, + replicationMode: 'replica', } ) } @@ -732,6 +737,7 @@ export const searchLibraryItems = async ( .getMany(), { uid: userId, + replicationMode: 'replica', } ) } @@ -778,6 +784,7 @@ export const findRecentLibraryItems = async ( .getMany(), { uid: userId, + replicationMode: 'replica', } ) } @@ -803,6 +810,7 @@ export const findLibraryItemsByIds = async ( .getMany(), { uid: userId, + replicationMode: 'replica', } ) } @@ -832,6 +840,7 @@ export const findLibraryItemById = async ( }), { uid: userId, + replicationMode: 'replica', } ) } @@ -855,6 +864,7 @@ export const findLibraryItemByUrl = async ( .getOne(), { uid: userId, + replicationMode: 'replica', } ) } @@ -1153,17 +1163,21 @@ export const findLibraryItemsByPrefix = async ( ): Promise => { 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', + } ) } @@ -1184,6 +1198,7 @@ export const countBySavedAt = async ( .getCount(), { uid: userId, + replicationMode: 'replica', } ) } @@ -1424,6 +1439,7 @@ export const findLibraryItemIdsByLabelId = async ( }, { uid: userId, + replicationMode: 'replica', } ) } diff --git a/packages/api/src/services/profile.ts b/packages/api/src/services/profile.ts index de62bca6c..6547ab09f 100644 --- a/packages/api/src/services/profile.ts +++ b/packages/api/src/services/profile.ts @@ -1,6 +1,6 @@ import { Profile } from '../entity/profile' import { User } from '../entity/user' -import { authTrx, getRepository } from '../repository' +import { getRepository } from '../repository' export const findProfile = async (user: User): Promise => { return getRepository(Profile).findOneBy({ user: { id: user.id } }) diff --git a/packages/api/src/services/received_emails.ts b/packages/api/src/services/received_emails.ts index 973e3f20d..c3df236fc 100644 --- a/packages/api/src/services/received_emails.ts +++ b/packages/api/src/services/received_emails.ts @@ -60,6 +60,7 @@ export const findReceivedEmailById = async (id: string, userId: string) => { t.getRepository(ReceivedEmail).findOneBy({ id, user: { id: userId } }), { uid: userId, + replicationMode: 'replica', } ) } From 822f249fb1674ad7e2d4bec3e0ebdfcead046286 Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 27 Jun 2024 19:10:50 +0800 Subject: [PATCH 7/8] cont --- packages/api/src/resolvers/article/index.ts | 128 ++++++++++-------- packages/api/src/resolvers/highlight/index.ts | 15 +- packages/api/src/resolvers/labels/index.ts | 28 ++-- 3 files changed, 101 insertions(+), 70 deletions(-) diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index 0fd905bf7..73f17a2c3 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/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, 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 { diff --git a/packages/api/src/resolvers/highlight/index.ts b/packages/api/src/resolvers/highlight/index.ts index a52b5d06c..6dcf3cfe9 100644 --- a/packages/api/src/resolvers/highlight/index.ts +++ b/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, 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) => { diff --git a/packages/api/src/resolvers/labels/index.ts b/packages/api/src/resolvers/labels/index.ts index 2e3b9c6d2..be5ea342a 100644 --- a/packages/api/src/resolvers/labels/index.ts +++ b/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 { @@ -41,7 +42,7 @@ import { analytics } from '../../utils/analytics' import { authorized } from '../../utils/gql-utils' export const labelsResolver = authorized( - async (_obj, _params, { authTrx, log, uid }) => { + async (_obj, _params, { log, uid }) => { try { const user = await userRepository.findById(uid) if (!user) { @@ -50,16 +51,21 @@ export const labelsResolver = authorized( } } - 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, From 3ad826392eb5e4c0a043c80f75358ab8210a8ada Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Thu, 27 Jun 2024 20:55:04 +0800 Subject: [PATCH 8/8] use digest-score service --- packages/api/src/services/score.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/api/src/services/score.ts b/packages/api/src/services/score.ts index 25c78f5a0..8752e4b4a 100644 --- a/packages/api/src/services/score.ts +++ b/packages/api/src/services/score.ts @@ -79,4 +79,4 @@ class ScoreClientImpl implements ScoreClient { } } -export const scoreClient = new StubScoreClientImpl() +export const scoreClient = new ScoreClientImpl()