From 1da9b51dec958c5c68ad62b1f72858203c25a63b Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 13:47:15 +0800
Subject: [PATCH 01/12] do not reuse candidates after refreshing

---
 packages/api/src/jobs/ai/create_digest.ts | 64 +++++++++++++++--------
 1 file changed, 42 insertions(+), 22 deletions(-)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index b6470d402..086f9f44f 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -1,26 +1,24 @@
-import { logger } from '../../utils/logger'
-import { v4 as uuid } from 'uuid'
-
-import { OpenAI } from '@langchain/openai'
+import { JsonOutputParser } from '@langchain/core/output_parsers'
 import { PromptTemplate } from '@langchain/core/prompts'
-import { LibraryItem } from '../../entity/library_item'
+import { OpenAI } from '@langchain/openai'
 import {
   htmlToSpeechFile,
   SpeechFile,
   SSMLOptions,
 } from '@omnivore/text-to-speech-handler'
 import axios from 'axios'
+import showdown from 'showdown'
+import yaml from 'yaml'
+import { LibraryItem } from '../../entity/library_item'
+import { TaskState } from '../../generated/graphql'
+import { redisDataSource } from '../../redis_data_source'
+import { Digest, writeDigest } from '../../services/digest'
 import {
   findLibraryItemsByIds,
   searchLibraryItems,
 } from '../../services/library_item'
-import { redisDataSource } from '../../redis_data_source'
+import { logger } from '../../utils/logger'
 import { htmlToMarkdown } from '../../utils/parser'
-import yaml from 'yaml'
-import { JsonOutputParser } from '@langchain/core/output_parsers'
-import showdown from 'showdown'
-import { Digest, writeDigest } from '../../services/digest'
-import { TaskState } from '../../generated/graphql'
 
 export type CreateDigestJobSchedule = 'daily' | 'weekly'
 
