Merge pull request #3460 from omnivore-app/feat/reading-progress-write-behind

Write behind cache for reading progress

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<ExpressContext, ResolverContext> = async ({
         return cb(tx)
       }),
     tracingSpan: tracer.startSpan('apollo.request'),
+    dataSources: {
+      readingProgress: new ReadingProgressDataSource(),
+    },
   }
 
   return ctx

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<ReadingProgressCacheItem | undefined> {
+    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<ReadingProgressCacheItem | undefined> {
+    const cacheItem: ReadingProgressCacheItem = {
+      uid,
+      libraryItemID,
+      updatedAt: new Date().toISOString(),
+      ...progress,
+    }
+    await pushCachedReadingPosition(uid, libraryItemID, cacheItem)
+    return fetchCachedReadingPosition(uid, libraryItemID)
+  }
+}

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)
+  }
+}

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,
   })

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] }
       }

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<LinkShareInfo | undefined> {
-    //   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) {

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<TResult>
   tracingSpan: Span
+  dataSources: {
+    readingProgress: ReadingProgressDataSource
+  }
 }
 
 export type ResolverContext = ApolloContext<RequestContext>

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<boolean> => {
+  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<boolean> => {
+  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<ReadingProgressCacheItem | undefined> => {
+  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
+}

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

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 {

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,

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,