From 86c80d991f97bdfa93add75fcc78929bef531545 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Tue, 30 Jan 2024 13:02:32 +0800 Subject: [PATCH 01/20] Write behing cache for reading progress --- packages/api/src/apollo.ts | 4 + .../reading_progress_data_source.ts | 86 +++++++++++++++++++ packages/api/src/resolvers/article/index.ts | 7 +- .../api/src/resolvers/function_resolvers.ts | 86 ++++++++++++++++--- packages/api/src/resolvers/types.ts | 4 + 5 files changed, 172 insertions(+), 15 deletions(-) create mode 100644 packages/api/src/datasources/reading_progress_data_source.ts diff --git a/packages/api/src/apollo.ts b/packages/api/src/apollo.ts index 7d4c8ea02..ef8ea5cb5 100644 --- a/packages/api/src/apollo.ts +++ b/packages/api/src/apollo.ts @@ -25,6 +25,7 @@ import { tracer } from './tracing' import { getClaimsByToken, setAuthInCookie } from './utils/auth' import { SetClaimsRole } from './utils/dictionary' import { logger } from './utils/logger' +import { ReadingProgressDataSource } from './datasources/reading_progress_data_source' const signToken = promisify(jwt.sign) const pubsub = createPubSubClient() @@ -84,6 +85,9 @@ const contextFunc: ContextFunction = async ({ return cb(tx) }), tracingSpan: tracer.startSpan('apollo.request'), + dataSources: { + readingProgress: new ReadingProgressDataSource(), + }, } return ctx diff --git a/packages/api/src/datasources/reading_progress_data_source.ts b/packages/api/src/datasources/reading_progress_data_source.ts new file mode 100644 index 000000000..1155aef1a --- /dev/null +++ b/packages/api/src/datasources/reading_progress_data_source.ts @@ -0,0 +1,86 @@ +import { redisDataSource } from '../redis_data_source' + +type ReadingProgressCacheItem = { + readingProgressPercent: number + readingProgressTopPercent: number | undefined + readingProgressAnchorIndex: number | undefined + updatedAt: Date +} + +export class ReadingProgressDataSource { + private cacheItems: { [id: string]: ReadingProgressCacheItem } = {} + + constructor() {} + + async getReadingProgress( + libraryItemID: string + ): Promise { + const cacheKey = `omnivore:reading-progress:${libraryItemID}` + const cached = this.cacheItems[cacheKey] + if (cached) { + return cached + } + return this.valueFromRedis(libraryItemID) + } + + async updateReadingProgress( + libraryItemID: string, + progress: { + readingProgressPercent: number + readingProgressTopPercent: number | undefined | null + readingProgressAnchorIndex: number | undefined | null + } + ): Promise { + const cacheKey = `omnivore:reading-progress:${libraryItemID}` + const existingItem = await this.valueFromRedis(cacheKey) + const cacheItem = { + readingProgressPercent: Math.max( + progress.readingProgressPercent, + existingItem?.readingProgressPercent ?? 0 + ), + readingProgressTopPercent: Math.max( + progress.readingProgressTopPercent ?? 0, + existingItem?.readingProgressTopPercent ?? 0 + ), + readingProgressAnchorIndex: Math.max( + progress.readingProgressAnchorIndex ?? 0, + existingItem?.readingProgressAnchorIndex ?? 0 + ), + updatedAt: new Date(), + } + + this.cacheItems[cacheKey] = cacheItem + if (await redisDataSource.redisClient?.hmset(cacheKey, cacheItem)) { + console.log('cached reading progress') + } else { + console.log('failed to cache reading progress') + } + } + + async valueFromRedis( + libraryItemID: string + ): Promise { + const cacheKey = `omnivore:reading-progress:${libraryItemID}` + const redisCached = await redisDataSource.redisClient?.hgetall(cacheKey) + if (redisCached) { + const readingProgressPercent = parseInt( + redisCached.readingProgressPercent, + 10 + ) + const updatedAt = new Date(parseInt(redisCached.updatedAt, 10)) + if (!Number.isNaN(readingProgressPercent) && updatedAt) { + return { + readingProgressPercent, + readingProgressTopPercent: redisCached.readingProgressTopPercent + ? parseInt(redisCached.readingProgressTopPercent, 10) + : undefined, + readingProgressAnchorIndex: redisCached.readingProgressAnchorIndex + ? parseInt(redisCached.readingProgressAnchorIndex, 10) + : undefined, + updatedAt, + } + } + } + return undefined + } +} diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index 921a01cf8..5a5aa38f9 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/packages/api/src/resolvers/article/index.ts @@ -607,7 +607,7 @@ export const saveArticleReadingProgressResolver = authorized< force, }, }, - { log, pubsub, uid } + { log, pubsub, uid, dataSources } ) => { if ( readingProgressPercent < 0 || @@ -640,6 +640,11 @@ export const saveArticleReadingProgressResolver = authorized< } } + dataSources.readingProgress.updateReadingProgress(id, { + readingProgressPercent, + readingProgressTopPercent, + readingProgressAnchorIndex, + }) // update reading progress only if the current value is lower const updatedItem = await updateLibraryItemReadingProgress( id, diff --git a/packages/api/src/resolvers/function_resolvers.ts b/packages/api/src/resolvers/function_resolvers.ts index 871782068..bcc17360b 100644 --- a/packages/api/src/resolvers/function_resolvers.ts +++ b/packages/api/src/resolvers/function_resolvers.ts @@ -312,20 +312,6 @@ export const functionResolvers = { publishedAt(article: { publishedAt: Date }) { return validatedDate(article.publishedAt) }, - // async shareInfo( - // article: { id: string; sharedBy?: User; shareInfo?: LinkShareInfo }, - // __: unknown, - // ctx: WithDataSourcesContext - // ): Promise { - // if (article.shareInfo) return article.shareInfo - // if (!ctx.claims?.uid) return undefined - // return getShareInfoForArticle( - // ctx.kx, - // ctx.claims?.uid, - // article.id, - // ctx.models - // ) - // }, image(article: { image?: string }): string | undefined { return article.image && createImageProxyUrl(article.image, 320, 320) }, @@ -342,6 +328,42 @@ export const functionResolvers = { return findLabelsByLibraryItemId(article.id, ctx.uid) }, + async readingProgressPercent( + article: { id: string; readingProgressPercent?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress(article.id) + if (readingProgress) { + return readingProgress.readingProgressPercent + } + return article.readingProgressPercent + }, + async readingProgressAnchorIndex( + article: { id: string; readingProgressAnchorIndex?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress(article.id) + if (readingProgress) { + return readingProgress.readingProgressAnchorIndex + } + return article.readingProgressAnchorIndex + }, + async readingProgressTopPercent( + article: { id: string; readingProgressTopPercent?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress(article.id) + if (readingProgress) { + return readingProgress.readingProgressTopPercent + } + return article.readingProgressTopPercent + }, }, Highlight: { // async reactions( @@ -447,6 +469,42 @@ export const functionResolvers = { const highlights = await findHighlightsByLibraryItemId(item.id, ctx.uid) return highlights.map(highlightDataToHighlight) }, + async readingProgressPercent( + article: { id: string; readingProgressPercent?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress(article.id) + if (readingProgress) { + return readingProgress.readingProgressPercent + } + return article.readingProgressPercent + }, + async readingProgressAnchorIndex( + article: { id: string; readingProgressAnchorIndex?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress(article.id) + if (readingProgress) { + return readingProgress.readingProgressAnchorIndex + } + return article.readingProgressAnchorIndex + }, + async readingProgressTopPercent( + article: { id: string; readingProgressTopPercent?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress(article.id) + if (readingProgress) { + return readingProgress.readingProgressTopPercent + } + return article.readingProgressTopPercent + }, }, Subscription: { newsletterEmail(subscription: Subscription) { diff --git a/packages/api/src/resolvers/types.ts b/packages/api/src/resolvers/types.ts index 6c842d650..6fc24405f 100644 --- a/packages/api/src/resolvers/types.ts +++ b/packages/api/src/resolvers/types.ts @@ -5,6 +5,7 @@ import * as jwt from 'jsonwebtoken' import { EntityManager } from 'typeorm' import winston from 'winston' import { PubsubClient } from '../pubsub' +import { ReadingProgressDataSource } from '../datasources/reading_progress_data_source' export interface Claims { uid: string @@ -37,6 +38,9 @@ export interface RequestContext { userRole?: string ) => Promise tracingSpan: Span + dataSources: { + readingProgress: ReadingProgressDataSource + } } export type ResolverContext = ApolloContext From c2327781e1e8f34597c72ca83efae5c75e12ddbf Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Tue, 30 Jan 2024 13:17:26 +0800 Subject: [PATCH 02/20] Remove empty ctor --- packages/api/src/datasources/reading_progress_data_source.ts | 2 -- 1 file changed, 2 deletions(-) diff --git a/packages/api/src/datasources/reading_progress_data_source.ts b/packages/api/src/datasources/reading_progress_data_source.ts index 1155aef1a..1421bedcf 100644 --- a/packages/api/src/datasources/reading_progress_data_source.ts +++ b/packages/api/src/datasources/reading_progress_data_source.ts @@ -10,8 +10,6 @@ type ReadingProgressCacheItem = { export class ReadingProgressDataSource { private cacheItems: { [id: string]: ReadingProgressCacheItem } = {} - constructor() {} - async getReadingProgress( libraryItemID: string ): Promise { From fbfa93447996ed862830a7e7dc9f78aacfc2a27e Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 09:28:09 +0800 Subject: [PATCH 03/20] Write behind reading progress as a list that can be reduced at commit or read time --- .../reading_progress_data_source.ts | 72 ++++------ packages/api/src/queue-processor.ts | 19 +++ packages/api/src/resolvers/article/index.ts | 46 +++++-- .../api/src/resolvers/function_resolvers.ts | 128 ++++++++---------- packages/api/src/utils/createTask.ts | 1 + .../components/templates/article/Article.tsx | 2 +- 6 files changed, 137 insertions(+), 131 deletions(-) diff --git a/packages/api/src/datasources/reading_progress_data_source.ts b/packages/api/src/datasources/reading_progress_data_source.ts index 1421bedcf..5f9e7358b 100644 --- a/packages/api/src/datasources/reading_progress_data_source.ts +++ b/packages/api/src/datasources/reading_progress_data_source.ts @@ -4,80 +4,64 @@ type ReadingProgressCacheItem = { readingProgressPercent: number readingProgressTopPercent: number | undefined readingProgressAnchorIndex: number | undefined - updatedAt: Date + updatedAt: string } export class ReadingProgressDataSource { private cacheItems: { [id: string]: ReadingProgressCacheItem } = {} + constructor() {} + async getReadingProgress( + uid: string, libraryItemID: string ): Promise { - const cacheKey = `omnivore:reading-progress:${libraryItemID}` + const cacheKey = `omnivore:reading-progress:${uid}:${libraryItemID}` const cached = this.cacheItems[cacheKey] if (cached) { return cached } - return this.valueFromRedis(libraryItemID) + return this.valueFromRedis(cacheKey) } async updateReadingProgress( + uid: string, libraryItemID: string, progress: { readingProgressPercent: number - readingProgressTopPercent: number | undefined | null - readingProgressAnchorIndex: number | undefined | null + readingProgressTopPercent: number | undefined + readingProgressAnchorIndex: number | undefined } ): Promise { - const cacheKey = `omnivore:reading-progress:${libraryItemID}` - const existingItem = await this.valueFromRedis(cacheKey) - const cacheItem = { - readingProgressPercent: Math.max( - progress.readingProgressPercent, - existingItem?.readingProgressPercent ?? 0 - ), - readingProgressTopPercent: Math.max( - progress.readingProgressTopPercent ?? 0, - existingItem?.readingProgressTopPercent ?? 0 - ), - readingProgressAnchorIndex: Math.max( - progress.readingProgressAnchorIndex ?? 0, - existingItem?.readingProgressAnchorIndex ?? 0 - ), - updatedAt: new Date(), + const cacheKey = `omnivore:reading-progress:${uid}:${libraryItemID}` + const cacheItem: ReadingProgressCacheItem = { + ...progress, + updatedAt: new Date().toISOString(), } this.cacheItems[cacheKey] = cacheItem - if (await redisDataSource.redisClient?.hmset(cacheKey, cacheItem)) { - console.log('cached reading progress') + if ( + await redisDataSource.redisClient?.lpush( + cacheKey, + JSON.stringify(cacheItem) + ) + ) { + console.log('cached reading progress', cacheKey) } else { console.log('failed to cache reading progress') } } async valueFromRedis( - libraryItemID: string + cacheKey: string ): Promise { - const cacheKey = `omnivore:reading-progress:${libraryItemID}` - const redisCached = await redisDataSource.redisClient?.hgetall(cacheKey) - if (redisCached) { - const readingProgressPercent = parseInt( - redisCached.readingProgressPercent, - 10 - ) - const updatedAt = new Date(parseInt(redisCached.updatedAt, 10)) - if (!Number.isNaN(readingProgressPercent) && updatedAt) { - return { - readingProgressPercent, - readingProgressTopPercent: redisCached.readingProgressTopPercent - ? parseInt(redisCached.readingProgressTopPercent, 10) - : undefined, - readingProgressAnchorIndex: redisCached.readingProgressAnchorIndex - ? parseInt(redisCached.readingProgressAnchorIndex, 10) - : undefined, - updatedAt, - } - } + const redisCached = await redisDataSource.redisClient?.lrange( + cacheKey, + 0, + 0 + ) + if (redisCached && redisCached.length > 0) { + return JSON.parse(redisCached[0]) } return undefined } diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index ebf6fcfdf..afaa5b3ab 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -28,6 +28,10 @@ import { import { updatePDFContentJob } from './jobs/update_pdf_content' import { redisDataSource } from './redis_data_source' import { CustomTypeOrmLogger } from './utils/logger' +import { + SYNC_READ_POSITIONS_JOB_NAME, + syncReadPositionsJob, +} from './jobs/sync_read_positions' export const QUEUE_NAME = 'omnivore-backend-queue' @@ -152,6 +156,21 @@ const main = async () => { const worker = createWorker(workerRedisClient) + const queue = await getBackendQueue() + if (queue) { + await queue.add( + SYNC_READ_POSITIONS_JOB_NAME, + {}, + { + priority: 1, + repeat: { + every: 10000, + limit: 100, + }, + } + ) + } + const queueEvents = new QueueEvents(QUEUE_NAME, { connection: workerRedisClient, }) diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index 5a5aa38f9..2b95913f8 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/packages/api/src/resolvers/article/index.ts @@ -60,7 +60,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' @@ -640,19 +640,37 @@ export const saveArticleReadingProgressResolver = authorized< } } - dataSources.readingProgress.updateReadingProgress(id, { - readingProgressPercent, - readingProgressTopPercent, - readingProgressAnchorIndex, - }) - // update reading progress only if the current value is lower - const updatedItem = await updateLibraryItemReadingProgress( - id, - uid, - readingProgressPercent, - readingProgressTopPercent, - readingProgressAnchorIndex - ) + let updatedItem: LibraryItem | null + if (env.redis.cache && env.redis.mq) { + // If redis caching and queueing are available we delay this write + dataSources.readingProgress.updateReadingProgress(uid, id, { + readingProgressPercent, + readingProgressTopPercent: readingProgressTopPercent ?? undefined, + readingProgressAnchorIndex: readingProgressAnchorIndex ?? undefined, + }) + + updatedItem = await authTrx( + async (t) => { + return t.getRepository(LibraryItem).findOne({ + where: { + id, + }, + }) + }, + undefined, + uid + ) + } else { + // update reading progress only if the current value is lower + updatedItem = await updateLibraryItemReadingProgress( + id, + uid, + readingProgressPercent, + readingProgressTopPercent, + readingProgressAnchorIndex + ) + } + if (!updatedItem) { return { errorCodes: [SaveArticleReadingProgressErrorCode.BadData] } } diff --git a/packages/api/src/resolvers/function_resolvers.ts b/packages/api/src/resolvers/function_resolvers.ts index bcc17360b..8d3fb3187 100644 --- a/packages/api/src/resolvers/function_resolvers.ts +++ b/packages/api/src/resolvers/function_resolvers.ts @@ -158,6 +158,60 @@ const resultResolveTypeResolver = ( }, }) +const readingProgressHandlers = { + async readingProgressPercent( + article: { id: string; readingProgressPercent?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + if (ctx.claims?.uid) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress( + ctx.claims?.uid, + article.id + ) + if (readingProgress) { + return readingProgress.readingProgressPercent + } + } + return article.readingProgressPercent + }, + async readingProgressAnchorIndex( + article: { id: string; readingProgressAnchorIndex?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + if (ctx.claims?.uid) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress( + ctx.claims?.uid, + article.id + ) + if (readingProgress) { + return readingProgress.readingProgressAnchorIndex + } + } + return article.readingProgressAnchorIndex + }, + async readingProgressTopPercent( + article: { id: string; readingProgressTopPercent?: number }, + _: unknown, + ctx: WithDataSourcesContext + ) { + if (ctx.claims?.uid) { + const readingProgress = + await ctx.dataSources.readingProgress.getReadingProgress( + ctx.claims?.uid, + article.id + ) + if (readingProgress) { + return readingProgress.readingProgressTopPercent + } + } + return article.readingProgressTopPercent + }, +} + // Provide resolver functions for your schema fields export const functionResolvers = { Mutation: { @@ -328,42 +382,7 @@ export const functionResolvers = { return findLabelsByLibraryItemId(article.id, ctx.uid) }, - async readingProgressPercent( - article: { id: string; readingProgressPercent?: number }, - _: unknown, - ctx: WithDataSourcesContext - ) { - const readingProgress = - await ctx.dataSources.readingProgress.getReadingProgress(article.id) - if (readingProgress) { - return readingProgress.readingProgressPercent - } - return article.readingProgressPercent - }, - async readingProgressAnchorIndex( - article: { id: string; readingProgressAnchorIndex?: number }, - _: unknown, - ctx: WithDataSourcesContext - ) { - const readingProgress = - await ctx.dataSources.readingProgress.getReadingProgress(article.id) - if (readingProgress) { - return readingProgress.readingProgressAnchorIndex - } - return article.readingProgressAnchorIndex - }, - async readingProgressTopPercent( - article: { id: string; readingProgressTopPercent?: number }, - _: unknown, - ctx: WithDataSourcesContext - ) { - const readingProgress = - await ctx.dataSources.readingProgress.getReadingProgress(article.id) - if (readingProgress) { - return readingProgress.readingProgressTopPercent - } - return article.readingProgressTopPercent - }, + ...readingProgressHandlers, }, Highlight: { // async reactions( @@ -469,42 +488,7 @@ export const functionResolvers = { const highlights = await findHighlightsByLibraryItemId(item.id, ctx.uid) return highlights.map(highlightDataToHighlight) }, - async readingProgressPercent( - article: { id: string; readingProgressPercent?: number }, - _: unknown, - ctx: WithDataSourcesContext - ) { - const readingProgress = - await ctx.dataSources.readingProgress.getReadingProgress(article.id) - if (readingProgress) { - return readingProgress.readingProgressPercent - } - return article.readingProgressPercent - }, - async readingProgressAnchorIndex( - article: { id: string; readingProgressAnchorIndex?: number }, - _: unknown, - ctx: WithDataSourcesContext - ) { - const readingProgress = - await ctx.dataSources.readingProgress.getReadingProgress(article.id) - if (readingProgress) { - return readingProgress.readingProgressAnchorIndex - } - return article.readingProgressAnchorIndex - }, - async readingProgressTopPercent( - article: { id: string; readingProgressTopPercent?: number }, - _: unknown, - ctx: WithDataSourcesContext - ) { - const readingProgress = - await ctx.dataSources.readingProgress.getReadingProgress(article.id) - if (readingProgress) { - return readingProgress.readingProgressTopPercent - } - return article.readingProgressTopPercent - }, + ...readingProgressHandlers, }, Subscription: { newsletterEmail(subscription: Subscription) { diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts index 6a30a9ccd..b05387574 100644 --- a/packages/api/src/utils/createTask.ts +++ b/packages/api/src/utils/createTask.ts @@ -666,6 +666,7 @@ export const enqueueTriggerRuleJob = async (data: TriggerRuleJobData) => { attempts: 1, removeOnComplete: true, removeOnFail: true, + priority: 1, }) } diff --git a/packages/web/components/templates/article/Article.tsx b/packages/web/components/templates/article/Article.tsx index 2c3f4a5e4..091ad7e92 100644 --- a/packages/web/components/templates/article/Article.tsx +++ b/packages/web/components/templates/article/Article.tsx @@ -92,7 +92,7 @@ export function Article(props: ArticleProps): JSX.Element { setReadingProgress(bottomProgress * 100) } - }, 2500) + }, 3500) // Scroll to initial anchor position useEffect(() => { From a40f5bed55786b0c3dda3a7afe14372091ee54f9 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 10:17:51 +0800 Subject: [PATCH 04/20] Create a service for interacting with cached read positions --- .../reading_progress_data_source.ts | 38 ++----- .../src/services/cached_reading_position.ts | 106 ++++++++++++++++++ 2 files changed, 118 insertions(+), 26 deletions(-) create mode 100644 packages/api/src/services/cached_reading_position.ts diff --git a/packages/api/src/datasources/reading_progress_data_source.ts b/packages/api/src/datasources/reading_progress_data_source.ts index 5f9e7358b..3c860556f 100644 --- a/packages/api/src/datasources/reading_progress_data_source.ts +++ b/packages/api/src/datasources/reading_progress_data_source.ts @@ -1,17 +1,14 @@ import { redisDataSource } from '../redis_data_source' - -type ReadingProgressCacheItem = { - readingProgressPercent: number - readingProgressTopPercent: number | undefined - readingProgressAnchorIndex: number | undefined - updatedAt: string -} +import { + ReadingProgressCacheItem, + fetchCachedReadingPosition, + keyForCachedReadingPosition, + pushCachedReadingPosition, +} from '../services/cached_reading_position' export class ReadingProgressDataSource { private cacheItems: { [id: string]: ReadingProgressCacheItem } = {} - constructor() {} - async getReadingProgress( uid: string, libraryItemID: string @@ -21,7 +18,7 @@ export class ReadingProgressDataSource { if (cached) { return cached } - return this.valueFromRedis(cacheKey) + return fetchCachedReadingPosition(uid, libraryItemID) } async updateReadingProgress( @@ -33,11 +30,14 @@ export class ReadingProgressDataSource { readingProgressAnchorIndex: number | undefined } ): Promise { - const cacheKey = `omnivore:reading-progress:${uid}:${libraryItemID}` const cacheItem: ReadingProgressCacheItem = { - ...progress, + uid, + libraryItemID, updatedAt: new Date().toISOString(), + ...progress, } + const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) + pushCachedReadingPosition(uid, libraryItemID, cacheItem) this.cacheItems[cacheKey] = cacheItem if ( @@ -51,18 +51,4 @@ export class ReadingProgressDataSource { console.log('failed to cache reading progress') } } - - async valueFromRedis( - cacheKey: string - ): Promise { - const redisCached = await redisDataSource.redisClient?.lrange( - cacheKey, - 0, - 0 - ) - if (redisCached && redisCached.length > 0) { - return JSON.parse(redisCached[0]) - } - return undefined - } } diff --git a/packages/api/src/services/cached_reading_position.ts b/packages/api/src/services/cached_reading_position.ts new file mode 100644 index 000000000..3ae1965ac --- /dev/null +++ b/packages/api/src/services/cached_reading_position.ts @@ -0,0 +1,106 @@ +import { redisDataSource } from '../redis_data_source' +import { logger } from '../utils/logger' + +export type ReadingProgressCacheItem = { + uid: string + libraryItemID: string + readingProgressPercent: number + readingProgressTopPercent: number | undefined + readingProgressAnchorIndex: number | undefined + updatedAt: string | undefined +} + +export const keyForCachedReadingPosition = ( + uid: string, + libraryItemID: string +): string => { + return `omnivore:reading-progress:${uid}:${libraryItemID}` +} + +// Reading positions are cached as an array of positions, when +// we fetch them from the cache we find the maximum values +export const clearCachedReadingPosition = async ( + uid: string, + libraryItemID: string +): Promise => { + const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) + try { + const res = await redisDataSource.redisClient?.del(cacheKey) + return res ? res > 0 : false + } catch (error) { + logger.error('exception clearing cached reading position', { + cacheKey, + error, + }) + } + return false +} + +export const pushCachedReadingPosition = async ( + uid: string, + libraryItemID: string, + position: ReadingProgressCacheItem +): Promise => { + const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) + try { + const result = await redisDataSource.redisClient?.lpush( + cacheKey, + JSON.stringify(position) + ) + return result ? result > 0 : false + } catch (error) { + logger.error('error writing cached reading position', { cacheKey, error }) + } + return false +} + +// Reading positions are cached as an array of positions, when +// we fetch them from the cache we find the maximum values +export const fetchCachedReadingPosition = async ( + uid: string, + libraryItemID: string +): Promise => { + const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) + try { + const cacheItemList = await redisDataSource.redisClient?.lrange( + cacheKey, + 0, + -1 + ) + const items = cacheItemList?.map((item) => JSON.parse(item)) + if (!items || items.length < 1) { + return undefined + } + + const percent = Math.max( + ...items.map((o) => + 'readingProgressPercent' in o ? o.readingProgressPercent : 0 + ) + ) + const top = Math.max( + ...items.map((o) => + 'readingProgressTopPercent' in o ? o.readingProgressTopPercent : 0 + ) + ) + const anchor = Math.max( + ...items.map((o) => + 'readingProgressAnchorIndex' in o ? o.readingProgressAnchorIndex : 0 + ) + ) + + return { + uid, + libraryItemID, + readingProgressPercent: percent, + readingProgressTopPercent: top, + readingProgressAnchorIndex: anchor, + updatedAt: undefined, + } + } catch (error) { + logger.error('exception looking up cached reading position', { + cacheKey, + error, + }) + } + return undefined +} From 7bb3718a8f4bf371fb84a7fb7bd509e48d04f4f7 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 12:15:32 +0800 Subject: [PATCH 05/20] Add job to sync read position data --- packages/api/src/jobs/sync_read_positions.ts | 68 +++++++++++++++++++ .../src/services/cached_reading_position.ts | 22 +++++- 2 files changed, 89 insertions(+), 1 deletion(-) create mode 100644 packages/api/src/jobs/sync_read_positions.ts diff --git a/packages/api/src/jobs/sync_read_positions.ts b/packages/api/src/jobs/sync_read_positions.ts new file mode 100644 index 000000000..a91182048 --- /dev/null +++ b/packages/api/src/jobs/sync_read_positions.ts @@ -0,0 +1,68 @@ +import Redis from 'ioredis' +import { redisDataSource } from '../redis_data_source' +import { + CACHED_READING_POSITION_PREFIX, + componentsForCachedReadingPositionKey, + fetchCachedReadingPosition, +} from '../services/cached_reading_position' +import { logger } from '../utils/logger' +import { updateLibraryItemReadingProgress } from '../services/library_item' + +export const SYNC_READ_POSITIONS_JOB_NAME = 'sync-read-positions' + +async function* getSyncUpdatesIterator(redis: Redis) { + const match = `${CACHED_READING_POSITION_PREFIX}:*` + let [cursor, batch]: [string | number, string[]] = [0, []] + do { + ;[cursor, batch] = await redis.scan(cursor, 'MATCH', match, 'COUNT', 100) + if (batch.length) { + for (const key of batch) { + yield key + } + } + } while (cursor !== '0') + return +} + +const syncReadPosition = async (cacheKey: string) => { + const components = componentsForCachedReadingPositionKey(cacheKey) + const position = components + ? await fetchCachedReadingPosition(components.uid, components.libraryItemID) + : undefined + if (components && position) { + const result = await updateLibraryItemReadingProgress( + components.libraryItemID, + components.uid, + position.readingProgressPercent, + position.readingProgressTopPercent, + position.readingProgressAnchorIndex + ) + if (!result) { + logger.error('unable to update reading progress', { cacheKey }) + } + } else { + logger.warning( + 'potential error, reading position cache key found with no data', + { cacheKey } + ) + } + // Even if there are errors above we want to delete the key, otherwise + // in error scenarios we could accumulate a huge number of keys for + // something that is not critical (reading position) + const result = await redisDataSource.redisClient?.del(cacheKey) + if (!result || result < 1) { + logger.warning('error deleting cache key', { cacheKey }) + } +} + +export const syncReadPositionsJob = async (data: any, attempts: number) => { + const redis = redisDataSource.redisClient + if (!redis) { + throw new Error('unable to sync reading position, no redis client') + } + + const updates = getSyncUpdatesIterator(redis) + for await (const value of updates) { + await syncReadPosition(value) + } +} diff --git a/packages/api/src/services/cached_reading_position.ts b/packages/api/src/services/cached_reading_position.ts index 3ae1965ac..44e7edc72 100644 --- a/packages/api/src/services/cached_reading_position.ts +++ b/packages/api/src/services/cached_reading_position.ts @@ -10,11 +10,28 @@ export type ReadingProgressCacheItem = { updatedAt: string | undefined } +export const CACHED_READING_POSITION_PREFIX = `omnivore:reading-progress` + export const keyForCachedReadingPosition = ( uid: string, libraryItemID: string ): string => { - return `omnivore:reading-progress:${uid}:${libraryItemID}` + return `${CACHED_READING_POSITION_PREFIX}:${uid}:${libraryItemID}` +} + +export const componentsForCachedReadingPositionKey = ( + cacheKey: string +): { uid: string; libraryItemID: string } | undefined => { + try { + const [_owner, _prefix, uid, libraryItemID] = cacheKey.split(':') + return { + uid, + libraryItemID, + } + } catch (error) { + logger.log('exception getting cache key components', { cacheKey, error }) + } + return undefined } // Reading positions are cached as an array of positions, when @@ -60,6 +77,7 @@ export const fetchCachedReadingPosition = async ( uid: string, libraryItemID: string ): Promise => { + console.log('checking uid', uid, 'libraryItemId', libraryItemID) const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) try { const cacheItemList = await redisDataSource.redisClient?.lrange( @@ -67,7 +85,9 @@ export const fetchCachedReadingPosition = async ( 0, -1 ) + console.log('cacheItemList: ', cacheKey, cacheItemList) const items = cacheItemList?.map((item) => JSON.parse(item)) + console.log(' items[]: ', items) if (!items || items.length < 1) { return undefined } From d8ea4ff1c5a496098d691841543e19ca48a53a48 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 12:39:09 +0800 Subject: [PATCH 06/20] Add reading position messages to exported metrics for queue processor --- packages/api/src/queue-processor.ts | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index afaa5b3ab..ea458a6fe 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -32,6 +32,7 @@ import { SYNC_READ_POSITIONS_JOB_NAME, syncReadPositionsJob, } from './jobs/sync_read_positions' +import { CACHED_READING_POSITION_PREFIX } from './services/cached_reading_position' export const QUEUE_NAME = 'omnivore-backend-queue' @@ -136,6 +137,26 @@ const main = async () => { output += `omnivore_queue_messages_${metric}{queue="${QUEUE_NAME}"} ${counts[metric]}\n` }) + if (redisDataSource.redisClient) { + // Add read-position count, if its more than 10K items just denote + // 10_001. As this should never occur and means there is some + // other serious issue occurring. + const [cursor, batch] = await redisDataSource.redisClient.scan( + 0, + 'MATCH', + `${CACHED_READING_POSITION_PREFIX}:*`, + 'COUNT', + 10_000 + ) + if (cursor != '0') { + output += `# TYPE omnivore_read_position_messages gauge\n` + output += `omnivore_read_position_messages{queue="${QUEUE_NAME}"} ${10_001}\n` + } else if (batch) { + output += `# TYPE omnivore_read_position_messages gauge\n` + output += `omnivore_read_position_messages{queue="${QUEUE_NAME}"} ${batch.length}\n` + } + } + res.status(200).setHeader('Content-Type', 'text/plain').send(output) }) From 9dc070243a92e8c7a28388c8293081a9539a942a Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 12:40:02 +0800 Subject: [PATCH 07/20] Revert temp --- packages/api/src/utils/createTask.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts index b05387574..6a30a9ccd 100644 --- a/packages/api/src/utils/createTask.ts +++ b/packages/api/src/utils/createTask.ts @@ -666,7 +666,6 @@ export const enqueueTriggerRuleJob = async (data: TriggerRuleJobData) => { attempts: 1, removeOnComplete: true, removeOnFail: true, - priority: 1, }) } From a060aaf4961fd9e9db8ffbd398c4eb52786417ac Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 13:32:14 +0800 Subject: [PATCH 08/20] Make reading position parsing safer --- .../reading_progress_data_source.ts | 14 ++------ .../src/services/cached_reading_position.ts | 33 +++++++++++++++---- 2 files changed, 28 insertions(+), 19 deletions(-) diff --git a/packages/api/src/datasources/reading_progress_data_source.ts b/packages/api/src/datasources/reading_progress_data_source.ts index 3c860556f..6c08eca35 100644 --- a/packages/api/src/datasources/reading_progress_data_source.ts +++ b/packages/api/src/datasources/reading_progress_data_source.ts @@ -37,18 +37,8 @@ export class ReadingProgressDataSource { ...progress, } const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) - pushCachedReadingPosition(uid, libraryItemID, cacheItem) - - this.cacheItems[cacheKey] = cacheItem - if ( - await redisDataSource.redisClient?.lpush( - cacheKey, - JSON.stringify(cacheItem) - ) - ) { - console.log('cached reading progress', cacheKey) - } else { - console.log('failed to cache reading progress') + if (await pushCachedReadingPosition(uid, libraryItemID, cacheItem)) { + this.cacheItems[cacheKey] = cacheItem } } } diff --git a/packages/api/src/services/cached_reading_position.ts b/packages/api/src/services/cached_reading_position.ts index 44e7edc72..4b47a49a5 100644 --- a/packages/api/src/services/cached_reading_position.ts +++ b/packages/api/src/services/cached_reading_position.ts @@ -1,6 +1,8 @@ import { redisDataSource } from '../redis_data_source' import { logger } from '../utils/logger' +export const CACHED_READING_POSITION_PREFIX = `omnivore:reading-progress` + export type ReadingProgressCacheItem = { uid: string libraryItemID: string @@ -10,7 +12,23 @@ export type ReadingProgressCacheItem = { updatedAt: string | undefined } -export const CACHED_READING_POSITION_PREFIX = `omnivore:reading-progress` +export const isReadingProgressCacheItem = ( + item: any +): item is ReadingProgressCacheItem => { + return ( + 'uid' in item && 'libraryItemID' in item && 'readingProgressPercent' in item + ) +} + +export const parseReadingProgressCacheItem = ( + item: any +): ReadingProgressCacheItem | undefined => { + const result = JSON.parse(item) as unknown + if (isReadingProgressCacheItem(result)) { + return result + } + return undefined +} export const keyForCachedReadingPosition = ( uid: string, @@ -77,7 +95,6 @@ export const fetchCachedReadingPosition = async ( uid: string, libraryItemID: string ): Promise => { - console.log('checking uid', uid, 'libraryItemId', libraryItemID) const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) try { const cacheItemList = await redisDataSource.redisClient?.lrange( @@ -85,9 +102,9 @@ export const fetchCachedReadingPosition = async ( 0, -1 ) - console.log('cacheItemList: ', cacheKey, cacheItemList) - const items = cacheItemList?.map((item) => JSON.parse(item)) - console.log(' items[]: ', items) + const items = cacheItemList + ?.map((item) => parseReadingProgressCacheItem(item)) + .filter(isReadingProgressCacheItem) if (!items || items.length < 1) { return undefined } @@ -99,12 +116,14 @@ export const fetchCachedReadingPosition = async ( ) const top = Math.max( ...items.map((o) => - 'readingProgressTopPercent' in o ? o.readingProgressTopPercent : 0 + 'readingProgressTopPercent' in o ? o.readingProgressTopPercent ?? 0 : 0 ) ) const anchor = Math.max( ...items.map((o) => - 'readingProgressAnchorIndex' in o ? o.readingProgressAnchorIndex : 0 + 'readingProgressAnchorIndex' in o + ? o.readingProgressAnchorIndex ?? 0 + : 0 ) ) From 97efbe4487e8f5b44f4a2cd1804e9a40eb9c71c4 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 14:15:09 +0800 Subject: [PATCH 09/20] Force read position writes when user explicitly sets it --- packages/web/lib/networking/queries/useGetLibraryItemsQuery.tsx | 2 ++ packages/web/pages/[username]/[slug]/index.tsx | 1 + 2 files changed, 3 insertions(+) diff --git a/packages/web/lib/networking/queries/useGetLibraryItemsQuery.tsx b/packages/web/lib/networking/queries/useGetLibraryItemsQuery.tsx index cddf3d6f4..c289b3f7c 100644 --- a/packages/web/lib/networking/queries/useGetLibraryItemsQuery.tsx +++ b/packages/web/lib/networking/queries/useGetLibraryItemsQuery.tsx @@ -385,6 +385,7 @@ export function useGetLibraryItemsQuery({ }) articleReadingProgressMutation({ id: item.node.id, + force: true, readingProgressPercent: 100, readingProgressTopPercent: 100, readingProgressAnchorIndex: 0, @@ -402,6 +403,7 @@ export function useGetLibraryItemsQuery({ }) articleReadingProgressMutation({ id: item.node.id, + force: true, readingProgressPercent: 0, readingProgressTopPercent: 0, readingProgressAnchorIndex: 0, diff --git a/packages/web/pages/[username]/[slug]/index.tsx b/packages/web/pages/[username]/[slug]/index.tsx index 0293a3da0..22237f74b 100644 --- a/packages/web/pages/[username]/[slug]/index.tsx +++ b/packages/web/pages/[username]/[slug]/index.tsx @@ -155,6 +155,7 @@ export default function Home(): JSX.Element { if (article) { articleReadingProgressMutation({ id: article.id, + force: true, readingProgressPercent: 100, readingProgressTopPercent: 100, readingProgressAnchorIndex: 0, From bdf01ddaab8da23d54087183578d6950c804c5c0 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 14:15:29 +0800 Subject: [PATCH 10/20] Clear cache on force, return updated read position values --- packages/api/src/resolvers/article/index.ts | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index 2b95913f8..116af997f 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/packages/api/src/resolvers/article/index.ts @@ -112,6 +112,7 @@ import { parsePreparedContent, } from '../../utils/parser' import { getStorageFileDetails } from '../../utils/uploads' +import { clearCachedReadingPosition } from '../../services/cached_reading_position' export enum ArticleFormat { Markdown = 'markdown', @@ -621,7 +622,10 @@ export const saveArticleReadingProgressResolver = authorized< } try { if (force) { - // update reading progress without checking the current value + // update reading progress without checking the current value, also + // clear any cached values. + await clearCachedReadingPosition(uid, id) + const updatedItem = await updateLibraryItem( id, { @@ -660,6 +664,17 @@ export const saveArticleReadingProgressResolver = authorized< undefined, uid ) + if (updatedItem) { + updatedItem.readAt = new Date() + updatedItem.readingProgressBottomPercent = readingProgressPercent + if (readingProgressTopPercent) { + updatedItem.readingProgressTopPercent = readingProgressTopPercent + } + if (readingProgressAnchorIndex) { + updatedItem.readingProgressLastReadAnchor = + readingProgressAnchorIndex + } + } } else { // update reading progress only if the current value is lower updatedItem = await updateLibraryItemReadingProgress( From c47a720fe4b15c00708c4569b6aca902dba90a29 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 15:09:40 +0800 Subject: [PATCH 11/20] Do not cache on write, so the most updated item is always available on read This is mostly an issue on tests where all the queries can be run in a single process, but if we cache on write, it means the next read will just have the cached value, instead of the calculated maximum value. --- .../reading_progress_data_source.ts | 8 ++--- packages/api/src/resolvers/article/index.ts | 31 ++++++++----------- 2 files changed, 16 insertions(+), 23 deletions(-) diff --git a/packages/api/src/datasources/reading_progress_data_source.ts b/packages/api/src/datasources/reading_progress_data_source.ts index 6c08eca35..1f1702810 100644 --- a/packages/api/src/datasources/reading_progress_data_source.ts +++ b/packages/api/src/datasources/reading_progress_data_source.ts @@ -29,16 +29,14 @@ export class ReadingProgressDataSource { readingProgressTopPercent: number | undefined readingProgressAnchorIndex: number | undefined } - ): Promise { + ): Promise { const cacheItem: ReadingProgressCacheItem = { uid, libraryItemID, updatedAt: new Date().toISOString(), ...progress, } - const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) - if (await pushCachedReadingPosition(uid, libraryItemID, cacheItem)) { - this.cacheItems[cacheKey] = cacheItem - } + await pushCachedReadingPosition(uid, libraryItemID, cacheItem) + return fetchCachedReadingPosition(uid, libraryItemID) } } diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index 116af997f..5bb26b2e6 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/packages/api/src/resolvers/article/index.ts @@ -112,7 +112,10 @@ import { parsePreparedContent, } from '../../utils/parser' import { getStorageFileDetails } from '../../utils/uploads' -import { clearCachedReadingPosition } from '../../services/cached_reading_position' +import { + clearCachedReadingPosition, + fetchCachedReadingPosition, +} from '../../services/cached_reading_position' export enum ArticleFormat { Markdown = 'markdown', @@ -647,12 +650,16 @@ export const saveArticleReadingProgressResolver = authorized< let updatedItem: LibraryItem | null if (env.redis.cache && env.redis.mq) { // If redis caching and queueing are available we delay this write - dataSources.readingProgress.updateReadingProgress(uid, id, { - readingProgressPercent, - readingProgressTopPercent: readingProgressTopPercent ?? undefined, - readingProgressAnchorIndex: readingProgressAnchorIndex ?? undefined, - }) + const updatedProgress = + await dataSources.readingProgress.updateReadingProgress(uid, id, { + readingProgressPercent, + readingProgressTopPercent: readingProgressTopPercent ?? undefined, + readingProgressAnchorIndex: readingProgressAnchorIndex ?? undefined, + }) + // 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 updatedItem = await authTrx( async (t) => { return t.getRepository(LibraryItem).findOne({ @@ -664,19 +671,7 @@ export const saveArticleReadingProgressResolver = authorized< undefined, uid ) - if (updatedItem) { - updatedItem.readAt = new Date() - updatedItem.readingProgressBottomPercent = readingProgressPercent - if (readingProgressTopPercent) { - updatedItem.readingProgressTopPercent = readingProgressTopPercent - } - if (readingProgressAnchorIndex) { - updatedItem.readingProgressLastReadAnchor = - readingProgressAnchorIndex - } - } } else { - // update reading progress only if the current value is lower updatedItem = await updateLibraryItemReadingProgress( id, uid, From 0e37bfa03d62738537011d3b5bbd3e75349d4325 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 15:23:39 +0800 Subject: [PATCH 12/20] We need to update readAt because that isnt resolved from the cache --- packages/api/src/resolvers/article/index.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts index 5bb26b2e6..b68bf7024 100644 --- a/packages/api/src/resolvers/article/index.ts +++ b/packages/api/src/resolvers/article/index.ts @@ -671,6 +671,9 @@ export const saveArticleReadingProgressResolver = authorized< undefined, uid ) + if (updatedItem) { + updatedItem.readAt = new Date() + } } else { updatedItem = await updateLibraryItemReadingProgress( id, From 3ecbd3f434e41fc2b28254fcd2554aae52c5f91e Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 15:24:09 +0800 Subject: [PATCH 13/20] Need to force this test to overwrite the cached value --- packages/api/test/resolvers/article.test.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/api/test/resolvers/article.test.ts b/packages/api/test/resolvers/article.test.ts index 1b49a5a83..2587e4754 100644 --- a/packages/api/test/resolvers/article.test.ts +++ b/packages/api/test/resolvers/article.test.ts @@ -628,7 +628,6 @@ describe('Article API', () => { ).expect(200) const savedItem = await findLibraryItemByUrl(url, user.id) - console.log('savedItem: ', savedItem) expect(savedItem?.archivedAt).to.not.be.null expect(savedItem?.labels?.map((l) => l.name)).to.eql(labels) }) @@ -779,7 +778,12 @@ describe('Article API', () => { it('saves topPercent as 0 if defined as 0', async () => { const topPercent = 0 - query = saveArticleReadingProgressQuery(itemId, progress, topPercent) + query = saveArticleReadingProgressQuery( + itemId, + progress, + topPercent, + true + ) const res = await graphqlRequest(query, authToken).expect(200) expect( res.body.data.saveArticleReadingProgress.updatedArticle From 38a90e39874ec91cb314b6f3a788375cee169668 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 15:29:52 +0800 Subject: [PATCH 14/20] Revert web change --- packages/web/components/templates/article/Article.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/web/components/templates/article/Article.tsx b/packages/web/components/templates/article/Article.tsx index 091ad7e92..2c3f4a5e4 100644 --- a/packages/web/components/templates/article/Article.tsx +++ b/packages/web/components/templates/article/Article.tsx @@ -92,7 +92,7 @@ export function Article(props: ArticleProps): JSX.Element { setReadingProgress(bottomProgress * 100) } - }, 3500) + }, 2500) // Scroll to initial anchor position useEffect(() => { From 570b7a4b7826b12a5875378e95c8c1609b4b99c6 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 20:20:27 +0800 Subject: [PATCH 15/20] Use a set in redis and push items to the set This lets us grab all available items, apply them, and then remove them from the set. Once completed the items that were applied can be removed from the set, if any new items were added during that time they wont be effected. --- packages/api/src/jobs/sync_read_positions.ts | 54 ++++++++++----- packages/api/src/queue-processor.ts | 3 +- .../src/services/cached_reading_position.ts | 67 ++++++++++++++++--- 3 files changed, 95 insertions(+), 29 deletions(-) diff --git a/packages/api/src/jobs/sync_read_positions.ts b/packages/api/src/jobs/sync_read_positions.ts index a91182048..2223ef6c7 100644 --- a/packages/api/src/jobs/sync_read_positions.ts +++ b/packages/api/src/jobs/sync_read_positions.ts @@ -3,7 +3,8 @@ import { redisDataSource } from '../redis_data_source' import { CACHED_READING_POSITION_PREFIX, componentsForCachedReadingPositionKey, - fetchCachedReadingPosition, + fetchCachedReadingPositionsAndMembers, + reduceCachedReadingPositionMembers, } from '../services/cached_reading_position' import { logger } from '../utils/logger' import { updateLibraryItemReadingProgress } from '../services/library_item' @@ -26,19 +27,43 @@ async function* getSyncUpdatesIterator(redis: Redis) { const syncReadPosition = async (cacheKey: string) => { const components = componentsForCachedReadingPositionKey(cacheKey) - const position = components - ? await fetchCachedReadingPosition(components.uid, components.libraryItemID) + const positions = components + ? await fetchCachedReadingPositionsAndMembers( + components.uid, + components.libraryItemID + ) : undefined - if (components && position) { - const result = await updateLibraryItemReadingProgress( - components.libraryItemID, + if ( + components && + positions && + positions.positionItems && + positions.positionItems.length > 0 + ) { + const position = await reduceCachedReadingPositionMembers( components.uid, - position.readingProgressPercent, - position.readingProgressTopPercent, - position.readingProgressAnchorIndex + components.libraryItemID, + positions.positionItems ) - if (!result) { - logger.error('unable to update reading progress', { cacheKey }) + if (position) { + // this will throw if there is an error + await updateLibraryItemReadingProgress( + components.libraryItemID, + components.uid, + position.readingProgressPercent, + position.readingProgressTopPercent, + position.readingProgressAnchorIndex + ) + } + + const removed = await redisDataSource.redisClient?.srem( + cacheKey, + ...positions.members + ) + if (!removed || removed < positions.members.length) { + logger.warning( + 'potential error, reading position cache key members not removed', + { cacheKey } + ) } } else { logger.warning( @@ -46,13 +71,6 @@ const syncReadPosition = async (cacheKey: string) => { { cacheKey } ) } - // Even if there are errors above we want to delete the key, otherwise - // in error scenarios we could accumulate a huge number of keys for - // something that is not critical (reading position) - const result = await redisDataSource.redisClient?.del(cacheKey) - if (!result || result < 1) { - logger.warning('error deleting cache key', { cacheKey }) - } } export const syncReadPositionsJob = async (data: any, attempts: number) => { diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index ea458a6fe..54a9c5bad 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -179,13 +179,14 @@ const main = async () => { const queue = await getBackendQueue() if (queue) { + // run every 60s await queue.add( SYNC_READ_POSITIONS_JOB_NAME, {}, { priority: 1, repeat: { - every: 10000, + every: 60_000, limit: 100, }, } diff --git a/packages/api/src/services/cached_reading_position.ts b/packages/api/src/services/cached_reading_position.ts index 4b47a49a5..c6635cb1a 100644 --- a/packages/api/src/services/cached_reading_position.ts +++ b/packages/api/src/services/cached_reading_position.ts @@ -78,7 +78,10 @@ export const pushCachedReadingPosition = async ( ): Promise => { const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) try { - const result = await redisDataSource.redisClient?.lpush( + // Its critical that the date is set so the entry will be a unique + // set value. + position.updatedAt = new Date().toISOString() + const result = await redisDataSource.redisClient?.sadd( cacheKey, JSON.stringify(position) ) @@ -95,16 +98,35 @@ export const fetchCachedReadingPosition = async ( uid: string, libraryItemID: string ): Promise => { - const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) try { - const cacheItemList = await redisDataSource.redisClient?.lrange( - cacheKey, - 0, - -1 + const items = await fetchCachedReadingPositionsAndMembers( + uid, + libraryItemID ) - const items = cacheItemList - ?.map((item) => parseReadingProgressCacheItem(item)) - .filter(isReadingProgressCacheItem) + if (!items) { + return undefined + } + return reduceCachedReadingPositionMembers( + uid, + libraryItemID, + items.positionItems + ) + } catch (error) { + logger.error('exception looking up cached reading position', { + uid, + libraryItemID, + error, + }) + } + return undefined +} + +export const reduceCachedReadingPositionMembers = async ( + uid: string, + libraryItemID: string, + items: ReadingProgressCacheItem[] +): Promise => { + try { if (!items || items.length < 1) { return undefined } @@ -126,7 +148,6 @@ export const fetchCachedReadingPosition = async ( : 0 ) ) - return { uid, libraryItemID, @@ -135,6 +156,32 @@ export const fetchCachedReadingPosition = async ( readingProgressAnchorIndex: anchor, updatedAt: undefined, } + } catch (error) { + logger.error('exception reducing cached reading items', { + uid, + libraryItemID, + error, + }) + } + return undefined +} + +export const fetchCachedReadingPositionsAndMembers = async ( + uid: string, + libraryItemID: string +): Promise< + { positionItems: ReadingProgressCacheItem[]; members: string[] } | undefined +> => { + const cacheKey = keyForCachedReadingPosition(uid, libraryItemID) + try { + const members = await redisDataSource.redisClient?.smembers(cacheKey) + if (!members) { + return undefined + } + const positionItems = members + ?.map((item) => parseReadingProgressCacheItem(item)) + .filter(isReadingProgressCacheItem) + return { members, positionItems } } catch (error) { logger.error('exception looking up cached reading position', { cacheKey, From ddf7d590fd22ab535c164b54f25e094afb33cc69 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Wed, 31 Jan 2024 21:04:21 +0800 Subject: [PATCH 16/20] Copy params so no extra params are sent --- .../mutations/mergeHighlightMutation.ts | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/packages/web/lib/networking/mutations/mergeHighlightMutation.ts b/packages/web/lib/networking/mutations/mergeHighlightMutation.ts index 4b7d23bda..9da67bb8d 100644 --- a/packages/web/lib/networking/mutations/mergeHighlightMutation.ts +++ b/packages/web/lib/networking/mutations/mergeHighlightMutation.ts @@ -57,7 +57,22 @@ export async function mergeHighlightMutation( ` try { - const data = await gqlFetcher(mutation, { input }) + const data = await gqlFetcher(mutation, { + input: { + id: input.id, + shortId: input.shortId, + articleId: input.articleId, + patch: input.patch, + quote: input.quote, + prefix: input.prefix, + suffix: input.suffix, + html: input.html, + annotation: input.annotation, + overlapHighlightIdList: input.overlapHighlightIdList, + highlightPositionPercent: input.highlightPositionPercent, + highlightPositionAnchorIndex: input.highlightPositionAnchorIndex, + }, + }) const output = data as MergeHighlightOutput | undefined return output?.mergeHighlight.highlight } catch { From 7d6645b87a55977c0aac32b878d9ed9bd28ce12e Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Thu, 1 Feb 2024 10:02:38 +0800 Subject: [PATCH 17/20] Rebase --- packages/api/src/queue-processor.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index 54a9c5bad..724a69f6e 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -82,6 +82,8 @@ export const createWorker = (connection: ConnectionOptions) => return updateLabels(job.data) case UPDATE_HIGHLIGHT_JOB: return updateHighlight(job.data) + case SYNC_READ_POSITIONS_JOB_NAME: + return syncReadPositionsJob(job.data) } }, { From 1b44cb860bea59234c3148c9d1824298e00ac76d Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Thu, 1 Feb 2024 10:33:51 +0800 Subject: [PATCH 18/20] Function for cron jobs dont include queue in the read position metrics since its not part of that queue --- packages/api/src/jobs/sync_read_positions.ts | 2 +- packages/api/src/queue-processor.ts | 40 +++++++++++--------- 2 files changed, 24 insertions(+), 18 deletions(-) diff --git a/packages/api/src/jobs/sync_read_positions.ts b/packages/api/src/jobs/sync_read_positions.ts index 2223ef6c7..818347865 100644 --- a/packages/api/src/jobs/sync_read_positions.ts +++ b/packages/api/src/jobs/sync_read_positions.ts @@ -73,7 +73,7 @@ const syncReadPosition = async (cacheKey: string) => { } } -export const syncReadPositionsJob = async (data: any, attempts: number) => { +export const syncReadPositionsJob = async (_data: any) => { const redis = redisDataSource.redisClient if (!redis) { throw new Error('unable to sync reading position, no redis client') diff --git a/packages/api/src/queue-processor.ts b/packages/api/src/queue-processor.ts index 724a69f6e..d1ec1ed2b 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -27,7 +27,7 @@ import { } from './jobs/update_db' import { updatePDFContentJob } from './jobs/update_pdf_content' import { redisDataSource } from './redis_data_source' -import { CustomTypeOrmLogger } from './utils/logger' +import { logger, CustomTypeOrmLogger } from './utils/logger' import { SYNC_READ_POSITIONS_JOB_NAME, syncReadPositionsJob, @@ -91,6 +91,26 @@ export const createWorker = (connection: ConnectionOptions) => } ) +const setupCronJobs = async () => { + const queue = await getBackendQueue() + if (!queue) { + logger.error('Unable to setup cron jobs. Queue is not available.') + return + } + + await queue.add( + SYNC_READ_POSITIONS_JOB_NAME, + {}, + { + priority: 1, + repeat: { + every: 60_000, + limit: 100, + }, + } + ) +} + const main = async () => { console.log('[queue-processor]: starting queue processor') @@ -155,7 +175,7 @@ const main = async () => { output += `omnivore_read_position_messages{queue="${QUEUE_NAME}"} ${10_001}\n` } else if (batch) { output += `# TYPE omnivore_read_position_messages gauge\n` - output += `omnivore_read_position_messages{queue="${QUEUE_NAME}"} ${batch.length}\n` + output += `omnivore_read_position_messages{} ${batch.length}\n` } } @@ -179,21 +199,7 @@ const main = async () => { const worker = createWorker(workerRedisClient) - const queue = await getBackendQueue() - if (queue) { - // run every 60s - await queue.add( - SYNC_READ_POSITIONS_JOB_NAME, - {}, - { - priority: 1, - repeat: { - every: 60_000, - limit: 100, - }, - } - ) - } + await setupCronJobs() const queueEvents = new QueueEvents(QUEUE_NAME, { connection: workerRedisClient, From 792f13edfca6f228758aff67ebd4672848394106 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Thu, 1 Feb 2024 10:49:48 +0800 Subject: [PATCH 19/20] Remove async from non-async function --- packages/api/src/services/cached_reading_position.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/api/src/services/cached_reading_position.ts b/packages/api/src/services/cached_reading_position.ts index c6635cb1a..a516c4978 100644 --- a/packages/api/src/services/cached_reading_position.ts +++ b/packages/api/src/services/cached_reading_position.ts @@ -121,7 +121,7 @@ export const fetchCachedReadingPosition = async ( return undefined } -export const reduceCachedReadingPositionMembers = async ( +export const reduceCachedReadingPositionMembers = ( uid: string, libraryItemID: string, items: ReadingProgressCacheItem[] From 02c28be403361f3906cedbac8692394b8b258f12 Mon Sep 17 00:00:00 2001 From: Jackson Harper Date: Thu, 1 Feb 2024 11:24:13 +0800 Subject: [PATCH 20/20] reduce unction is no longer async --- packages/api/src/jobs/sync_read_positions.ts | 2 +- packages/api/src/services/cached_reading_position.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/api/src/jobs/sync_read_positions.ts b/packages/api/src/jobs/sync_read_positions.ts index 818347865..1b25eab13 100644 --- a/packages/api/src/jobs/sync_read_positions.ts +++ b/packages/api/src/jobs/sync_read_positions.ts @@ -39,7 +39,7 @@ const syncReadPosition = async (cacheKey: string) => { positions.positionItems && positions.positionItems.length > 0 ) { - const position = await reduceCachedReadingPositionMembers( + const position = reduceCachedReadingPositionMembers( components.uid, components.libraryItemID, positions.positionItems diff --git a/packages/api/src/services/cached_reading_position.ts b/packages/api/src/services/cached_reading_position.ts index a516c4978..b69a82ac8 100644 --- a/packages/api/src/services/cached_reading_position.ts +++ b/packages/api/src/services/cached_reading_position.ts @@ -125,7 +125,7 @@ export const reduceCachedReadingPositionMembers = ( uid: string, libraryItemID: string, items: ReadingProgressCacheItem[] -): Promise => { +): ReadingProgressCacheItem | undefined => { try { if (!items || items.length < 1) { return undefined