@@ -131,7 +129,7 @@ const getPreferencesList = async (userId: string): Promise<LibraryItem[]> => {
 // Makes multiple DB queries and combines the results
 const getCandidatesList = async (
   userId: string,
-  libraryItemIds?: string[]
+  selectedLibraryItemIds?: string[]
 ): Promise<LibraryItem[]> => {
   // use the queries from the digest definitions to lookup preferences
   // There should be a list of multiple queries we use. For now we can
@@ -140,18 +138,25 @@ const getCandidatesList = async (
   //   count: 100
   //   reason: "most recent 100 items saved over 500 words
 
-  if (libraryItemIds) {
-    logger.info('Using libraryItemIds')
-    return findLibraryItemsByIds(libraryItemIds, userId)
+  if (selectedLibraryItemIds) {
+    return findLibraryItemsByIds(selectedLibraryItemIds, userId)
   }
 
+  // get the existing candidate ids from cache
+  const key = `digest:existingCandidateIds:${userId}`
+  const existingCandidateIds = await redisDataSource.redisClient?.get(key)
+
+  logger.info('existingCandidateIds: ', { existingCandidateIds })
+
   const candidates = await Promise.all(
     digestDefinition.candidateSelectors.map(async (selector) => {
       // use the selector to fetch items
       const results = await searchLibraryItems(
         {
           includeContent: true,
-          query: selector.query,
+          query: existingCandidateIds
+            ? `(${selector.query}) -includes:${existingCandidateIds}` // exclude the existing candidates
+            : selector.query,
           size: selector.count,
         },
         userId
@@ -172,6 +177,15 @@ const getCandidatesList = async (
       readableContent: htmlToMarkdown(item.readableContent),
     })) // convert the html content to markdown
 
+  if (dedupedCandidates.length === 0) {
+    logger.info('No new candidates found')
+    return []
+  }
+
+  // store the ids in cache
+  const candidateIds = dedupedCandidates.map((item) => item.id).join(',')
+  await redisDataSource.redisClient?.set(key, candidateIds)
+
   return dedupedCandidates
 }
 
@@ -203,7 +217,7 @@ const createUserProfile = async (
 // it to redis
 const findOrCreateUserProfile = async (userId: string): Promise<string> => {
   // check redis for user profile, return if found
-  const key = `userProfile:${userId}`
+  const key = `digest:userProfile:${userId}`
   const existingProfile = await redisDataSource.redisClient?.get(key)
   if (existingProfile) {
     return existingProfile
@@ -376,11 +390,8 @@ const generateDescription = (
   summaries: RankedItem[],
   rankedTopics: string[]
 ): string =>
-  `We selected ${
-    summaries.length
-  } articles from your last 24 hours of saved items, covering ${rankedTopics.join(
-    ', '
-  )}.`
+  `We selected ${summaries.length} articles from your last 24 hours of saved items` +
+  (rankedTopics.length ? `, covering ${rankedTopics.join(', ')}.` : '.')
 
 // generate content based on the summaries
 const generateContent = (summaries: RankedItem[]): string =>
@@ -401,6 +412,15 @@ export const createDigestJob = async (jobData: CreateDigestJobData) => {
     jobData.userId,
     jobData.libraryItemIds
   )
+  if (candidates.length === 0) {
+    logger.info('No candidates found')
+    return writeDigest(jobData.userId, {
+      id: jobData.id,
+      jobState: TaskState.Succeeded,
+      title: 'No articles found',
+    })
+  }
+
   const userProfile = await findOrCreateUserProfile(jobData.userId)
   const rankedCandidates = await rankCandidates(candidates, userProfile)
   const { finalSelections, rankedTopics } =

From 0d8b17f6aaa48e563b025bb9b0489dd134fd06c2 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 13:57:02 +0800
Subject: [PATCH 02/12] remove urlsToAudio from digest

---
 packages/api/src/jobs/ai/create_digest.ts | 1 -
 packages/api/src/services/digest.ts       | 1 -
 2 files changed, 2 deletions(-)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index 086f9f44f..807aa6a7f 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -440,7 +440,6 @@ export const createDigestJob = async (jobData: CreateDigestJobData) => {
     id: jobData.id,
     title,
     content: generateContent(summaries),
-    urlsToAudio: [],
     jobState: TaskState.Succeeded,
     speechFiles,
     chapters: filteredSummaries.map((item, index) => ({
diff --git a/packages/api/src/services/digest.ts b/packages/api/src/services/digest.ts
index 811202efd..31ee983c9 100644
--- a/packages/api/src/services/digest.ts
+++ b/packages/api/src/services/digest.ts
@@ -23,7 +23,6 @@ export interface Digest {
   content?: string
   chapters?: Chapter[]
 
-  urlsToAudio?: string[]
   speechFiles?: SpeechFile[]
 }
 

From 099310ed9511969761f4ab0e18556d85d735211a Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 14:30:40 +0800
Subject: [PATCH 03/12] reuse existing candidates if there is no one

---
 packages/api/src/jobs/ai/create_digest.ts | 14 ++++++++++++--
 1 file changed, 12 insertions(+), 2 deletions(-)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index 807aa6a7f..9b99f2e97 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -179,6 +179,14 @@ const getCandidatesList = async (
 
   if (dedupedCandidates.length === 0) {
     logger.info('No new candidates found')
+
+    if (existingCandidateIds) {
+      // reuse the existing candidates
+      const existingIds = existingCandidateIds.split(',')
+      return findLibraryItemsByIds(existingIds, userId)
+    }
+
+    // return empty array if no existing candidates
     return []
   }
 
@@ -303,7 +311,6 @@ const chooseRankedSelections = (rankedCandidates: RankedItem[]) => {
   }
 
   logger.info('rankedTopics: ', rankedTopics)
-  logger.info('finalSelections: ', selected)
 
   const finalSelections = []
 
@@ -312,7 +319,10 @@ const chooseRankedSelections = (rankedCandidates: RankedItem[]) => {
     finalSelections.push(...matches)
   }
 
-  logger.info('finalSelections: ', finalSelections)
+  logger.info(
+    'finalSelections: ',
+    finalSelections.map((item) => item.libraryItem.title)
+  )
 
   return { finalSelections, rankedTopics }
 }

From b93573460c08b44fe610f959c21ecc311ee00645 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 14:49:05 +0800
Subject: [PATCH 04/12] filter out summary if the length is longer than the
 original content

---
 packages/api/src/jobs/ai/create_digest.ts | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index 9b99f2e97..dcd9f7176 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -143,7 +143,7 @@ const getCandidatesList = async (
   }
 
   // get the existing candidate ids from cache
-  const key = `digest:existingCandidateIds:${userId}`
+  const key = `digest:${userId}:existingCandidateIds`
   const existingCandidateIds = await redisDataSource.redisClient?.get(key)
 
   logger.info('existingCandidateIds: ', { existingCandidateIds })
@@ -225,7 +225,7 @@ const createUserProfile = async (
 // it to redis
 const findOrCreateUserProfile = async (userId: string): Promise<string> => {
   // check redis for user profile, return if found
-  const key = `digest:userProfile:${userId}`
+  const key = `digest:${userId}:userProfile`
   const existingProfile = await redisDataSource.redisClient?.get(key)
   if (existingProfile) {
     return existingProfile
@@ -288,6 +288,9 @@ const rankCandidates = async (
   return rankedItems
 }
 
+const filterTopics = (rankedTopics: string[]) =>
+  rankedTopics.filter((topic) => topic.length > 0)
+
 // Does some grouping by topic while trying to maintain ranking
 // adds some basic topic diversity
 const chooseRankedSelections = (rankedCandidates: RankedItem[]) => {
@@ -324,7 +327,10 @@ const chooseRankedSelections = (rankedCandidates: RankedItem[]) => {
     finalSelections.map((item) => item.libraryItem.title)
   )
 
-  return { finalSelections, rankedTopics }
+  return {
+    finalSelections,
+    rankedTopics: filterTopics(rankedTopics),
+  }
 }
 
 const summarizeItems = async (
@@ -387,7 +393,9 @@ const generateSpeechFiles = (
 // we should have a QA step here that does some
 // basic checks to make sure the summaries are good.
 const filterSummaries = (summaries: RankedItem[]): RankedItem[] => {
-  return summaries.filter((item) => item.summary.length > 100)
+  return summaries.filter(
+    (item) => item.summary.length < item.libraryItem.readableContent.length
+  )
 }
 
 // we can use something more sophisticated to generate titles

From 7871154160ba13f3c4eeb3dd8f453b546d47234b Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 17:04:23 +0800
Subject: [PATCH 05/12] mark failed task

---
 packages/api/src/jobs/ai/create_digest.ts | 108 +++++++++++++---------
 packages/api/src/routers/digest_router.ts |  12 +--
 packages/api/src/utils/createTask.ts      |  55 ++++++++---
 yarn.lock                                 |  95 +------------------
 4 files changed, 115 insertions(+), 155 deletions(-)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index dcd9f7176..90260cf4c 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -71,9 +71,18 @@ interface RankedTitle {
 }
 
 export const CREATE_DIGEST_JOB = 'create-digest'
+export const CRON_PATTERNS = {
+  // every day at 1PM UTC
+  daily: '0 13 * * *',
+  // every Sunday at 1PM UTC
+  weekly: '0 13 * * 7',
+}
 
 let digestDefinition: DigestDefinition
 
+export const getCronPattern = (schedule: CreateDigestJobSchedule) =>
+  CRON_PATTERNS[schedule]
+
 const fetchDigestDefinition = async (): Promise<DigestDefinition> => {
   const promptFileUrl = process.env.PROMPT_FILE_URL
   if (!promptFileUrl) {
@@ -289,7 +298,7 @@ const rankCandidates = async (
 }
 
 const filterTopics = (rankedTopics: string[]) =>
-  rankedTopics.filter((topic) => topic.length > 0)
+  rankedTopics.filter((topic) => topic?.length > 0)
 
 // Does some grouping by topic while trying to maintain ranking
 // adds some basic topic diversity
@@ -424,53 +433,62 @@ const generateByline = (summaries: RankedItem[]): string =>
     .join(', ')
 
 export const createDigestJob = async (jobData: CreateDigestJobData) => {
-  digestDefinition = await fetchDigestDefinition()
+  try {
+    digestDefinition = await fetchDigestDefinition()
 
-  const candidates = await getCandidatesList(
-    jobData.userId,
-    jobData.libraryItemIds
-  )
-  if (candidates.length === 0) {
-    logger.info('No candidates found')
-    return writeDigest(jobData.userId, {
+    const candidates = await getCandidatesList(
+      jobData.userId,
+      jobData.libraryItemIds
+    )
+    if (candidates.length === 0) {
+      logger.info('No candidates found')
+      return writeDigest(jobData.userId, {
+        id: jobData.id,
+        jobState: TaskState.Succeeded,
+        title: 'No articles found',
+      })
+    }
+
+    const userProfile = await findOrCreateUserProfile(jobData.userId)
+    const rankedCandidates = await rankCandidates(candidates, userProfile)
+    const { finalSelections, rankedTopics } =
+      chooseRankedSelections(rankedCandidates)
+
+    const summaries = await summarizeItems(finalSelections)
+
+    const filteredSummaries = filterSummaries(summaries)
+
+    const speechFiles = generateSpeechFiles(filteredSummaries, {
+      ...jobData,
+      primaryVoice: jobData.voices?.[0],
+      secondaryVoice: jobData.voices?.[1],
+    })
+    const title = generateTitle(summaries)
+    const digest: Digest = {
       id: jobData.id,
+      title,
+      content: generateContent(summaries),
       jobState: TaskState.Succeeded,
-      title: 'No articles found',
+      speechFiles,
+      chapters: filteredSummaries.map((item, index) => ({
+        title: item.libraryItem.title,
+        id: item.libraryItem.id,
+        url: item.libraryItem.originalUrl,
+        thumbnail: item.libraryItem.thumbnail ?? undefined,
+        wordCount: speechFiles[index].wordCount,
+      })),
+      createdAt: new Date(),
+      description: generateDescription(summaries, rankedTopics),
+      byline: generateByline(summaries),
+    }
+
+    await writeDigest(jobData.userId, digest)
+  } catch (error) {
+    logger.error('createDigestJob error', error)
+
+    await writeDigest(jobData.userId, {
+      id: jobData.id,
+      jobState: TaskState.Failed,
     })
   }
-
-  const userProfile = await findOrCreateUserProfile(jobData.userId)
-  const rankedCandidates = await rankCandidates(candidates, userProfile)
-  const { finalSelections, rankedTopics } =
-    chooseRankedSelections(rankedCandidates)
-
-  const summaries = await summarizeItems(finalSelections)
-
-  const filteredSummaries = filterSummaries(summaries)
-
-  const speechFiles = generateSpeechFiles(filteredSummaries, {
-    ...jobData,
-    primaryVoice: jobData.voices?.[0],
-    secondaryVoice: jobData.voices?.[1],
-  })
-  const title = generateTitle(summaries)
-  const digest: Digest = {
-    id: jobData.id,
-    title,
-    content: generateContent(summaries),
-    jobState: TaskState.Succeeded,
-    speechFiles,
-    chapters: filteredSummaries.map((item, index) => ({
-      title: item.libraryItem.title,
-      id: item.libraryItem.id,
-      url: item.libraryItem.originalUrl,
-      thumbnail: item.libraryItem.thumbnail ?? undefined,
-      wordCount: speechFiles[index].wordCount,
-    })),
-    createdAt: new Date(),
-    description: generateDescription(summaries, rankedTopics),
-    byline: generateByline(summaries),
-  }
-
-  await writeDigest(jobData.userId, digest)
 }
diff --git a/packages/api/src/routers/digest_router.ts b/packages/api/src/routers/digest_router.ts
index 2cc8649c7..86f0a4636 100644
--- a/packages/api/src/routers/digest_router.ts
+++ b/packages/api/src/routers/digest_router.ts
@@ -1,5 +1,6 @@
 import cors from 'cors'
 import express from 'express'
+import { v4 as uuid } from 'uuid'
 import { env } from '../env'
 import { TaskState } from '../generated/graphql'
 import { CreateDigestJobSchedule } from '../jobs/ai/create_digest'
@@ -11,7 +12,6 @@ import { getClaimsByToken, getTokenByRequest } from '../utils/auth'
 import { corsConfig } from '../utils/corsConfig'
 import { enqueueCreateDigest } from '../utils/createTask'
 import { logger } from '../utils/logger'
-import { v4 as uuid } from 'uuid'
 
 interface Feedback {
   digestRating: number
@@ -153,15 +153,11 @@ export function digestRouter() {
         return res.sendStatus(404)
       }
 
-      if (digest.jobState === TaskState.Running) {
-        // if job is running then return job state
-        return res.send({
-          jobId: digest.id,
-          jobState: digest.jobState,
-        })
+      if (digest.jobState === TaskState.Failed) {
+        logger.error(`Digest job failed: ${userId}`)
+        return res.sendStatus(500)
       }
 
-      // if job is done then return the digest
       return res.send(digest)
     } catch (error) {
       logger.error('Error while getting digest', error)
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index eeefd3692..e6b02456f 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -22,9 +22,12 @@ import {
   CreateDigestJobResponse,
   CreateDigestJobSchedule,
   CREATE_DIGEST_JOB,
+  CRON_PATTERNS,
+  getCronPattern,
 } from '../jobs/ai/create_digest'
 import { BulkActionData, BULK_ACTION_JOB_NAME } from '../jobs/bulk_action'
 import { CallWebhookJobData, CALL_WEBHOOK_JOB_NAME } from '../jobs/call_webhook'
+import { SendEmailJobData, SEND_EMAIL_JOB } from '../jobs/email/send_email'
 import { THUMBNAIL_JOB } from '../jobs/find_thumbnail'
 import { EXPORT_ALL_ITEMS_JOB_NAME } from '../jobs/integration/export_all_items'
 import {
@@ -42,7 +45,6 @@ import {
   REFRESH_ALL_FEEDS_JOB_NAME,
   REFRESH_FEED_JOB_NAME,
 } from '../jobs/rss/refreshAllFeeds'
-import { SendEmailJobData, SEND_EMAIL_JOB } from '../jobs/email/send_email'
 import { SYNC_READ_POSITIONS_JOB_NAME } from '../jobs/sync_read_positions'
 import { TriggerRuleJobData, TRIGGER_RULE_JOB_NAME } from '../jobs/trigger_rule'
 import {
@@ -53,13 +55,13 @@ import {
 } from '../jobs/update_db'
 import { getBackendQueue, JOB_VERSION } from '../queue-processor'
 import { redisDataSource } from '../redis_data_source'
+import { writeDigest } from '../services/digest'
 import { signFeatureToken } from '../services/features'
 import { OmnivoreAuthorizationHeader } from './auth'
 import { CreateTaskError } from './errors'
 import { stringToHash } from './helpers'
 import { logger } from './logger'
 import View = google.cloud.tasks.v2.Task.View
-import { writeDigest } from '../services/digest'
 
 // Instantiates a client.
 const client = new CloudTasksClient()
@@ -870,22 +872,51 @@ export const enqueueCreateDigest = async (
     throw new Error('No queue found')
   }
 
+  const jobId = `${CREATE_DIGEST_JOB}_${data.userId}`
+
+  let repeatOptions
+
+  if (schedule) {
+    const pattern = getCronPattern(schedule)
+    repeatOptions = {
+      immediately: true, // run immediately
+      pattern,
+    }
+
+    await Promise.all(
+      Object.keys(CRON_PATTERNS).map(async (key) => {
+        // remove existing repeated job if any
+        const isDeleted = await queue.removeRepeatable(
+          CREATE_DIGEST_JOB,
+          {
+            immediately: true, // run immediately
+            pattern: CRON_PATTERNS[key as keyof typeof CRON_PATTERNS],
+          },
+          jobId
+        )
+
+        if (isDeleted) {
+          logger.info('existing repeated job removed', { jobId, schedule: key })
+        }
+      })
+    )
+  }
+
   const job = await queue.add(CREATE_DIGEST_JOB, data, {
-    jobId: data.id, // dedupe by job id
+    jobId, // dedupe by job id
     removeOnComplete: true,
     removeOnFail: true,
-    attempts: 3,
+    attempts: 1,
     priority: getJobPriority(CREATE_DIGEST_JOB),
-    repeat: schedule
-      ? {
-          immediately: true, // run immediately
-          pattern: schedule === 'daily' ? '0 13 * * *' : '0 13 * * 7', // every day or every Sunday at 1PM
-          utc: true,
-        }
-      : undefined,
+    repeat: repeatOptions,
   })
 
-  logger.info('create digest job enqueued', { jobId: job.id })
+  if (!job || !job.id) {
+    logger.error('Error while enqueuing create digest job', data)
+    throw new Error('Error while enqueuing create digest job')
+  }
+
+  logger.info('create digest job enqueued', { jobId, schedule })
 
   const digest = {
     id: data.id,
diff --git a/yarn.lock b/yarn.lock
index 7b354c30f..9c651f55f 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -6493,11 +6493,6 @@
   resolved "https://registry.yarnpkg.com/@sqltools/formatter/-/formatter-1.2.3.tgz#1185726610acc37317ddab11c3c7f9066966bd20"
   integrity sha512-O3uyB/JbkAEMZaP3YqyHH7TMnex7tWyCbCI4EfJdOCoN6HIhqdJBWTM6aCCiWQ/5f5wxjgU735QAIpJbjDvmzg==
 
-"@sqltools/formatter@^1.2.5":
-  version "1.2.5"
-  resolved "https://registry.yarnpkg.com/@sqltools/formatter/-/formatter-1.2.5.tgz#3abc203c79b8c3e90fd6c156a0c62d5403520e12"
-  integrity sha512-Uy0+khmZqUrUGm5dmMqVlnvufZRSK0FbYzVgp0UMstm+F5+W2/jnEEQyc9vo1ZR/E5ZI/B1WjjoTqBqwJL6Krw==
-
 "@stitches/react@^1.2.5":
   version "1.2.8"
   resolved "https://registry.yarnpkg.com/@stitches/react/-/react-1.2.8.tgz#954f8008be8d9c65c4e58efa0937f32388ce3a38"
@@ -8353,13 +8348,6 @@
   dependencies:
     undici-types "~5.26.4"
 
-"@types/node@^20.11.0":
-  version "20.12.7"
-  resolved "https://registry.yarnpkg.com/@types/node/-/node-20.12.7.tgz#04080362fa3dd6c5822061aa3124f5c152cff384"
-  integrity sha512-wq0cICSkRLVaf3UGLMGItu/PtdY7oaXaI/RVU+xliKVOtRna3PRY57ZDfztpDL0n11vfymMUnXv8QwYCO7L1wg==
-  dependencies:
-    undici-types "~5.26.4"
-
 "@types/node@^20.8.4":
   version "20.11.30"
   resolved "https://registry.yarnpkg.com/@types/node/-/node-20.11.30.tgz#9c33467fc23167a347e73834f788f4b9f399d66f"
@@ -9928,11 +9916,6 @@ app-root-path@^3.0.0:
   resolved "https://registry.yarnpkg.com/app-root-path/-/app-root-path-3.0.0.tgz#210b6f43873227e18a4b810a032283311555d5ad"
   integrity sha512-qMcx+Gy2UZynHjOHOIXPNvpf+9cjvk3cWrBBK7zg4gH9+clobJRb9NGzcT7mQTcV/6Gm/1WelUtqxVXnNlrwcw==
 
-app-root-path@^3.1.0:
-  version "3.1.0"
-  resolved "https://registry.yarnpkg.com/app-root-path/-/app-root-path-3.1.0.tgz#5971a2fc12ba170369a7a1ef018c71e6e47c2e86"
-  integrity sha512-biN3PwB2gUtjaYy/isrU3aNWI5w+fAfvHkSvCKeQGxhmYpwKFUxudR3Yya+KqVRHBmEDYh+/lTozYCFbmzX4nA==
-
 apparatus@^0.0.10:
   version "0.0.10"
   resolved "https://registry.yarnpkg.com/apparatus/-/apparatus-0.0.10.tgz#81ea756772ada77863db54ceee8202c109bdca3e"
@@ -13397,7 +13380,7 @@ dateformat@^3.0.0, dateformat@^3.0.3:
   resolved "https://registry.yarnpkg.com/dateformat/-/dateformat-3.0.3.tgz#a6e37499a4d9a9cf85ef5872044d62901c9889ae"
   integrity sha512-jyCETtSl3VMZMWeRo7iY1FL19ges1t55hMo5yaam4Jrsm5EPL89UQkoQRyiI+Yf4k8r2ZpdngkV8hr1lIdjb3Q==
 
-dayjs@1.x, dayjs@^1.10.4, dayjs@^1.11.7, dayjs@^1.11.9:
+dayjs@1.x, dayjs@^1.10.4, dayjs@^1.11.7:
   version "1.11.10"
   resolved "https://registry.yarnpkg.com/dayjs/-/dayjs-1.11.10.tgz#68acea85317a6e164457d6d6947564029a6a16a0"
   integrity sha512-vjAczensTgRcqDERK0SR2XMwsF/tSvnvlv6VcF2GIhg6Sx4yOIt/irsr1RDJsKiIyBzJDpCoXiWWq28MqH2cnQ==
@@ -14133,7 +14116,7 @@ dotenv@^16.0.1:
   resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-16.0.1.tgz#8f8f9d94876c35dac989876a5d3a82a267fdce1d"
   integrity sha512-1K6hR6wtk2FviQ4kEiSjFiH5rpzEVi8WW0x96aztHVMhEspNpc4DVOUTEHtEva5VThQ8IaBX1Pe4gSzpVVUsKQ==
 
-dotenv@^16.0.3, dotenv@^16.3.1:
+dotenv@^16.3.1:
   version "16.4.5"
   resolved "https://registry.yarnpkg.com/dotenv/-/dotenv-16.4.5.tgz#cdd3b3b604cb327e286b4762e13502f717cb099f"
   integrity sha512-ZmdL2rui+eB2YwhsWzjInR8LldtZHGDoQ1ugH85ppHKwpUHL7j7rN0Ti9NCnGiQbhaZ11FpR+7ao1dNsmduNUg==
@@ -16828,17 +16811,6 @@ glob@^10.2.2:
     minipass "^5.0.0 || ^6.0.2 || ^7.0.0"
     path-scurry "^1.10.1"
 
-glob@^10.3.10:
-  version "10.3.12"
-  resolved "https://registry.yarnpkg.com/glob/-/glob-10.3.12.tgz#3a65c363c2e9998d220338e88a5f6ac97302960b"
-  integrity sha512-TCNv8vJ+xz4QiqTpfOJA7HvYv+tNIRHKfUWw/q+v2jdgN4ebz+KY9tGx5J4rHP0o84mNP+ApH66HRX8us3Khqg==
-  dependencies:
-    foreground-child "^3.1.0"
-    jackspeak "^2.3.6"
-    minimatch "^9.0.1"
-    minipass "^7.0.4"
-    path-scurry "^1.10.2"
-
 glob@^8.0.0:
   version "8.0.3"
   resolved "https://registry.yarnpkg.com/glob/-/glob-8.0.3.tgz#415c6eb2deed9e502c68fa44a272e6da6eeca42e"
@@ -19292,7 +19264,7 @@ iterator.prototype@^1.1.2:
     reflect.getprototypeof "^1.0.4"
     set-function-name "^2.0.1"
 
-jackspeak@^2.3.5, jackspeak@^2.3.6:
+jackspeak@^2.3.5:
   version "2.3.6"
   resolved "https://registry.yarnpkg.com/jackspeak/-/jackspeak-2.3.6.tgz#647ecc472238aee4b06ac0e461acc21a8c505ca8"
   integrity sha512-N3yCS/NegsOBokc8GAdM8UcmfsKiSS8cipheD/nivzr700H+nsMOxJjQnvwOcRYVuFkdH0wGUvW2WbXGmrZGbQ==
@@ -21284,11 +21256,6 @@ lowlight@^1.14.0:
     fault "^1.0.0"
     highlight.js "~10.7.0"
 
-lru-cache@^10.2.0:
-  version "10.2.0"
-  resolved "https://registry.yarnpkg.com/lru-cache/-/lru-cache-10.2.0.tgz#0bd445ca57363465900f4d1f9bd8db343a4d95c3"
-  integrity sha512-2bIM8x+VAf6JT4bKAljS1qUWgMsqZRPGJS6FSahIMPVvctcNhyVp7AJu7quxOW9jwkryBReKZY5tY5JYv2n/7Q==
-
 lru-cache@^4.1.5:
   version "4.1.5"
   resolved "https://registry.yarnpkg.com/lru-cache/-/lru-cache-4.1.5.tgz#8bbe50ea85bed59bc9e33dcab8235ee9bcf443cd"
@@ -22514,7 +22481,7 @@ minipass@^5.0.0:
   resolved "https://registry.yarnpkg.com/minipass/-/minipass-5.0.0.tgz#3e9788ffb90b694a5d0ec94479a45b5d8738133d"
   integrity sha512-3FnjYuehv9k6ovOEbyOswadCDPX1piCfhV8ncmYtHOjuPwylVWsghTLo7rabjC3Rx5xD4HDx8Wm1xnMF7S5qFQ==
 
-"minipass@^5.0.0 || ^6.0.2 || ^7.0.0", minipass@^7.0.3, minipass@^7.0.4:
+"minipass@^5.0.0 || ^6.0.2 || ^7.0.0", minipass@^7.0.3:
   version "7.0.4"
   resolved "https://registry.yarnpkg.com/minipass/-/minipass-7.0.4.tgz#dbce03740f50a4786ba994c1fb908844d27b038c"
   integrity sha512-jYofLM5Dam9279rdkWzqHozUo4ybjdZmCsDHePy5V/PbBcVMiSZR97gmAy45aqi8CK1lG2ECd356FU86avfwUQ==
@@ -22602,11 +22569,6 @@ mkdirp@^1.0.3, mkdirp@^1.0.4:
   resolved "https://registry.yarnpkg.com/mkdirp/-/mkdirp-1.0.4.tgz#3eb5ed62622756d79a5f0e2a221dfebad75c2f7e"
   integrity sha512-vVqVZQyf3WLx2Shd0qJ9xuvqgAyKPLAiqITEtqW0oIUjzo3PePDd6fW9iFz30ef7Ysp/oiWqbhszeGWW2T6Gzw==
 
-mkdirp@^2.1.3:
-  version "2.1.6"
-  resolved "https://registry.yarnpkg.com/mkdirp/-/mkdirp-2.1.6.tgz#964fbcb12b2d8c5d6fbc62a963ac95a273e2cc19"
-  integrity sha512-+hEnITedc8LAtIP9u3HJDFIdcLV2vXP33sqLLIzkv1Db1zO/1OxbvYf0Y1OC/S/Qo5dxHXepofhmxL02PsKe+A==
-
 mkdirp@~0.3.5:
   version "0.3.5"
   resolved "https://registry.yarnpkg.com/mkdirp/-/mkdirp-0.3.5.tgz#de3e5f8961c88c787ee1368df849ac4413eca8d7"
@@ -24930,14 +24892,6 @@ path-scurry@^1.10.1, path-scurry@^1.6.1:
     lru-cache "^9.1.1 || ^10.0.0"
     minipass "^5.0.0 || ^6.0.2 || ^7.0.0"
 
-path-scurry@^1.10.2:
-  version "1.10.2"
-  resolved "https://registry.yarnpkg.com/path-scurry/-/path-scurry-1.10.2.tgz#8f6357eb1239d5fa1da8b9f70e9c080675458ba7"
-  integrity sha512-7xTavNy5RQXnsjANvVvMkEjvloOinkAjv/Z6Ildz9v2RinZ4SBKTWFOVRbaF8p0vpHnyjV/UwNDdKuUv6M5qcA==
-  dependencies:
-    lru-cache "^10.2.0"
-    minipass "^5.0.0 || ^6.0.2 || ^7.0.0"
-
 path-to-regexp@0.1.7:
   version "0.1.7"
   resolved "https://registry.yarnpkg.com/path-to-regexp/-/path-to-regexp-0.1.7.tgz#df604178005f522f15eb4490e7247a1bfaa67f8c"
@@ -27007,14 +26961,6 @@ read-pkg@^7.1.0:
     parse-json "^5.2.0"
     type-fest "^2.0.0"
 
-read-yaml-file@^2.1.0:
-  version "2.1.0"
-  resolved "https://registry.yarnpkg.com/read-yaml-file/-/read-yaml-file-2.1.0.tgz#c5866712db9ef5343b4d02c2413bada53c41c4a9"
-  integrity sha512-UkRNRIwnhG+y7hpqnycCL/xbTk7+ia9VuVTC0S+zVbwd65DI9eUpRMfsWIGrCWxTU/mi+JW8cHQCrv+zfCbEPQ==
-  dependencies:
-    js-yaml "^4.0.0"
-    strip-bom "^4.0.0"
-
 read@1, read@^1.0.7, read@~1.0.7:
   version "1.0.7"
   resolved "https://registry.yarnpkg.com/read/-/read-1.0.7.tgz#b3da19bd052431a97671d44a42634adf710b40c4"
@@ -27159,11 +27105,6 @@ reflect-metadata@^0.1.13:
   resolved "https://registry.yarnpkg.com/reflect-metadata/-/reflect-metadata-0.1.13.tgz#67ae3ca57c972a2aa1642b10fe363fe32d49dc08"
   integrity sha512-Ts1Y/anZELhSsjMcU605fU9RE4Oi3p5ORujwbIKXfWa+0Zxs510Qrmrce5/Jowq3cHSZSJqBjypxmHarc+vEWg==
 
-reflect-metadata@^0.2.1:
-  version "0.2.2"
-  resolved "https://registry.yarnpkg.com/reflect-metadata/-/reflect-metadata-0.2.2.tgz#400c845b6cba87a21f2c65c4aeb158f4fa4d9c5b"
-  integrity sha512-urBwgfrvVP/eAyXx4hluJivBKzuEbSQs9rKWCrCkbSxNv8mxPcUZKeuoF3Uy4mJl3Lwprp6yy5/39VWigZ4K6Q==
-
 reflect.getprototypeof@^1.0.4:
   version "1.0.4"
   resolved "https://registry.yarnpkg.com/reflect.getprototypeof/-/reflect.getprototypeof-1.0.4.tgz#aaccbf41aca3821b87bb71d9dcbc7ad0ba50a3f3"
@@ -30230,7 +30171,7 @@ tslib@^1.0.0, tslib@^1.11.1, tslib@^1.8.1, tslib@^1.9.0, tslib@^1.9.3:
   resolved "https://registry.yarnpkg.com/tslib/-/tslib-1.14.1.tgz#cf2d38bdc34a134bcaf1091c41f6619e2f672d00"
   integrity sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg==
 
-tslib@^2, tslib@^2.0.0, tslib@^2.0.1, tslib@^2.0.3, tslib@^2.1.0, tslib@^2.3.0, tslib@^2.3.1, tslib@^2.4.0, tslib@^2.5.0, tslib@^2.6.2:
+tslib@^2, tslib@^2.0.0, tslib@^2.0.1, tslib@^2.0.3, tslib@^2.1.0, tslib@^2.3.0, tslib@^2.3.1, tslib@^2.4.0, tslib@^2.6.2:
   version "2.6.2"
   resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.6.2.tgz#703ac29425e7b37cd6fd456e92404d46d1f3e4ae"
   integrity sha512-AEYxH93jGFPn/a2iVAwW87VuUIkR1FVUKB77NwMF7nBTDkDrrT/Hpt/IrCJ0QXhW27jTBDcf5ZY7w6RiqTMw2Q==
@@ -30446,27 +30387,6 @@ typeorm-naming-strategies@^4.1.0:
   resolved "https://registry.yarnpkg.com/typeorm-naming-strategies/-/typeorm-naming-strategies-4.1.0.tgz#1ec6eb296c8d7b69bb06764d5b9083ff80e814a9"
   integrity sha512-vPekJXzZOTZrdDvTl1YoM+w+sUIfQHG4kZTpbFYoTsufyv9NIBRe4Q+PdzhEAFA2std3D9LZHEb1EjE9zhRpiQ==
 
-typeorm@^0.3.19:
-  version "0.3.20"
-  resolved "https://registry.yarnpkg.com/typeorm/-/typeorm-0.3.20.tgz#4b61d737c6fed4e9f63006f88d58a5e54816b7ab"
-  integrity sha512-sJ0T08dV5eoZroaq9uPKBoNcGslHBR4E4y+EBHs//SiGbblGe7IeduP/IH4ddCcj0qp3PHwDwGnuvqEAnKlq/Q==
-  dependencies:
-    "@sqltools/formatter" "^1.2.5"
-    app-root-path "^3.1.0"
-    buffer "^6.0.3"
-    chalk "^4.1.2"
-    cli-highlight "^2.1.11"
-    dayjs "^1.11.9"
-    debug "^4.3.4"
-    dotenv "^16.0.3"
-    glob "^10.3.10"
-    mkdirp "^2.1.3"
-    reflect-metadata "^0.2.1"
-    sha.js "^2.4.11"
-    tslib "^2.5.0"
-    uuid "^9.0.0"
-    yargs "^17.6.2"
-
 typeorm@^0.3.4:
   version "0.3.7"
   resolved "https://registry.yarnpkg.com/typeorm/-/typeorm-0.3.7.tgz#5776ed5058f0acb75d64723b39ff458d21de64c1"
@@ -30505,11 +30425,6 @@ typescript@^4.4.4:
   resolved "https://registry.yarnpkg.com/typescript/-/typescript-4.9.5.tgz#095979f9bcc0d09da324d58d03ce8f8374cbe65a"
   integrity sha512-1FXk9E2Hm+QzZQ7z+McJiHL4NW1F2EzMu9Nq9i3zAaGqibafqYwCVU6WyWAuyQRRzOlxou8xZSyXLEN8oKj24g==
 
-typescript@^5.3.3:
-  version "5.4.5"
-  resolved "https://registry.yarnpkg.com/typescript/-/typescript-5.4.5.tgz#42ccef2c571fdbd0f6718b1d1f5e6e5ef006f611"
-  integrity sha512-vcI4UpRgg81oIRUFwR0WSIHKt11nJ7SAVlYNIu+QpqeyXP+gpQJy/Z4+F0aGxSE4MqwjyXvW/TzgkLAx2AGHwQ==
-
 ua-parser-js@^0.7.30:
   version "0.7.33"
   resolved "https://registry.yarnpkg.com/ua-parser-js/-/ua-parser-js-0.7.33.tgz#1d04acb4ccef9293df6f70f2c3d22f3030d8b532"

From efa930a31a7c3854da50ab314f412ff69380fcaa Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 17:46:06 +0800
Subject: [PATCH 06/12] send push notifications when digest is ready

---
 packages/api/src/jobs/ai/create_digest.ts | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index 90260cf4c..1bc85be4c 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -17,8 +17,10 @@ import {
   findLibraryItemsByIds,
   searchLibraryItems,
 } from '../../services/library_item'
+import { findDeviceTokensByUserId } from '../../services/user_device_tokens'
 import { logger } from '../../utils/logger'
 import { htmlToMarkdown } from '../../utils/parser'
+import { sendMulticastPushNotifications } from '../../utils/sendNotification'
 
 export type CreateDigestJobSchedule = 'daily' | 'weekly'
 
@@ -490,5 +492,19 @@ export const createDigestJob = async (jobData: CreateDigestJobData) => {
       id: jobData.id,
       jobState: TaskState.Failed,
     })
+  } finally {
+    // send notification
+    const tokens = await findDeviceTokensByUserId(jobData.userId)
+    if (tokens.length > 0) {
+      const message = {
+        notification: {
+          title: 'Digest ready',
+          body: 'Your digest is ready to listen',
+        },
+        tokens: tokens.map((token) => token.token),
+      }
+
+      await sendMulticastPushNotifications(jobData.userId, message, 'reminder')
+    }
   }
 }

From a69d1240c8f689d7863178d731b73b1d70b60152 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 19:02:13 +0800
Subject: [PATCH 07/12] add urlsToAudio back

---
 packages/api/src/jobs/ai/create_digest.ts | 1 +
 packages/api/src/services/digest.ts       | 1 +
 2 files changed, 2 insertions(+)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index 1bc85be4c..21a34baaf 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -482,6 +482,7 @@ export const createDigestJob = async (jobData: CreateDigestJobData) => {
       createdAt: new Date(),
       description: generateDescription(summaries, rankedTopics),
       byline: generateByline(summaries),
+      urlsToAudio: [],
     }
 
     await writeDigest(jobData.userId, digest)
diff --git a/packages/api/src/services/digest.ts b/packages/api/src/services/digest.ts
index 31ee983c9..811202efd 100644
--- a/packages/api/src/services/digest.ts
+++ b/packages/api/src/services/digest.ts
@@ -23,6 +23,7 @@ export interface Digest {
   content?: string
   chapters?: Chapter[]
 
+  urlsToAudio?: string[]
   speechFiles?: SpeechFile[]
 }
 

From 07475e69a4d028f83ccf2f91ab9c54bc20f8fed5 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 19:05:37 +0800
Subject: [PATCH 08/12] change cron to schedule at 10:30

---
 packages/api/src/jobs/ai/create_digest.ts | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index 21a34baaf..81558f179 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -74,10 +74,10 @@ interface RankedTitle {
 
 export const CREATE_DIGEST_JOB = 'create-digest'
 export const CRON_PATTERNS = {
-  // every day at 1PM UTC
-  daily: '0 13 * * *',
-  // every Sunday at 1PM UTC
-  weekly: '0 13 * * 7',
+  // every day at 10:30 UTC
+  daily: '0 30 10 * * *',
+  // every Sunday at 10:30 UTC
+  weekly: '0 30 10 * * 7',
 }
 
 let digestDefinition: DigestDefinition

From 273ff9d4be162f25855a3846bed4afa4c5b4dc89 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 21:12:26 +0800
Subject: [PATCH 09/12] increase max ai-digest beta users to 10

---
 packages/api/src/services/features.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/api/src/services/features.ts b/packages/api/src/services/features.ts
index d22e65c74..6400c4779 100644
--- a/packages/api/src/services/features.ts
+++ b/packages/api/src/services/features.ts
@@ -9,7 +9,7 @@ import { logger } from '../utils/logger'
 const MAX_ULTRA_REALISTIC_USERS = 1500
 const MAX_YOUTUBE_TRANSCRIPT_USERS = 500
 const MAX_NOTION_USERS = 1000
-const MAX_AIDIGEST_USERS = 5
+const MAX_AIDIGEST_USERS = 10
 
 export enum FeatureName {
   AISummaries = 'ai-summaries',

From ac775ae8f255de1d22cec35c2a1bf780dd4d9d55 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 21:30:00 +0800
Subject: [PATCH 10/12] Drain httpServer when apollo server stops

---
 packages/api/src/apollo.ts | 37 ++++++++++++++++++++++++++-----------
 packages/api/src/server.ts | 20 +++++---------------
 2 files changed, 31 insertions(+), 26 deletions(-)

diff --git a/packages/api/src/apollo.ts b/packages/api/src/apollo.ts
index 26d255e5d..68e1c3645 100644
--- a/packages/api/src/apollo.ts
+++ b/packages/api/src/apollo.ts
@@ -4,16 +4,24 @@
 /* eslint-disable @typescript-eslint/no-unsafe-member-access */
 /* eslint-disable @typescript-eslint/no-unsafe-assignment */
 /* eslint-disable @typescript-eslint/require-await */
+import { createPrometheusExporterPlugin } from '@bmatei/apollo-prometheus-exporter'
 import { makeExecutableSchema } from '@graphql-tools/schema'
 import * as Sentry from '@sentry/node'
-import { ContextFunction, PluginDefinition } from 'apollo-server-core'
-import { Express } from 'express'
+import {
+  ApolloServerPluginDrainHttpServer,
+  ContextFunction,
+  PluginDefinition,
+} from 'apollo-server-core'
 import { ApolloServer } from 'apollo-server-express'
 import { ExpressContext } from 'apollo-server-express/dist/ApolloServer'
+import { ApolloServerPlugin } from 'apollo-server-plugin-base'
+import { Express } from 'express'
 import * as httpContext from 'express-http-context2'
+import type http from 'http'
 import * as jwt from 'jsonwebtoken'
 import { EntityManager } from 'typeorm'
 import { promisify } from 'util'
+import { ReadingProgressDataSource } from './datasources/reading_progress_data_source'
 import { appDataSource } from './data_source'
 import { sanitizeDirectiveTransformer } from './directives'
 import { env } from './env'
@@ -22,17 +30,14 @@ import { functionResolvers } from './resolvers/function_resolvers'
 import { ClaimsToSet, RequestContext, ResolverContext } from './resolvers/types'
 import ScalarResolvers from './scalars'
 import typeDefs from './schema'
-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'
-import { createPrometheusExporterPlugin } from '@bmatei/apollo-prometheus-exporter'
-import { ApolloServerPlugin } from 'apollo-server-plugin-base'
 import {
   countDailyServiceUsage,
   createServiceUsage,
 } from './services/service_usage'
+import { tracer } from './tracing'
+import { getClaimsByToken, setAuthInCookie } from './utils/auth'
+import { SetClaimsRole } from './utils/dictionary'
+import { logger } from './utils/logger'
 
 const signToken = promisify(jwt.sign)
 const pubsub = createPubSubClient()
@@ -100,7 +105,10 @@ const contextFunc: ContextFunction<ExpressContext, ResolverContext> = async ({
   return ctx
 }
 
-export function makeApolloServer(app: Express): ApolloServer {
+export function makeApolloServer(
+  app: Express,
+  httpServer: http.Server
+): ApolloServer {
   let schema = makeExecutableSchema({
     resolvers,
     typeDefs,
@@ -169,7 +177,14 @@ export function makeApolloServer(app: Express): ApolloServer {
   const apollo = new ApolloServer({
     schema: schema,
     context: contextFunc,
-    plugins: [promExporter, usageLimitPlugin],
+    plugins: [
+      // Our httpServer handles incoming requests to our Express app.
+      // Below, we tell Apollo Server to "drain" this httpServer,
+      // enabling our servers to shut down gracefully.
+      ApolloServerPluginDrainHttpServer({ httpServer }),
+      promExporter,
+      usageLimitPlugin,
+    ],
     formatError: (err) => {
       logger.info('server error', err)
       Sentry.captureException(err)
diff --git a/packages/api/src/server.ts b/packages/api/src/server.ts
index e1603a6cb..58cdafbda 100755
--- a/packages/api/src/server.ts
+++ b/packages/api/src/server.ts
@@ -4,13 +4,12 @@
 /* eslint-disable @typescript-eslint/no-misused-promises */
 import * as lw from '@google-cloud/logging-winston'
 import * as Sentry from '@sentry/node'
-import { ApolloServer } from 'apollo-server-express'
 import { json, urlencoded } from 'body-parser'
 import cookieParser from 'cookie-parser'
 import express, { Express } from 'express'
 import * as httpContext from 'express-http-context2'
 import promBundle from 'express-prom-bundle'
-import { createServer, Server } from 'http'
+import { createServer } from 'http'
 import * as prom from 'prom-client'
 import { config, loggers } from 'winston'
 import { makeApolloServer } from './apollo'
@@ -150,7 +149,8 @@ const main = async (): Promise<void> => {
   }
 
   const app = createApp()
-  const apollo = makeApolloServer(app)
+  const httpServer = createServer(app)
+  const apollo = makeApolloServer(app, httpServer)
   await apollo.start()
   apollo.applyMiddleware({ app, path: '/api/graphql', cors: corsConfig })
 
@@ -159,7 +159,7 @@ const main = async (): Promise<void> => {
   const mw = await lw.express.makeMiddleware(mwLogger, transport)
   app.use(mw)
 
-  const listener = app.listen({ port: PORT }, async () => {
+  const listener = httpServer.listen({ port: PORT }, async () => {
     const logger = buildLogger('app.dispatch')
     logger.notice(`🚀 Server ready at ${apollo.graphqlPath}`)
   })
@@ -176,22 +176,12 @@ const main = async (): Promise<void> => {
   const gracefulShutdown = async (signal: string) => {
     console.log(`[api]: Received ${signal}, closing server...`)
     await apollo.stop()
-    console.log('[api]: Apollo server stopped')
+    console.log('[api]: Express server stopped')
 
     console.log('[posthog]: flushing events')
     await analytics.shutdownAsync()
     console.log('[posthog]: events flushed')
 
-    await new Promise<void>((resolve) => {
-      listener.close((err) => {
-        console.log('[api]: Express listener closed')
-        if (err) {
-          console.log('[api]: error stopping listener', { err })
-        }
-        resolve()
-      })
-    })
-
     // Shutdown redis before DB because the quit sequence can
     // cause appDataSource to get reloaded in the callback
     await redisDataSource.shutdown()

From fa4b3bf46283c7e976a1527e2113bbbff3f41951 Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Thu, 18 Apr 2024 22:09:04 +0800
Subject: [PATCH 11/12] fix cron schedule bug with bull

---
 packages/api/src/jobs/ai/create_digest.ts |  4 +-
 packages/api/src/utils/createTask.ts      | 71 +++++++++++++----------
 packages/api/test/util.ts                 |  4 +-
 3 files changed, 45 insertions(+), 34 deletions(-)

diff --git a/packages/api/src/jobs/ai/create_digest.ts b/packages/api/src/jobs/ai/create_digest.ts
index 81558f179..a3be248a0 100644
--- a/packages/api/src/jobs/ai/create_digest.ts
+++ b/packages/api/src/jobs/ai/create_digest.ts
@@ -75,9 +75,9 @@ interface RankedTitle {
 export const CREATE_DIGEST_JOB = 'create-digest'
 export const CRON_PATTERNS = {
   // every day at 10:30 UTC
-  daily: '0 30 10 * * *',
+  daily: '30 10 * * *',
   // every Sunday at 10:30 UTC
-  weekly: '0 30 10 * * 7',
+  weekly: '30 10 * * 7',
 }
 
 let digestDefinition: DigestDefinition
diff --git a/packages/api/src/utils/createTask.ts b/packages/api/src/utils/createTask.ts
index e6b02456f..b59914dc4 100644
--- a/packages/api/src/utils/createTask.ts
+++ b/packages/api/src/utils/createTask.ts
@@ -872,43 +872,14 @@ export const enqueueCreateDigest = async (
     throw new Error('No queue found')
   }
 
+  // enqueue create digest job immediately
   const jobId = `${CREATE_DIGEST_JOB}_${data.userId}`
-
-  let repeatOptions
-
-  if (schedule) {
-    const pattern = getCronPattern(schedule)
-    repeatOptions = {
-      immediately: true, // run immediately
-      pattern,
-    }
-
-    await Promise.all(
-      Object.keys(CRON_PATTERNS).map(async (key) => {
-        // remove existing repeated job if any
-        const isDeleted = await queue.removeRepeatable(
-          CREATE_DIGEST_JOB,
-          {
-            immediately: true, // run immediately
-            pattern: CRON_PATTERNS[key as keyof typeof CRON_PATTERNS],
-          },
-          jobId
-        )
-
-        if (isDeleted) {
-          logger.info('existing repeated job removed', { jobId, schedule: key })
-        }
-      })
-    )
-  }
-
   const job = await queue.add(CREATE_DIGEST_JOB, data, {
     jobId, // dedupe by job id
     removeOnComplete: true,
     removeOnFail: true,
     attempts: 1,
     priority: getJobPriority(CREATE_DIGEST_JOB),
-    repeat: repeatOptions,
   })
 
   if (!job || !job.id) {
@@ -916,7 +887,7 @@ export const enqueueCreateDigest = async (
     throw new Error('Error while enqueuing create digest job')
   }
 
-  logger.info('create digest job enqueued', { jobId, schedule })
+  logger.info('create digest job enqueued', { jobId })
 
   const digest = {
     id: data.id,
@@ -926,6 +897,44 @@ export const enqueueCreateDigest = async (
   // update digest job state in redis
   await writeDigest(data.userId, digest)
 
+  if (schedule) {
+    await Promise.all(
+      Object.keys(CRON_PATTERNS).map(async (key) => {
+        // remove existing repeated job if any
+        const isDeleted = await queue.removeRepeatable(
+          CREATE_DIGEST_JOB,
+          {
+            pattern: CRON_PATTERNS[key as keyof typeof CRON_PATTERNS],
+            tz: 'UTC',
+          },
+          jobId
+        )
+
+        if (isDeleted) {
+          logger.info('existing repeated job removed', { jobId, schedule: key })
+        }
+      })
+    )
+
+    // schedule repeated job
+    const job = await queue.add(CREATE_DIGEST_JOB, data, {
+      attempts: 1,
+      priority: getJobPriority(CREATE_DIGEST_JOB),
+      repeat: {
+        pattern: getCronPattern(schedule),
+        jobId,
+        tz: 'UTC',
+      },
+    })
+
+    if (!job || !job.id) {
+      logger.error('Error while scheduling create digest job', data)
+      throw new Error('Error while scheduling create digest job')
+    }
+
+    logger.info('create digest job scheduled', { jobId, schedule })
+  }
+
   return {
     jobId: digest.id,
     jobState: digest.jobState,
diff --git a/packages/api/test/util.ts b/packages/api/test/util.ts
index ee1c30bdc..42b5ecd60 100644
--- a/packages/api/test/util.ts
+++ b/packages/api/test/util.ts
@@ -1,4 +1,5 @@
 import { ConnectionOptions, Job, QueueEvents, Worker } from 'bullmq'
+import { createServer } from 'http'
 import { nanoid } from 'nanoid'
 import supertest from 'supertest'
 import { v4 } from 'uuid'
@@ -8,7 +9,8 @@ import { createApp } from '../src/server'
 import { corsConfig } from '../src/utils/corsConfig'
 
 const app = createApp()
-const apollo = makeApolloServer(app)
+const httpServer = createServer(app)
+const apollo = makeApolloServer(app, httpServer)
 export const request = supertest(app)
 let worker: Worker
 let queueEvents: QueueEvents

From 850150956496a6b066b3f3e99153eb952731f1dc Mon Sep 17 00:00:00 2001
From: Hongbo Wu <hongbo@omnivore.app>
Date: Fri, 19 Apr 2024 14:26:17 +0800
Subject: [PATCH 12/12] include content in updateSince api response by default
 for offline use

---
 packages/api/src/resolvers/article/index.ts | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/api/src/resolvers/article/index.ts b/packages/api/src/resolvers/article/index.ts
index 96023f6c0..f2fbe9c3a 100644
--- a/packages/api/src/resolvers/article/index.ts
+++ b/packages/api/src/resolvers/article/index.ts
@@ -786,6 +786,7 @@ export const updatesSinceResolver = authorized<
       size: size + 1, // fetch one more item to get next cursor
       includeDeleted: true,
       query,
+      includeContent: true, // by default include content for offline use for now
     },
     uid
   )