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..1f1702810 --- /dev/null +++ b/packages/api/src/datasources/reading_progress_data_source.ts @@ -0,0 +1,42 @@ +import { redisDataSource } from '../redis_data_source' +import { + ReadingProgressCacheItem, + fetchCachedReadingPosition, + keyForCachedReadingPosition, + pushCachedReadingPosition, +} from '../services/cached_reading_position' + +export class ReadingProgressDataSource { + private cacheItems: { [id: string]: ReadingProgressCacheItem } = {} + + async getReadingProgress( + uid: string, + libraryItemID: string + ): Promise { + const cacheKey = `omnivore:reading-progress:${uid}:${libraryItemID}` + const cached = this.cacheItems[cacheKey] + if (cached) { + return cached + } + return fetchCachedReadingPosition(uid, libraryItemID) + } + + async updateReadingProgress( + uid: string, + libraryItemID: string, + progress: { + readingProgressPercent: number + readingProgressTopPercent: number | undefined + readingProgressAnchorIndex: number | undefined + } + ): Promise { + const cacheItem: ReadingProgressCacheItem = { + uid, + libraryItemID, + updatedAt: new Date().toISOString(), + ...progress, + } + await pushCachedReadingPosition(uid, libraryItemID, cacheItem) + return fetchCachedReadingPosition(uid, libraryItemID) + } +} 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..1b25eab13 --- /dev/null +++ b/packages/api/src/jobs/sync_read_positions.ts @@ -0,0 +1,86 @@ +import Redis from 'ioredis' +import { redisDataSource } from '../redis_data_source' +import { + CACHED_READING_POSITION_PREFIX, + componentsForCachedReadingPositionKey, + fetchCachedReadingPositionsAndMembers, + reduceCachedReadingPositionMembers, +} 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 positions = components + ? await fetchCachedReadingPositionsAndMembers( + components.uid, + components.libraryItemID + ) + : undefined + if ( + components && + positions && + positions.positionItems && + positions.positionItems.length > 0 + ) { + const position = reduceCachedReadingPositionMembers( + components.uid, + components.libraryItemID, + positions.positionItems + ) + 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( + 'potential error, reading position cache key found with no data', + { cacheKey } + ) + } +} + +export const syncReadPositionsJob = async (_data: any) => { + 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/queue-processor.ts b/packages/api/src/queue-processor.ts index ebf6fcfdf..d1ec1ed2b 100644 --- a/packages/api/src/queue-processor.ts +++ b/packages/api/src/queue-processor.ts @@ -27,7 +27,12 @@ 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, +} from './jobs/sync_read_positions' +import { CACHED_READING_POSITION_PREFIX } from './services/cached_reading_position' export const QUEUE_NAME = 'omnivore-backend-queue' @@ -77,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) } }, { @@ -84,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') @@ -132,6 +159,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{} ${batch.length}\n` + } + } + res.status(200).setHeader('Content-Type', 'text/plain').send(output) }) @@ -152,6 +199,8 @@ const main = async () => { const worker = createWorker(workerRedisClient) + await setupCronJobs() + 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 921a01cf8..b68bf7024 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' @@ -112,6 +112,10 @@ import { parsePreparedContent, } from '../../utils/parser' import { getStorageFileDetails } from '../../utils/uploads' +import { + clearCachedReadingPosition, + fetchCachedReadingPosition, +} from '../../services/cached_reading_position' export enum ArticleFormat { Markdown = 'markdown', @@ -607,7 +611,7 @@ export const saveArticleReadingProgressResolver = authorized< force, }, }, - { log, pubsub, uid } + { log, pubsub, uid, dataSources } ) => { if ( readingProgressPercent < 0 || @@ -621,7 +625,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, { @@ -640,14 +647,43 @@ export const saveArticleReadingProgressResolver = authorized< } } - // 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 + 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({ + where: { + id, + }, + }) + }, + undefined, + uid + ) + if (updatedItem) { + updatedItem.readAt = new Date() + } + } else { + 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 871782068..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: { @@ -312,20 +366,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 +382,7 @@ export const functionResolvers = { return findLabelsByLibraryItemId(article.id, ctx.uid) }, + ...readingProgressHandlers, }, Highlight: { // async reactions( @@ -447,6 +488,7 @@ export const functionResolvers = { const highlights = await findHighlightsByLibraryItemId(item.id, ctx.uid) return highlights.map(highlightDataToHighlight) }, + ...readingProgressHandlers, }, 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 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..b69a82ac8 --- /dev/null +++ b/packages/api/src/services/cached_reading_position.ts @@ -0,0 +1,192 @@ +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 + readingProgressPercent: number + readingProgressTopPercent: number | undefined + readingProgressAnchorIndex: number | undefined + updatedAt: string | undefined +} + +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, + libraryItemID: string +): string => { + 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 +// 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 { + // 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) + ) + 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 => { + try { + const items = await fetchCachedReadingPositionsAndMembers( + uid, + libraryItemID + ) + 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 = ( + uid: string, + libraryItemID: string, + items: ReadingProgressCacheItem[] +): ReadingProgressCacheItem | undefined => { + try { + 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 : 0 + ) + ) + const anchor = Math.max( + ...items.map((o) => + 'readingProgressAnchorIndex' in o + ? o.readingProgressAnchorIndex ?? 0 + : 0 + ) + ) + return { + uid, + libraryItemID, + readingProgressPercent: percent, + readingProgressTopPercent: top, + 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, + error, + }) + } + return undefined +} 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 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 { 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,